producerConfigOpt;
+
+ public ConsoleProducerConfig(String[] args) {
+ super(args);
+ topicOpt = parser.accepts("topic", "REQUIRED: The topic id to produce messages to.")
+ .withRequiredArg()
+ .describedAs("topic")
+ .ofType(String.class);
+ brokerListOpt = parser.accepts("broker-list", "DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-server is specified. The broker list string in the form HOST1:PORT1,HOST2:PORT2.")
+ .withRequiredArg()
+ .describedAs("broker-list")
+ .ofType(String.class);
+ bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED unless --broker-list(deprecated) is specified. The server(s) to connect to. The broker list string in the form HOST1:PORT1,HOST2:PORT2.")
+ .requiredUnless("broker-list")
+ .withRequiredArg()
+ .describedAs("server to connect to")
+ .ofType(String.class);
+ syncOpt = parser.accepts("sync", "If set message send requests to the brokers are synchronously, one at a time as they arrive.");
+ compressionCodecOpt = parser.accepts("compression-codec", "The compression codec: either 'none', 'gzip', 'snappy', 'lz4', or 'zstd'." +
+ "If specified without value, then it defaults to 'gzip'")
+ .withOptionalArg()
+ .describedAs("compression-codec")
+ .ofType(String.class);
+ batchSizeOpt = parser.accepts("batch-size", "Number of messages to send in a single batch if they are not being sent synchronously. " +
+ "please note that this option will be replaced if max-partition-memory-bytes is also set")
+ .withRequiredArg()
+ .describedAs("size")
+ .ofType(Integer.class)
+ .defaultsTo(16 * 1024);
+ messageSendMaxRetriesOpt = parser.accepts("message-send-max-retries", "Brokers can fail receiving the message for multiple reasons, " +
+ "and being unavailable transiently is just one of them. This property specifies the number of retries before the producer give up and drop this message. " +
+ "This is the option to control `retries` in producer configs.")
+ .withRequiredArg()
+ .ofType(Integer.class)
+ .defaultsTo(3);
+ retryBackoffMsOpt = parser.accepts("retry-backoff-ms", "Before each retry, the producer refreshes the metadata of relevant topics. " +
+ "Since leader election takes a bit of time, this property specifies the amount of time that the producer waits before refreshing the metadata. " +
+ "This is the option to control `retry.backoff.ms` in producer configs.")
+ .withRequiredArg()
+ .ofType(Long.class)
+ .defaultsTo(100L);
+ sendTimeoutOpt = parser.accepts("timeout", "If set and the producer is running in asynchronous mode, this gives the maximum amount of time" +
+ " a message will queue awaiting sufficient batch size. The value is given in ms. " +
+ "This is the option to control `linger.ms` in producer configs.")
+ .withRequiredArg()
+ .describedAs("timeout_ms")
+ .ofType(Long.class)
+ .defaultsTo(1000L);
+ requestRequiredAcksOpt = parser.accepts("request-required-acks", "The required `acks` of the producer requests")
+ .withRequiredArg()
+ .describedAs("request required acks")
+ .ofType(String.class)
+ .defaultsTo("-1");
+ requestTimeoutMsOpt = parser.accepts("request-timeout-ms", "The ack timeout of the producer requests. Value must be non-negative and non-zero.")
+ .withRequiredArg()
+ .describedAs("request timeout ms")
+ .ofType(Integer.class)
+ .defaultsTo(1500);
+ metadataExpiryMsOpt = parser.accepts("metadata-expiry-ms",
+ "The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any leadership changes. " +
+ "This is the option to control `metadata.max.age.ms` in producer configs.")
+ .withRequiredArg()
+ .describedAs("metadata expiration interval")
+ .ofType(Long.class)
+ .defaultsTo(5 * 60 * 1000L);
+ maxBlockMsOpt = parser.accepts("max-block-ms",
+ "The max time that the producer will block for during a send request.")
+ .withRequiredArg()
+ .describedAs("max block on send")
+ .ofType(Long.class)
+ .defaultsTo(60 * 1000L);
+ maxMemoryBytesOpt = parser.accepts("max-memory-bytes",
+ "The total memory used by the producer to buffer records waiting to be sent to the server. " +
+ "This is the option to control `buffer.memory` in producer configs.")
+ .withRequiredArg()
+ .describedAs("total memory in bytes")
+ .ofType(Long.class)
+ .defaultsTo(32 * 1024 * 1024L);
+ maxPartitionMemoryBytesOpt = parser.accepts("max-partition-memory-bytes",
+ "The buffer size allocated for a partition. When records are received which are smaller than this size the producer " +
+ "will attempt to optimistically group them together until this size is reached. " +
+ "This is the option to control `batch.size` in producer configs.")
+ .withRequiredArg()
+ .describedAs("memory in bytes per partition")
+ .ofType(Integer.class)
+ .defaultsTo(16 * 1024);
+ messageReaderOpt = parser.accepts("line-reader", "The class name of the class to use for reading lines from standard in. " +
+ "By default each line is read as a separate message.")
+ .withRequiredArg()
+ .describedAs("reader_class")
+ .ofType(String.class)
+ .defaultsTo(LineMessageReader.class.getName());
+ socketBufferSizeOpt = parser.accepts("socket-buffer-size", "The size of the tcp RECV size. " +
+ "This is the option to control `send.buffer.bytes` in producer configs.")
+ .withRequiredArg()
+ .describedAs("size")
+ .ofType(Integer.class)
+ .defaultsTo(1024 * 100);
+ propertyOpt = parser.accepts("property",
+ "A mechanism to pass user-defined properties in the form key=value to the message reader. This allows custom configuration for a user-defined message reader." +
+ "Default properties include:" +
+ "\n parse.key=false" +
+ "\n parse.headers=false" +
+ "\n ignore.error=false" +
+ "\n key.separator=\\t" +
+ "\n headers.delimiter=\\t" +
+ "\n headers.separator=," +
+ "\n headers.key.separator=:" +
+ "\n null.marker= When set, any fields (key, value and headers) equal to this will be replaced by null" +
+ "\nDefault parsing pattern when:" +
+ "\n parse.headers=true and parse.key=true:" +
+ "\n \"h1:v1,h2:v2...\\tkey\\tvalue\"" +
+ "\n parse.key=true:" +
+ "\n \"key\\tvalue\"" +
+ "\n parse.headers=true:" +
+ "\n \"h1:v1,h2:v2...\\tvalue\"")
+ .withRequiredArg()
+ .describedAs("prop")
+ .ofType(String.class);
+ readerConfigOpt = parser.accepts("reader-config", "Config properties file for the message reader. Note that " + propertyOpt + " takes precedence over this config.")
+ .withRequiredArg()
+ .describedAs("config file")
+ .ofType(String.class);
+ producerPropertyOpt = parser.accepts("producer-property", "A mechanism to pass user-defined properties in the form key=value to the producer.")
+ .withRequiredArg()
+ .describedAs("producer_prop")
+ .ofType(String.class);
+ producerConfigOpt = parser.accepts("producer.config", "Producer config properties file. Note that " + producerPropertyOpt + " takes precedence over this config.")
+ .withRequiredArg()
+ .describedAs("config file")
+ .ofType(String.class);
+
+ try {
+ options = parser.parse(args);
+
+ } catch (OptionException e) {
+ CommandLineUtils.printUsageAndExit(parser, e.getMessage());
+ }
+
+ CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to read data from standard input and publish it to Kafka.");
+ CommandLineUtils.checkRequiredArgs(parser, options, topicOpt);
+
+ ToolsUtils.validatePortOrExit(parser, brokerHostsAndPorts());
+ }
+
+ String brokerHostsAndPorts() {
+ return options.has(bootstrapServerOpt) ? options.valueOf(bootstrapServerOpt) : options.valueOf(brokerListOpt);
+ }
+
+ boolean sync() {
+ return options.has(syncOpt);
+ }
+
+ String compressionCodec() {
+ if (options.has(compressionCodecOpt)) {
+ String codecOptValue = options.valueOf(compressionCodecOpt);
+ // Defaults to gzip if no value is provided.
+ return codecOptValue == null || codecOptValue.isEmpty() ? CompressionType.GZIP.name : codecOptValue;
+ }
+
+ return CompressionType.NONE.name;
+ }
+
+ String readerClass() {
+ return options.valueOf(messageReaderOpt);
+ }
+
+ Properties getReaderProps() throws IOException {
+ Properties properties = new Properties();
+
+ if (options.has(readerConfigOpt)) {
+ properties.putAll(loadProps(options.valueOf(readerConfigOpt)));
+ }
+
+ properties.put("topic", options.valueOf(topicOpt));
+ properties.putAll(parseKeyValueArgs(options.valuesOf(propertyOpt)));
+ return properties;
+ }
+
+ Properties getProducerProps() throws IOException {
+ Properties properties = new Properties();
+
+ if (options.has(producerConfigOpt)) {
+ properties.putAll(loadProps(options.valueOf(producerConfigOpt)));
+ }
+
+ properties.putAll(parseKeyValueArgs(options.valuesOf(producerPropertyOpt)));
+ properties.put(BOOTSTRAP_SERVERS_CONFIG, brokerHostsAndPorts());
+ properties.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
+ properties.putIfAbsent(CLIENT_ID_CONFIG, "console-producer");
+ properties.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+ properties.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+
+ maybeMergeOptions(properties, LINGER_MS_CONFIG, options, sendTimeoutOpt);
+ maybeMergeOptions(properties, ACKS_CONFIG, options, requestRequiredAcksOpt);
+ maybeMergeOptions(properties, REQUEST_TIMEOUT_MS_CONFIG, options, requestTimeoutMsOpt);
+ maybeMergeOptions(properties, RETRIES_CONFIG, options, messageSendMaxRetriesOpt);
+ maybeMergeOptions(properties, RETRY_BACKOFF_MS_CONFIG, options, retryBackoffMsOpt);
+ maybeMergeOptions(properties, SEND_BUFFER_CONFIG, options, socketBufferSizeOpt);
+ maybeMergeOptions(properties, BUFFER_MEMORY_CONFIG, options, maxMemoryBytesOpt);
+ // We currently have 2 options to set the batch.size value. We'll deprecate/remove one of them in KIP-717.
+ maybeMergeOptions(properties, BATCH_SIZE_CONFIG, options, batchSizeOpt);
+ maybeMergeOptions(properties, BATCH_SIZE_CONFIG, options, maxPartitionMemoryBytesOpt);
+ maybeMergeOptions(properties, METADATA_MAX_AGE_CONFIG, options, metadataExpiryMsOpt);
+ maybeMergeOptions(properties, MAX_BLOCK_MS_CONFIG, options, maxBlockMsOpt);
+
+ return properties;
+ }
+ }
+}
diff --git a/tools/src/main/java/org/apache/kafka/tools/LineMessageReader.java b/tools/src/main/java/org/apache/kafka/tools/LineMessageReader.java
new file mode 100644
index 0000000000000..8fa3c0ffd46c9
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/LineMessageReader.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+import static java.util.Arrays.stream;
+
+/**
+ * The default implementation of {@link MessageReader} for the {@link ConsoleProducer}. This reader comes with
+ * the ability to parse a record's headers, key and value based on configurable separators. The reader configuration
+ * is defined as follows:
+ *
+ *
+ * parse.key : indicates if a record's key is included in a line input and needs to be parsed. (default: false).
+ * key.separator : the string separating a record's key from its value. (default: \t).
+ * parse.headers : indicates if record headers are included in a line input and need to be parsed. (default: false).
+ * headers.delimiter : the string separating the list of headers from the record key. (default: \t).
+ * headers.key.separator : the string separating the key and value within a header. (default: :).
+ * ignore.error : whether best attempts should be made to ignore parsing errors. (default: false).
+ * null.marker : record key, record value, header key, header value which match this marker are replaced by null. (default: null).
+ *
+ */
+public final class LineMessageReader implements MessageReader {
+ private String topic;
+ private BufferedReader reader;
+ private boolean parseKey;
+ private String keySeparator = "\t";
+ private boolean parseHeaders;
+ private String headersDelimiter = "\t";
+ private String headersSeparator = ",";
+ private String headersKeySeparator = ":";
+ private boolean ignoreError;
+ private int lineNumber;
+ private boolean printPrompt = System.console() != null;
+ private Pattern headersSeparatorPattern;
+ private String nullMarker;
+
+ @Override
+ public void init(InputStream inputStream, Properties props) {
+ topic = props.getProperty("topic");
+ if (props.containsKey("parse.key"))
+ parseKey = props.getProperty("parse.key").trim().equalsIgnoreCase("true");
+ if (props.containsKey("key.separator"))
+ keySeparator = props.getProperty("key.separator");
+ if (props.containsKey("parse.headers"))
+ parseHeaders = props.getProperty("parse.headers").trim().equalsIgnoreCase("true");
+ if (props.containsKey("headers.delimiter"))
+ headersDelimiter = props.getProperty("headers.delimiter");
+ if (props.containsKey("headers.separator"))
+ headersSeparator = props.getProperty("headers.separator");
+ headersSeparatorPattern = Pattern.compile(headersSeparator);
+ if (props.containsKey("headers.key.separator"))
+ headersKeySeparator = props.getProperty("headers.key.separator");
+ if (props.containsKey("ignore.error"))
+ ignoreError = props.getProperty("ignore.error").trim().equalsIgnoreCase("true");
+ if (headersDelimiter.equals(headersSeparator))
+ throw new KafkaException("headers.delimiter and headers.separator may not be equal");
+ if (headersDelimiter.equals(headersKeySeparator))
+ throw new KafkaException("headers.delimiter and headers.key.separator may not be equal");
+ if (headersSeparator.equals(headersKeySeparator))
+ throw new KafkaException("headers.separator and headers.key.separator may not be equal");
+ if (props.containsKey("null.marker"))
+ nullMarker = props.getProperty("null.marker");
+ if (keySeparator.equals(nullMarker))
+ throw new KafkaException("null.marker and key.separator may not be equal");
+ if (headersSeparator.equals(nullMarker))
+ throw new KafkaException("null.marker and headers.separator may not be equal");
+ if (headersDelimiter.equals(nullMarker))
+ throw new KafkaException("null.marker and headers.delimiter may not be equal");
+ if (headersKeySeparator.equals(nullMarker))
+ throw new KafkaException("null.marker and headers.key.separator may not be equal");
+
+ reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public ProducerRecord readMessage() {
+ ++lineNumber;
+ if (printPrompt) {
+ System.out.print(">");
+ }
+
+ String line;
+ try {
+ line = reader.readLine();
+
+ } catch (IOException e) {
+ throw new KafkaException(e);
+ }
+
+ if (line == null) {
+ return null;
+ }
+
+ String headers = parse(parseHeaders, line, 0, headersDelimiter, "headers delimiter");
+ int headerOffset = headers == null ? 0 : headers.length() + headersDelimiter.length();
+
+ String key = parse(parseKey, line, headerOffset, keySeparator, "key separator");
+ int keyOffset = key == null ? 0 : key.length() + keySeparator.length();
+
+ String value = line.substring(headerOffset + keyOffset);
+
+ ProducerRecord record = new ProducerRecord<>(
+ topic,
+ key != null && !key.equals(nullMarker) ? key.getBytes(StandardCharsets.UTF_8) : null,
+ value != null && !value.equals(nullMarker) ? value.getBytes(StandardCharsets.UTF_8) : null
+ );
+
+ if (headers != null && !headers.equals(nullMarker)) {
+ stream(splitHeaders(headers)).forEach(header -> record.headers().add(header.key(), header.value()));
+ }
+
+ return record;
+ }
+
+ private String parse(boolean enabled, String line, int startIndex, String demarcation, String demarcationName) {
+ if (!enabled) {
+ return null;
+ }
+ int index = line.indexOf(demarcation, startIndex);
+ if (index == -1) {
+ if (ignoreError) {
+ return null;
+ }
+ throw new KafkaException("No " + demarcationName + " found on line number " + lineNumber + ": '" + line + "'");
+ }
+ return line.substring(startIndex, index);
+ }
+
+ private Header[] splitHeaders(String headers) {
+ return stream(headersSeparatorPattern.split(headers))
+ .map(pair -> {
+ int i = pair.indexOf(headersKeySeparator);
+ if (i == -1) {
+ if (ignoreError) {
+ return new RecordHeader(pair, null);
+ }
+ throw new KafkaException("No header key separator found in pair '" + pair + "' on line number " + lineNumber);
+ }
+
+ String headerKey = pair.substring(0, i);
+ if (headerKey.equals(nullMarker)) {
+ throw new KafkaException("Header keys should not be equal to the null marker '" + nullMarker + "' as they can't be null");
+ }
+
+ String value = pair.substring(i + headersKeySeparator.length());
+ byte[] headerValue = value.equals(nullMarker) ? null : value.getBytes(StandardCharsets.UTF_8);
+ return new RecordHeader(headerKey, headerValue);
+
+ }).toArray(Header[]::new);
+ }
+
+ // VisibleForTesting
+ String keySeparator() {
+ return keySeparator;
+ }
+
+ // VisibleForTesting
+ boolean parseKey() {
+ return parseKey;
+ }
+
+ // VisibleForTesting
+ boolean parseHeaders() {
+ return parseHeaders;
+ }
+}
diff --git a/tools/src/main/java/org/apache/kafka/tools/MessageReader.java b/tools/src/main/java/org/apache/kafka/tools/MessageReader.java
new file mode 100644
index 0000000000000..ef6322c51096c
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/MessageReader.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.io.InputStream;
+import java.util.Properties;
+
+/**
+ * Typical implementations of this interface convert data from an {@link InputStream} received via
+ * {@link MessageReader#init(InputStream, Properties)} into a {@link ProducerRecord} instance on each
+ * invocation of `{@link MessageReader#readMessage()}`.
+ *
+ * This is used by the {@link ConsoleProducer}.
+ */
+public interface MessageReader {
+
+ /**
+ * Sets the {@link InputStream} consumed by this reader, and provides configuration properties.
+ */
+ default void init(InputStream inputStream, Properties props) {}
+
+ /**
+ * Reads the next message in the underlying input stream.
+ */
+ ProducerRecord readMessage();
+
+ /**
+ * Closes this reader. There is no guarantee the underlying stream is closed.
+ */
+ default void close() {}
+
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
new file mode 100644
index 0000000000000..08c34d9af5c2b
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.test.TestUtils;
+import org.apache.kafka.tools.ConsoleProducer.ConsoleProducerConfig;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.util.Properties;
+import java.util.concurrent.Future;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Arrays.asList;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.kafka.clients.producer.ProducerConfig.BATCH_SIZE_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class ConsoleProducerTest {
+ @Mock
+ KafkaProducer producerMock;
+ @Mock
+ MessageReader readerMock;
+ @Spy
+ ConsoleProducer consoleProducerSpy;
+
+ private static final String[] BROKER_LIST_VALID_ARGS = new String[] {
+ "--broker-list",
+ "localhost:1001,localhost:1002",
+ "--topic",
+ "t3",
+ "--property",
+ "parse.key=true",
+ "--property",
+ "key.separator=#"
+ };
+
+ private static final String[] BOOTSTRAP_SERVER_VALID_ARGS = new String[] {
+ "--bootstrap-server",
+ "localhost:1003,localhost:1004",
+ "--topic",
+ "t3",
+ "--property",
+ "parse.key=true",
+ "--property",
+ "key.separator=#"
+ };
+
+ private static final String[] INVALID_ARGS = new String[] {
+ "--t", // not a valid argument
+ "t3"
+ };
+
+ private static final String[] BOOTSTRAP_SERVER_OVERRIDE = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--bootstrap-server",
+ "localhost:1002",
+ "--topic",
+ "t3"
+ };
+
+ private static final String[] CLIENT_ID_OVERRIDE = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--topic",
+ "t3",
+ "--producer-property",
+ "client.id=producer-1"
+ };
+
+ private static final String[] BATCH_SIZE_OVERRIDDEN_BY_MAX_PARTITION_MEMORY_BYTES_VALUE = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--bootstrap-server",
+ "localhost:1002",
+ "--topic",
+ "t3",
+ "--batch-size",
+ "123",
+ "--max-partition-memory-bytes",
+ "456"
+ };
+
+ private static final String[] BATCH_SIZE_SET_AND_MAX_PARTITION_MEMORY_BYTES_NOT_SET = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--bootstrap-server",
+ "localhost:1002",
+ "--topic",
+ "t3",
+ "--batch-size",
+ "123"
+ };
+
+ private static final String[] BATCH_SIZE_NOT_SET_AND_MAX_PARTITION_MEMORY_BYTES_SET = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--bootstrap-server",
+ "localhost:1002",
+ "--topic",
+ "t3",
+ "--max-partition-memory-bytes",
+ "456"
+ };
+
+ private static final String[] BATCH_SIZE_DEFAULT = new String[] {
+ "--broker-list",
+ "localhost:1001",
+ "--bootstrap-server",
+ "localhost:1002",
+ "--topic",
+ "t3"
+ };
+
+ @Test
+ public void testValidConfigsBrokerList() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BROKER_LIST_VALID_ARGS);
+ ProducerConfig consoleProducerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(asList("localhost:1001", "localhost:1002"),
+ consoleProducerConfig.getList(BOOTSTRAP_SERVERS_CONFIG));
+ }
+
+ @Test
+ public void testValidConfigsBootstrapServer() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BOOTSTRAP_SERVER_VALID_ARGS);
+ ProducerConfig consoleProducerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(asList("localhost:1003", "localhost:1004"),
+ consoleProducerConfig.getList(BOOTSTRAP_SERVERS_CONFIG));
+ }
+
+ @Test
+ public void testInvalidConfigs() {
+ Exit.setExitProcedure((statusCode, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleProducerConfig(INVALID_ARGS));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void testParseKeyProp() throws Exception {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BROKER_LIST_VALID_ARGS);
+ LineMessageReader reader = (LineMessageReader) Class.forName(config.readerClass()).getDeclaredConstructor().newInstance();
+ reader.init(System.in, config.getReaderProps());
+
+ assertEquals("#", reader.keySeparator());
+ assertTrue(reader.parseKey());
+ }
+
+ @Test
+ public void testParseReaderConfigFile() throws Exception {
+ File propsFile = TestUtils.tempFile();
+ OutputStream propsStream = Files.newOutputStream(propsFile.toPath());
+ propsStream.write("parse.key=true\n".getBytes());
+ propsStream.write("key.separator=|".getBytes());
+ propsStream.close();
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--property", "key.separator=;",
+ "--property", "parse.headers=true",
+ "--reader-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleProducerConfig config = new ConsoleProducerConfig(args);
+ LineMessageReader reader = (LineMessageReader) Class.forName(config.readerClass()).getDeclaredConstructor().newInstance();
+ reader.init(System.in, config.getReaderProps());
+
+ assertEquals(";", reader.keySeparator());
+ assertTrue(reader.parseKey());
+ assertTrue(reader.parseHeaders());
+ }
+
+ @Test
+ public void testBootstrapServerOverride() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BOOTSTRAP_SERVER_OVERRIDE);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(asList("localhost:1002"), producerConfig.getList(BOOTSTRAP_SERVERS_CONFIG));
+ }
+
+ @Test
+ public void testClientIdOverride() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(CLIENT_ID_OVERRIDE);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals("producer-1", producerConfig.getString(CLIENT_ID_CONFIG));
+ }
+
+ @Test
+ public void testDefaultClientId() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BROKER_LIST_VALID_ARGS);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals("console-producer", producerConfig.getString(CLIENT_ID_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeOverriddenByMaxPartitionMemoryBytesValue() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BATCH_SIZE_OVERRIDDEN_BY_MAX_PARTITION_MEMORY_BYTES_VALUE);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(456, producerConfig.getInt(BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeSetAndMaxPartitionMemoryBytesNotSet() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BATCH_SIZE_SET_AND_MAX_PARTITION_MEMORY_BYTES_NOT_SET);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(123, producerConfig.getInt(BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testDefaultBatchSize() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BATCH_SIZE_DEFAULT);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(16 * 1024, producerConfig.getInt(BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeNotSetAndMaxPartitionMemoryBytesSet() throws IOException {
+ ConsoleProducerConfig config = new ConsoleProducerConfig(BATCH_SIZE_NOT_SET_AND_MAX_PARTITION_MEMORY_BYTES_SET);
+ ProducerConfig producerConfig = new ProducerConfig(config.getProducerProps());
+
+ assertEquals(456, producerConfig.getInt(BATCH_SIZE_CONFIG));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = { false, true })
+ public void testRecordsFromTheReaderAreSentByTheProducer(boolean sync) throws Exception {
+ Exit.setExitProcedure((statusCode, message) -> {
+ if (statusCode != 0) {
+ throw new AssertionError(message);
+ }
+ });
+
+ try {
+ String topic = "p1nKFl0yD";
+ byte[] one = "1".getBytes(UTF_8), two = "2".getBytes(UTF_8), three = "3".getBytes(UTF_8);
+ ProducerRecord
+ r1 = new ProducerRecord<>(topic, one, one),
+ r2 = new ProducerRecord<>(topic, two, two),
+ r3 = new ProducerRecord<>(topic, three, three);
+
+ Future producerResponse = completedFuture(null);
+
+ if (sync) {
+ doReturn(producerResponse).when(producerMock).send(any());
+
+ } else {
+ doReturn(producerResponse).when(producerMock).send(any(), any());
+ }
+
+ doReturn(readerMock).when(consoleProducerSpy).createMessageReader(any(ConsoleProducerConfig.class));
+ doReturn(producerMock).when(consoleProducerSpy).createKafkaProducer(any(Properties.class));
+ when(readerMock.readMessage())
+ .thenReturn(r1)
+ .thenReturn(r2)
+ .thenReturn(r3)
+ .thenReturn(null);
+
+ String[] args = new String[] {
+ "--bootstrap-server", "localhost:9092",
+ "--topic", topic,
+ sync ? "--sync" : ""
+ };
+
+ consoleProducerSpy.start(args);
+
+ if (sync) {
+ verify(producerMock).send(eq(r1));
+ verify(producerMock).send(eq(r2));
+ verify(producerMock).send(eq(r3));
+
+ } else {
+ verify(producerMock).send(eq(r1), any(Callback.class));
+ verify(producerMock).send(eq(r2), any(Callback.class));
+ verify(producerMock).send(eq(r3), any(Callback.class));
+ }
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/LineMessageReaderTest.java b/tools/src/test/java/org/apache/kafka/tools/LineMessageReaderTest.java
new file mode 100644
index 0000000000000..8048fbbc7a32b
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/LineMessageReaderTest.java
@@ -0,0 +1,380 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.util.List;
+import java.util.Properties;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class LineMessageReaderTest {
+ private static Properties defaultTestProps() {
+ Properties props = new Properties();
+ props.put("topic", "topic");
+ props.put("parse.key", "true");
+ props.put("parse.headers", "true");
+ return props;
+ }
+
+ @Test
+ public void testLineReader() {
+ String input = "key0\tvalue0\nkey1\tvalue1";
+
+ Properties props = defaultTestProps();
+ props.put("parse.headers", "false");
+
+ runTest(props, input, record("key0", "value0"), record("key1", "value1"));
+ }
+
+ @Test
+ public void testLineReaderHeader() {
+ String input = "headerKey0:headerValue0,headerKey1:headerValue1\tkey0\tvalue0\n";
+ ProducerRecord expected = record(
+ "key0",
+ "value0",
+ asList(new RecordHeader("headerKey0", "headerValue0".getBytes(UTF_8)),
+ new RecordHeader("headerKey1", "headerValue1".getBytes(UTF_8))));
+
+ runTest(defaultTestProps(), input, expected);
+ }
+
+ @Test
+ public void testMinimalValidInputWithHeaderKeyAndValue() {
+ runTest(defaultTestProps(), ":\t\t",
+ record("", "", singletonList(new RecordHeader("", "".getBytes(UTF_8)))));
+ }
+
+ @Test
+ public void testKeyMissingValue() {
+ Properties props = defaultTestProps();
+ props.put("parse.headers", "false");
+ runTest(props, "key\t", record("key", ""));
+ }
+
+ @Test
+ public void testDemarcationsLongerThanOne() {
+ Properties props = defaultTestProps();
+ props.put("key.separator", "\t\t");
+ props.put("headers.delimiter", "\t\t");
+ props.put("headers.separator", "---");
+ props.put("headers.key.separator", "::::");
+
+ runTest(
+ props,
+ "headerKey0.0::::headerValue0.0---headerKey1.0::::\t\tkey\t\tvalue",
+ record("key",
+ "value",
+ asList(new RecordHeader("headerKey0.0", "headerValue0.0".getBytes(UTF_8)),
+ new RecordHeader("headerKey1.0", "".getBytes(UTF_8)))));
+ }
+
+ @Test
+ public void testLineReaderHeaderNoKey() {
+ String input = "headerKey:headerValue\tvalue\n";
+
+ Properties props = defaultTestProps();
+ props.put("parse.key", "false");
+
+ runTest(props, input, record(null, "value",
+ singletonList(new RecordHeader("headerKey", "headerValue".getBytes(UTF_8)))));
+ }
+
+ @Test
+ public void testLineReaderOnlyValue() {
+ Properties props = defaultTestProps();
+ props.put("parse.key", "false");
+ props.put("parse.headers", "false");
+
+ runTest(props, "value\n", record(null, "value"));
+ }
+
+ @Test
+ public void testParseHeaderEnabledWithCustomDelimiterAndVaryingNumberOfKeyValueHeaderPairs() {
+ Properties props = defaultTestProps();
+ props.put("key.separator", "#");
+ props.put("headers.delimiter", "!");
+ props.put("headers.separator", "&");
+ props.put("headers.key.separator", ":");
+
+ String input =
+ "headerKey0.0:headerValue0.0&headerKey0.1:headerValue0.1!key0#value0\n" +
+ "headerKey1.0:headerValue1.0!key1#value1";
+
+ ProducerRecord record0 = record(
+ "key0",
+ "value0",
+ asList(new RecordHeader("headerKey0.0", "headerValue0.0".getBytes(UTF_8)),
+ new RecordHeader("headerKey0.1", "headerValue0.1".getBytes(UTF_8))));
+
+ ProducerRecord record1 = record(
+ "key1",
+ "value1",
+ asList(new RecordHeader("headerKey1.0", "headerValue1.0".getBytes(UTF_8))));
+
+ runTest(props, input, record0, record1);
+ }
+
+ @Test
+ public void testMissingKeySeparator() {
+ MessageReader lineReader = new LineMessageReader();
+ String input =
+ "headerKey0.0:headerValue0.0,headerKey0.1:headerValue0.1\tkey0\tvalue0\n" +
+ "headerKey1.0:headerValue1.0\tkey1[MISSING-DELIMITER]value1";
+
+ lineReader.init(new ByteArrayInputStream(input.getBytes(UTF_8)), defaultTestProps());
+ lineReader.readMessage();
+
+ KafkaException expectedException = assertThrows(KafkaException.class, lineReader::readMessage);
+
+ assertEquals(
+ "No key separator found on line number 2: 'headerKey1.0:headerValue1.0\tkey1[MISSING-DELIMITER]value1'",
+ expectedException.getMessage()
+ );
+ }
+
+ @Test
+ public void testMissingHeaderKeySeparator() {
+ MessageReader lineReader = new LineMessageReader();
+ String input = "key[MISSING-DELIMITER]val\tkey0\tvalue0\n";
+ lineReader.init(new ByteArrayInputStream(input.getBytes(UTF_8)), defaultTestProps());
+
+ KafkaException expectedException = assertThrows(KafkaException.class, lineReader::readMessage);
+
+ assertEquals(
+ "No header key separator found in pair 'key[MISSING-DELIMITER]val' on line number 1",
+ expectedException.getMessage()
+ );
+ }
+
+ @Test
+ public void testHeaderDemarcationCollision() {
+ Properties props = defaultTestProps();
+ props.put("headers.delimiter", "\t");
+ props.put("headers.separator", "\t");
+ props.put("headers.key.separator", "\t");
+
+ assertThrowsOnInvalidPatternConfig(props, "headers.delimiter and headers.separator may not be equal");
+
+ props.put("headers.separator", ",");
+ assertThrowsOnInvalidPatternConfig(props, "headers.delimiter and headers.key.separator may not be equal");
+
+ props.put("headers.key.separator", ",");
+ assertThrowsOnInvalidPatternConfig(props, "headers.separator and headers.key.separator may not be equal");
+ }
+
+ private static void assertThrowsOnInvalidPatternConfig(Properties props, String expectedMessage) {
+ KafkaException exception = assertThrows(KafkaException.class, () -> new LineMessageReader().init(null, props));
+ assertEquals(expectedMessage, exception.getMessage());
+ }
+
+ @Test
+ public void testIgnoreErrorInInput() {
+ String input =
+ "headerKey0.0:headerValue0.0\tkey0\tvalue0\n" +
+ "headerKey1.0:headerValue1.0,headerKey1.1:headerValue1.1[MISSING-HEADER-DELIMITER]key1\tvalue1\n" +
+ "headerKey2.0:headerValue2.0\tkey2[MISSING-KEY-DELIMITER]value2\n" +
+ "headerKey3.0:headerValue3.0[MISSING-HEADER-DELIMITER]key3[MISSING-KEY-DELIMITER]value3\n";
+
+ Properties props = defaultTestProps();
+ props.put("ignore.error", "true");
+
+ ProducerRecord validRecord = record(
+ "key0",
+ "value0",
+ asList(new RecordHeader("headerKey0.0", "headerValue0.0".getBytes(UTF_8))));
+
+ ProducerRecord missingHeaderDelimiter = record(
+ null,
+ "value1",
+ asList(new RecordHeader("headerKey1.0", "headerValue1.0".getBytes(UTF_8)),
+ new RecordHeader("headerKey1.1", "headerValue1.1[MISSING-HEADER-DELIMITER]key1".getBytes(UTF_8))));
+
+ ProducerRecord missingKeyDelimiter = record(
+ null,
+ "key2[MISSING-KEY-DELIMITER]value2",
+ asList(new RecordHeader("headerKey2.0", "headerValue2.0".getBytes(UTF_8))));
+
+ ProducerRecord missingKeyHeaderDelimiter = record(
+ null,
+ "headerKey3.0:headerValue3.0[MISSING-HEADER-DELIMITER]key3[MISSING-KEY-DELIMITER]value3",
+ emptyList());
+
+ runTest(props, input, validRecord, missingHeaderDelimiter, missingKeyDelimiter, missingKeyHeaderDelimiter);
+ }
+
+ @Test
+ public void testMalformedHeaderIgnoreError() {
+ String input = "key-val\tkey0\tvalue0\n";
+
+ Properties props = defaultTestProps();
+ props.put("ignore.error", "true");
+
+ ProducerRecord expected = record(
+ "key0",
+ "value0",
+ singletonList(new RecordHeader("key-val", null)));
+
+ runTest(props, input, expected);
+ }
+
+ @Test
+ public void testNullMarker() {
+ String input =
+ "key\t\n" +
+ "key\t\n" +
+ "key\tvalue\n" +
+ "\tvalue\n" +
+ "\t";
+
+ Properties props = defaultTestProps();
+ props.put("null.marker", "");
+ props.put("parse.headers", "false");
+ runTest(props, input,
+ record("key", ""),
+ record("key", null),
+ record("key", "value"),
+ record(null, "value"),
+ record(null, null));
+
+ // If the null marker is not set
+ props.remove("null.marker");
+ runTest(props, input,
+ record("key", ""),
+ record("key", ""),
+ record("key", "value"),
+ record("", "value"),
+ record("", ""));
+ }
+
+ @Test
+ public void testNullMarkerWithHeaders() {
+ String input =
+ "h0:v0,h1:v1\t\tvalue\n" +
+ "\tkey\t\n" +
+ "h0:,h1:v1\t\t\n" +
+ "h0:,h1:v1\tkey\t\n" +
+ "h0:,h1:value\tkey\t\n";
+ Header header = new RecordHeader("h1", "v1".getBytes(UTF_8));
+
+ Properties props = defaultTestProps();
+ props.put("null.marker", "");
+ runTest(props, input,
+ record(null, "value", asList(new RecordHeader("h0", "v0".getBytes(UTF_8)), header)),
+ record("key", null),
+ record(null, null, asList(new RecordHeader("h0", "".getBytes(UTF_8)), header)),
+ record("key", null, asList(new RecordHeader("h0", null), header)),
+ record("key", null, asList(new RecordHeader("h0", null), new RecordHeader("h1", "value".getBytes(UTF_8)))));
+
+ // If the null marker is not set
+ MessageReader lineReader = new LineMessageReader();
+ props.remove("null.marker");
+ lineReader.init(new ByteArrayInputStream(input.getBytes(UTF_8)), props);
+ assertRecordEquals(record(
+ "",
+ "value",
+ asList(new RecordHeader("h0", "v0".getBytes(UTF_8)), header)),
+ lineReader.readMessage());
+ // line 2 is not valid anymore
+ KafkaException expectedException = assertThrows(KafkaException.class, lineReader::readMessage);
+ assertEquals(
+ "No header key separator found in pair '' on line number 2",
+ expectedException.getMessage()
+ );
+ assertRecordEquals(record("", "", asList(new RecordHeader("h0", "".getBytes(UTF_8)), header)), lineReader.readMessage());
+ assertRecordEquals(record("key", "", asList(new RecordHeader("h0", "".getBytes(UTF_8)), header)), lineReader.readMessage());
+ assertRecordEquals(record("key", "", asList(new RecordHeader("h0", "".getBytes(UTF_8)), new RecordHeader("h1", "value".getBytes(UTF_8)))), lineReader.readMessage());
+ }
+
+ @Test
+ public void testNullMarkerHeaderKeyThrows() {
+ String input = ":v0,h1:v1\tkey\tvalue\n";
+
+ Properties props = defaultTestProps();
+ props.put("null.marker", "");
+ MessageReader lineReader = new LineMessageReader();
+ lineReader.init(new ByteArrayInputStream(input.getBytes(UTF_8)), props);
+ KafkaException expectedException = assertThrows(KafkaException.class, lineReader::readMessage);
+ assertEquals(
+ "Header keys should not be equal to the null marker '' as they can't be null",
+ expectedException.getMessage()
+ );
+
+ // If the null marker is not set
+ props.remove("null.marker");
+ runTest(props, input, record(
+ "key",
+ "value",
+ asList(new RecordHeader("", "v0".getBytes(UTF_8)), new RecordHeader("h1", "v1".getBytes(UTF_8)))));
+ }
+
+ @Test
+ public void testInvalidNullMarker() {
+ Properties props = defaultTestProps();
+ props.put("headers.delimiter", "-");
+ props.put("headers.separator", ":");
+ props.put("headers.key.separator", "/");
+
+ props.put("null.marker", "-");
+ assertThrowsOnInvalidPatternConfig(props, "null.marker and headers.delimiter may not be equal");
+
+ props.put("null.marker", ":");
+ assertThrowsOnInvalidPatternConfig(props, "null.marker and headers.separator may not be equal");
+
+ props.put("null.marker", "/");
+ assertThrowsOnInvalidPatternConfig(props, "null.marker and headers.key.separator may not be equal");
+ }
+
+ @SafeVarargs
+ private static void runTest(Properties props, String input, ProducerRecord... expectedRecords) {
+ MessageReader lineReader = new LineMessageReader();
+ lineReader.init(new ByteArrayInputStream(input.getBytes(UTF_8)), props);
+
+ for (ProducerRecord expectedRecord: expectedRecords) {
+ assertRecordEquals(expectedRecord, lineReader.readMessage());
+ }
+ }
+
+ // The equality method of ProducerRecord compares memory references for the header iterator, this is why this custom equality check is used.
+ private static void assertRecordEquals(ProducerRecord expected, ProducerRecord actual) {
+ assertEquals(expected.key(), actual.key() == null ? null : new String(actual.key()));
+ assertEquals(expected.value(), actual.value() == null ? null : new String(actual.value()));
+ assertArrayEquals(expected.headers().toArray(), actual.headers().toArray());
+ }
+
+ private static ProducerRecord record(K key, V value, List headers) {
+ ProducerRecord record = new ProducerRecord<>("topic", key, value);
+ headers.forEach(h -> record.headers().add(h.key(), h.value()));
+ return record;
+ }
+
+ private static ProducerRecord record(K key, V value) {
+ return new ProducerRecord<>("topic", key, value);
+ }
+}