properties = new HashMap<>();
+
+ if (options.has(readerConfigOpt)) {
+ properties.putAll(propsToStringMap(loadProps(options.valueOf(readerConfigOpt))));
+ }
+
+ properties.put("topic", options.valueOf(topicOpt));
+ properties.putAll(propsToStringMap(parseKeyValueArgs(options.valuesOf(propertyOpt))));
+
+ return properties;
+ }
+
+ Properties producerProps() throws IOException {
+ Properties props = new Properties();
+
+ if (options.has(producerConfigOpt)) {
+ props.putAll(loadProps(options.valueOf(producerConfigOpt)));
+ }
+
+ props.putAll(parseKeyValueArgs(options.valuesOf(producerPropertyOpt)));
+ props.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
+ props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
+
+ if (props.getProperty(CLIENT_ID_CONFIG) == null) {
+ props.put(CLIENT_ID_CONFIG, "console-producer");
+ }
+
+ props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+ props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+
+ CommandLineUtils.maybeMergeOptions(props, LINGER_MS_CONFIG, options, sendTimeoutOpt);
+ CommandLineUtils.maybeMergeOptions(props, ACKS_CONFIG, options, requestRequiredAcksOpt);
+ CommandLineUtils.maybeMergeOptions(props, REQUEST_TIMEOUT_MS_CONFIG, options, requestTimeoutMsOpt);
+ CommandLineUtils.maybeMergeOptions(props, RETRIES_CONFIG, options, messageSendMaxRetriesOpt);
+ CommandLineUtils.maybeMergeOptions(props, RETRY_BACKOFF_MS_CONFIG, options, retryBackoffMsOpt);
+ CommandLineUtils.maybeMergeOptions(props, SEND_BUFFER_CONFIG, options, socketBufferSizeOpt);
+ CommandLineUtils.maybeMergeOptions(props, 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.
+ CommandLineUtils.maybeMergeOptions(props, BATCH_SIZE_CONFIG, options, batchSizeOpt);
+ CommandLineUtils.maybeMergeOptions(props, BATCH_SIZE_CONFIG, options, maxPartitionMemoryBytesOpt);
+ CommandLineUtils.maybeMergeOptions(props, METADATA_MAX_AGE_CONFIG, options, metadataExpiryMsOpt);
+ CommandLineUtils.maybeMergeOptions(props, MAX_BLOCK_MS_CONFIG, options, maxBlockMsOpt);
+
+ return props;
+ }
+ }
+}
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..13f90d3745eac
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/LineMessageReader.java
@@ -0,0 +1,218 @@
+/*
+ * 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.apache.kafka.tools.api.RecordReader;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.regex.Pattern;
+
+import static java.util.Arrays.stream;
+
+/**
+ * The default implementation of {@link RecordReader} 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.separator : the string separating headers. (default: ,).
+ * 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 class LineMessageReader implements RecordReader {
+ private String topic;
+ 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 final boolean printPrompt = System.console() != null;
+ private Pattern headersSeparatorPattern;
+ private String nullMarker;
+
+ @Override
+ public void configure(Map props) {
+ topic = props.get("topic").toString();
+ if (props.containsKey("parse.key"))
+ parseKey = props.get("parse.key").toString().trim().equalsIgnoreCase("true");
+ if (props.containsKey("key.separator"))
+ keySeparator = props.get("key.separator").toString();
+ if (props.containsKey("parse.headers"))
+ parseHeaders = props.get("parse.headers").toString().trim().equalsIgnoreCase("true");
+ if (props.containsKey("headers.delimiter"))
+ headersDelimiter = props.get("headers.delimiter").toString();
+ if (props.containsKey("headers.separator"))
+ headersSeparator = props.get("headers.separator").toString();
+ headersSeparatorPattern = Pattern.compile(headersSeparator);
+ if (props.containsKey("headers.key.separator"))
+ headersKeySeparator = props.get("headers.key.separator").toString();
+ if (props.containsKey("ignore.error"))
+ ignoreError = props.get("ignore.error").toString().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.get("null.marker").toString();
+ 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");
+ }
+
+ @Override
+ public Iterator> readRecords(InputStream inputStream) {
+ return new Iterator>() {
+ private final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+ private ProducerRecord current;
+
+ @Override
+ public boolean hasNext() {
+ if (current != null) {
+ return true;
+ } else {
+ lineNumber += 1;
+ if (printPrompt) {
+ System.out.print(">");
+ }
+
+ String line;
+ try {
+ line = reader.readLine();
+ } catch (IOException e) {
+ throw new KafkaException(e);
+ }
+
+ if (line == null) {
+ current = null;
+ } else {
+ 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()));
+ }
+ current = record;
+ }
+
+ return current != null;
+ }
+ }
+
+ @Override
+ public ProducerRecord next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException("no more record");
+ } else {
+ try {
+ return current;
+ } finally {
+ current = null;
+ }
+ }
+ }
+ };
+ }
+
+ 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);
+ }
+
+ // Visible for testing
+ String keySeparator() {
+ return keySeparator;
+ }
+
+ // Visible for testing
+ boolean parseKey() {
+ return parseKey;
+ }
+
+ // Visible for testing
+ boolean parseHeaders() {
+ return parseHeaders;
+ }
+}
diff --git a/tools/src/main/java/org/apache/kafka/tools/ToolsUtils.java b/tools/src/main/java/org/apache/kafka/tools/ToolsUtils.java
index bbc30bea01e6f..57239ee4b89ee 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ToolsUtils.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ToolsUtils.java
@@ -161,8 +161,7 @@ public static Set minus(Set set, T...toRemove) {
/**
* This is a simple wrapper around `CommandLineUtils.printUsageAndExit`.
* It is needed for tools migration (KAFKA-14525), as there is no Java equivalent for return type `Nothing`.
- * Can be removed once [[kafka.tools.ConsoleConsumer]]
- * and [[kafka.tools.ConsoleProducer]] are migrated.
+ * Can be removed once [[kafka.tools.ConsoleConsumer]] are migrated.
*
* @param parser Command line options parser.
* @param message Error message.
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..3804224ef1750
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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 kafka.utils.TestUtils;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.tools.ConsoleProducer.ConsoleProducerOptions;
+import org.apache.kafka.tools.api.RecordReader;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ConsoleProducerTest {
+ 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[]{
+ "--bootstrap-server", "localhost:1002",
+ "--topic", "t3",
+ };
+ private static final String[] CLIENT_ID_OVERRIDE = new String[]{
+ "--bootstrap-server", "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[]{
+ "--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[]{
+ "--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[]{
+ "--bootstrap-server", "localhost:1002",
+ "--topic", "t3",
+ "--max-partition-memory-bytes", "456"
+ };
+ private static final String[] BATCH_SIZE_DEFAULT = new String[]{
+ "--bootstrap-server", "localhost:1002",
+ "--topic", "t3"
+ };
+ private static final String[] TEST_RECORD_READER = new String[]{
+ "--bootstrap-server", "localhost:1002",
+ "--topic", "t3",
+ "--line-reader", TestRecordReader.class.getName()
+ };
+
+ @Test
+ public void testValidConfigsBootstrapServer() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_VALID_ARGS);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(asList("localhost:1003", "localhost:1004"), producerConfig.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+ }
+
+ @Test
+ public void testInvalidConfigs() {
+ Exit.setExitProcedure((statusCode, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleProducerOptions(INVALID_ARGS));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void testParseKeyProp() throws ReflectiveOperationException, IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_VALID_ARGS);
+ LineMessageReader reader = (LineMessageReader) Class.forName(opts.readerClass()).getDeclaredConstructor().newInstance();
+ reader.configure(opts.readerProps());
+
+ 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()
+ };
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ LineMessageReader reader = (LineMessageReader) Class.forName(opts.readerClass()).getDeclaredConstructor().newInstance();
+ reader.configure(opts.readerProps());
+
+ assertEquals(";", reader.keySeparator());
+ assertTrue(reader.parseKey());
+ assertTrue(reader.parseHeaders());
+ }
+
+ @Test
+ public void testBootstrapServerOverride() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_OVERRIDE);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(Collections.singletonList("localhost:1002"), producerConfig.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+ }
+
+ @Test
+ public void testClientIdOverride() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(CLIENT_ID_OVERRIDE);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("producer-1", producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG));
+ }
+
+ @Test
+ public void testDefaultClientId() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_VALID_ARGS);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("console-producer", producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeOverriddenByMaxPartitionMemoryBytesValue() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BATCH_SIZE_OVERRIDDEN_BY_MAX_PARTITION_MEMORY_BYTES_VALUE);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(456, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeSetAndMaxPartitionMemoryBytesNotSet() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BATCH_SIZE_SET_AND_MAX_PARTITION_MEMORY_BYTES_NOT_SET);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(123, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testDefaultBatchSize() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BATCH_SIZE_DEFAULT);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(16 * 1024, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testBatchSizeNotSetAndMaxPartitionMemoryBytesSet() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BATCH_SIZE_NOT_SET_AND_MAX_PARTITION_MEMORY_BYTES_SET);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals(456, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testNewReader() throws Exception {
+ ConsoleProducer producer = new ConsoleProducer();
+ TestRecordReader reader = (TestRecordReader) producer.messageReader(new ConsoleProducerOptions(TEST_RECORD_READER));
+
+ assertEquals(1, reader.configureCount());
+ assertEquals(0, reader.closeCount());
+
+ reader.close();
+ assertEquals(1, reader.closeCount());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testLoopReader() throws Exception {
+ ConsoleProducer producer = new ConsoleProducer();
+ TestRecordReader reader = (TestRecordReader) producer.messageReader(new ConsoleProducerOptions(TEST_RECORD_READER));
+
+ producer.loopReader(Mockito.mock(Producer.class), reader, false);
+
+ assertEquals(1, reader.configureCount());
+ assertEquals(1, reader.closeCount());
+ }
+
+ public static class TestRecordReader implements RecordReader {
+ private int configureCount = 0;
+ private int closeCount = 0;
+
+ @Override
+ public void configure(Map configs) {
+ configureCount += 1;
+ }
+
+ @Override
+ public Iterator> readRecords(InputStream inputStream) {
+ return Collections.emptyIterator();
+ }
+
+ @Override
+ public void close() {
+ closeCount += 1;
+ }
+
+ public int configureCount() {
+ return configureCount;
+ }
+
+ public int closeCount() {
+ return closeCount;
+ }
+ }
+}
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..5864869df1a16
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/LineMessageReaderTest.java
@@ -0,0 +1,403 @@
+/*
+ * 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.apache.kafka.tools.api.RecordReader;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Properties;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+import static org.apache.kafka.common.utils.Utils.propsToStringMap;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class LineMessageReaderTest {
+ @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",
+ singletonList(new RecordHeader("headerKey1.0", "headerValue1.0".getBytes(UTF_8)))
+ );
+
+ runTest(props, input, record0, record1);
+ }
+
+ @Test
+ public void testMissingKeySeparator() {
+ RecordReader 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.configure(propsToStringMap(defaultTestProps()));
+ Iterator> iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes()));
+ iter.next();
+
+ KafkaException expectedException = assertThrows(KafkaException.class, iter::next);
+
+ assertEquals(
+ "No key separator found on line number 2: 'headerKey1.0:headerValue1.0\tkey1[MISSING-DELIMITER]value1'",
+ expectedException.getMessage()
+ );
+ }
+
+ @Test
+ public void testMissingHeaderKeySeparator() {
+ RecordReader lineReader = new LineMessageReader();
+ String input = "key[MISSING-DELIMITER]val\tkey0\tvalue0\n";
+ lineReader.configure(propsToStringMap(defaultTestProps()));
+ Iterator> iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes()));
+
+ KafkaException expectedException = assertThrows(KafkaException.class, iter::next);
+
+ 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");
+ }
+
+ @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",
+ singletonList(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",
+ singletonList(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",
+ Collections.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
+ RecordReader lineReader = new LineMessageReader();
+ props.remove("null.marker");
+ lineReader.configure(propsToStringMap(props));
+ Iterator> iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes()));
+ assertRecordEquals(record("", "value", asList(new RecordHeader("h0", "v0".getBytes(UTF_8)), header)), iter.next());
+ // line 2 is not valid anymore
+ KafkaException expectedException = assertThrows(KafkaException.class, iter::next);
+ assertEquals(
+ "No header key separator found in pair '' on line number 2",
+ expectedException.getMessage()
+ );
+ assertRecordEquals(record("", "", asList(new RecordHeader("h0", "".getBytes(UTF_8)), header)), iter.next());
+ assertRecordEquals(record("key", "", asList(new RecordHeader("h0", "".getBytes(UTF_8)), header)), iter.next());
+ assertRecordEquals(record("key", "", asList(
+ new RecordHeader("h0", "".getBytes(UTF_8)),
+ new RecordHeader("h1", "value".getBytes(UTF_8)))), iter.next()
+ );
+ }
+
+ @Test
+ public void testNullMarkerHeaderKeyThrows() {
+ String input = ":v0,h1:v1\tkey\tvalue\n";
+
+ Properties props = defaultTestProps();
+ props.put("null.marker", "");
+ RecordReader lineReader = new LineMessageReader();
+ lineReader.configure(propsToStringMap(props));
+ Iterator> iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes()));
+ KafkaException expectedException = assertThrows(KafkaException.class, iter::next);
+ 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");
+ }
+
+ private Properties defaultTestProps() {
+ Properties props = new Properties();
+ props.put("topic", "topic");
+ props.put("parse.key", "true");
+ props.put("parse.headers", "true");
+
+ return props;
+ }
+
+ private void assertThrowsOnInvalidPatternConfig(Properties props, String expectedMessage) {
+ KafkaException exception = assertThrows(KafkaException.class, () -> new LineMessageReader().configure(propsToStringMap(props)));
+ assertEquals(expectedMessage, exception.getMessage());
+ }
+
+ @SafeVarargs
+ private final void runTest(Properties props, String input, ProducerRecord... expectedRecords) {
+ RecordReader lineReader = new LineMessageReader();
+ lineReader.configure(propsToStringMap(props));
+ Iterator> iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes()));
+
+ for (ProducerRecord record : expectedRecords) {
+ assertRecordEquals(record, iter.next());
+ }
+
+ assertFalse(iter.hasNext());
+ assertThrows(NoSuchElementException.class, iter::next);
+ }
+
+ // The equality method of ProducerRecord compares memory references for the header iterator, this is why this custom equality check is used.
+ private 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 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 ProducerRecord record(K key, V value) {
+ return new ProducerRecord<>("topic", key, value);
+ }
+}
diff --git a/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java b/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
index 305e68af7ad66..f8340ddcda1f8 100644
--- a/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
+++ b/tools/tools-api/src/main/java/org/apache/kafka/tools/api/RecordReader.java
@@ -28,7 +28,7 @@
* Typical implementations of this interface convert data from an `InputStream` received via `readRecords` into a
* iterator of `ProducerRecord` instance. Note that implementations must have a public nullary constructor.
*
- * This is used by the `kafka.tools.ConsoleProducer`.
+ * This is used by the `org.apache.kafka.tools.ConsoleProducer`.
*/
public interface RecordReader extends Closeable, Configurable {