diff --git a/clients/src/main/java/org/apache/kafka/tools/RecordReader.java b/clients/src/main/java/org/apache/kafka/tools/RecordReader.java new file mode 100644 index 0000000000000..635dec6ee0358 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/tools/RecordReader.java @@ -0,0 +1,51 @@ +/* + * 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.Configurable; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Iterator; +import java.util.Map; + +/** + * 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`. + */ +public interface RecordReader extends Closeable, Configurable { + + default void configure(Map configs) {} + + /** + * read byte array from input stream and then generate an iterator of producer record + * @param {@link InputStream} of messages. the implementation does not need to close the input stream. + * @return an iterator of producer record. It should implement following rules. 1) the hasNext() method must be idempotent. + * 2) the convert error should be thrown by next() method. + */ + Iterator> readRecords(InputStream inputStream); + + + /** + * Closes this reader. + * This method is invoked if the iterator from readRecords either has no more records or throws exception. + */ + default void close() {} +} diff --git a/core/src/main/scala/kafka/common/MessageReader.scala b/core/src/main/scala/kafka/common/MessageReader.scala index de456e16ae532..304e15f320429 100644 --- a/core/src/main/scala/kafka/common/MessageReader.scala +++ b/core/src/main/scala/kafka/common/MessageReader.scala @@ -28,6 +28,8 @@ import org.apache.kafka.clients.producer.ProducerRecord * * This is used by the `ConsoleProducer`. */ +@deprecated("This class has been deprecated and will be removed in 4.0." + + "Please use org.apache.kafka.tools.RecordReader instead", "3.5.0") trait MessageReader { def init(inputStream: InputStream, props: Properties): Unit = {} diff --git a/core/src/main/scala/kafka/tools/ConsoleProducer.scala b/core/src/main/scala/kafka/tools/ConsoleProducer.scala index 245212471d064..a11905fdf96ba 100644 --- a/core/src/main/scala/kafka/tools/ConsoleProducer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleProducer.scala @@ -24,33 +24,81 @@ import java.util.regex.Pattern import joptsimple.{OptionException, OptionParser, OptionSet} import kafka.common.MessageReader import kafka.utils.Implicits._ -import kafka.utils.{Exit, ToolsUtils} +import kafka.utils.{Exit, Logging, ToolsUtils} import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback -import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} +import org.apache.kafka.clients.producer.{KafkaProducer, Producer, ProducerConfig, ProducerRecord} import org.apache.kafka.common.KafkaException import org.apache.kafka.common.record.CompressionType import org.apache.kafka.common.utils.Utils import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils} +import org.apache.kafka.tools.RecordReader + +import scala.annotation.nowarn + +@nowarn("cat=deprecation") +object ConsoleProducer extends Logging { + + private[tools] def newReader(className: String, prop: Properties): RecordReader = { + val reader = Class.forName(className).getDeclaredConstructor().newInstance() + reader match { + case r: RecordReader => + r.configure(prop.asInstanceOf[java.util.Map[String, _]]) + r + case r: MessageReader => + logger.warn("MessageReader is deprecated. Please use org.apache.kafka.tools.RecordReader instead") + new RecordReader { + private[this] var initialized = false + + override def readRecords(inputStream: InputStream): java.util.Iterator[ProducerRecord[Array[Byte], Array[Byte]]] = { + if (initialized) throw new IllegalStateException("It is invalid to call readRecords again when the reader is based on deprecated MessageReader") + if (!initialized) { + r.init(inputStream, prop) + initialized = true + } + new java.util.Iterator[ProducerRecord[Array[Byte], Array[Byte]]] { + private[this] var current: ProducerRecord[Array[Byte], Array[Byte]] = _ + // a flag used to avoid accessing readMessage again after it does return null + private[this] var done: Boolean = false + + override def hasNext: Boolean = { + if (current != null) true + else if (done) false + else { + current = r.readMessage() + done = current == null + !done + } + } + + override def next(): ProducerRecord[Array[Byte], Array[Byte]] = + try if (hasNext) current + else throw new NoSuchElementException("no more records from input stream") + finally current = null + } + } + override def close(): Unit = r.close() + } + case _ => throw new IllegalArgumentException(f"the reader must extend ${classOf[RecordReader].getName}") + } + } -object ConsoleProducer { + private[tools] def loopReader(producer: Producer[Array[Byte], Array[Byte]], + reader: RecordReader, + inputStream: InputStream, + sync: Boolean): Unit = { + val iter = reader.readRecords(inputStream) + try while (iter.hasNext) send(producer, iter.next(), sync) finally reader.close() + } def main(args: Array[String]): Unit = { try { - val config = new ProducerConfig(args) - val reader = Class.forName(config.readerClass).getDeclaredConstructor().newInstance().asInstanceOf[MessageReader] - reader.init(System.in, getReaderProps(config)) - - val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps(config)) - - Exit.addShutdownHook("producer-shutdown-hook", producer.close) - - var record: ProducerRecord[Array[Byte], Array[Byte]] = null - do { - record = reader.readMessage() - if (record != null) - send(producer, record, config.sync) - } while (record != null) + val config = new ProducerConfig(args) + val input = System.in + val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps(config)) + try loopReader(producer, newReader(config.readerClass, getReaderProps(config)), input, config.sync) + finally producer.close() + Exit.exit(0) } catch { case e: joptsimple.OptionException => System.err.println(e.getMessage) @@ -59,10 +107,9 @@ object ConsoleProducer { e.printStackTrace Exit.exit(1) } - Exit.exit(0) } - private def send(producer: KafkaProducer[Array[Byte], Array[Byte]], + private def send(producer: Producer[Array[Byte], Array[Byte]], record: ProducerRecord[Array[Byte], Array[Byte]], sync: Boolean): Unit = { if (sync) producer.send(record).get() @@ -292,9 +339,8 @@ object ConsoleProducer { } } - class LineMessageReader extends MessageReader { + class LineMessageReader extends RecordReader { var topic: String = _ - var reader: BufferedReader = _ var parseKey = false var keySeparator = "\t" var parseHeaders = false @@ -307,23 +353,23 @@ object ConsoleProducer { var headersSeparatorPattern: Pattern = _ var nullMarker: String = _ - override def init(inputStream: InputStream, props: Properties): Unit = { - topic = props.getProperty("topic") + override def configure(props: java.util.Map[String, _]): Unit = { + topic = props.get("topic").toString if (props.containsKey("parse.key")) - parseKey = props.getProperty("parse.key").trim.equalsIgnoreCase("true") + parseKey = props.get("parse.key").toString.trim.equalsIgnoreCase("true") if (props.containsKey("key.separator")) - keySeparator = props.getProperty("key.separator") + keySeparator = props.get("key.separator").toString if (props.containsKey("parse.headers")) - parseHeaders = props.getProperty("parse.headers").trim.equalsIgnoreCase("true") + parseHeaders = props.get("parse.headers").toString.trim.equalsIgnoreCase("true") if (props.containsKey("headers.delimiter")) - headersDelimiter = props.getProperty("headers.delimiter") + headersDelimiter = props.get("headers.delimiter").toString if (props.containsKey("headers.separator")) - headersSeparator = props.getProperty("headers.separator") + headersSeparator = props.get("headers.separator").toString headersSeparatorPattern = Pattern.compile(headersSeparator) if (props.containsKey("headers.key.separator")) - headersKeySeparator = props.getProperty("headers.key.separator") + headersKeySeparator = props.get("headers.key.separator").toString if (props.containsKey("ignore.error")) - ignoreError = props.getProperty("ignore.error").trim.equalsIgnoreCase("true") + ignoreError = props.get("ignore.error").toString.trim.equalsIgnoreCase("true") if (headersDelimiter == headersSeparator) throw new KafkaException("headers.delimiter and headers.separator may not be equal") if (headersDelimiter == headersKeySeparator) @@ -331,7 +377,7 @@ object ConsoleProducer { if (headersSeparator == headersKeySeparator) throw new KafkaException("headers.separator and headers.key.separator may not be equal") if (props.containsKey("null.marker")) - nullMarker = props.getProperty("null.marker") + nullMarker = props.get("null.marker").toString if (nullMarker == keySeparator) throw new KafkaException("null.marker and key.separator may not be equal") if (nullMarker == headersSeparator) @@ -340,38 +386,48 @@ object ConsoleProducer { throw new KafkaException("null.marker and headers.delimiter may not be equal") if (nullMarker == headersKeySeparator) throw new KafkaException("null.marker and headers.key.separator may not be equal") - reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)) } - override def readMessage(): ProducerRecord[Array[Byte], Array[Byte]] = { - lineNumber += 1 - if (printPrompt) print(">") - val line = reader.readLine() - line match { - case null => null - case line => - val headers = parse(parseHeaders, line, 0, headersDelimiter, "headers delimiter") - val headerOffset = if (headers == null) 0 else headers.length + headersDelimiter.length - - val key = parse(parseKey, line, headerOffset, keySeparator, "key separator") - val keyOffset = if (key == null) 0 else key.length + keySeparator.length - - val value = line.substring(headerOffset + keyOffset) - - val record = new ProducerRecord[Array[Byte], Array[Byte]]( - topic, - if (key != null && key != nullMarker) key.getBytes(StandardCharsets.UTF_8) else null, - if (value != null && value != nullMarker) value.getBytes(StandardCharsets.UTF_8) else null, - ) - - if (headers != null && headers != nullMarker) { - splitHeaders(headers) - .foreach(header => record.headers.add(header._1, header._2)) + override def readRecords(inputStream: InputStream): java.util.Iterator[ProducerRecord[Array[Byte], Array[Byte]]] = + new java.util.Iterator[ProducerRecord[Array[Byte], Array[Byte]]] { + private[this] val reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)) + private[this] var current: ProducerRecord[Array[Byte], Array[Byte]] = _ + override def hasNext: Boolean = + if (current != null) true + else { + lineNumber += 1 + if (printPrompt) print(">") + val line = reader.readLine() + current = line match { + case null => null + case line => + val headers = parse(parseHeaders, line, 0, headersDelimiter, "headers delimiter") + val headerOffset = if (headers == null) 0 else headers.length + headersDelimiter.length + + val key = parse(parseKey, line, headerOffset, keySeparator, "key separator") + val keyOffset = if (key == null) 0 else key.length + keySeparator.length + + val value = line.substring(headerOffset + keyOffset) + + val record = new ProducerRecord[Array[Byte], Array[Byte]]( + topic, + if (key != null && key != nullMarker) key.getBytes(StandardCharsets.UTF_8) else null, + if (value != null && value != nullMarker) value.getBytes(StandardCharsets.UTF_8) else null, + ) + + if (headers != null && headers != nullMarker) { + splitHeaders(headers) + .foreach(header => record.headers.add(header._1, header._2)) + } + record + } + current != null } - record + override def next(): ProducerRecord[Array[Byte], Array[Byte]] = if (!hasNext) throw new NoSuchElementException("no more record") + else try current finally current = null } - } + private def parse(enabled: Boolean, line: String, startIndex: Int, demarcation: String, demarcationName: String): String = { (enabled, line.indexOf(demarcation, startIndex)) match { diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala index 8a594f92a300a..5e0a941f85555 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala @@ -17,14 +17,21 @@ package kafka.tools +import kafka.common.MessageReader + import java.nio.file.Files import kafka.tools.ConsoleProducer.LineMessageReader import kafka.utils.{Exit, TestUtils} -import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.clients.producer.{Producer, ProducerConfig, ProducerRecord} +import org.apache.kafka.tools.RecordReader import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue} import org.junit.jupiter.api.Test +import org.mockito.Mockito +import java.io.InputStream import java.util +import java.util.Properties +import scala.annotation.nowarn class ConsoleProducerTest { @@ -136,7 +143,7 @@ class ConsoleProducerTest { def testParseKeyProp(): Unit = { val config = new ConsoleProducer.ProducerConfig(brokerListValidArgs) val reader = Class.forName(config.readerClass).getDeclaredConstructor().newInstance().asInstanceOf[LineMessageReader] - reader.init(System.in, ConsoleProducer.getReaderProps(config)) + reader.configure(ConsoleProducer.getReaderProps(config).asInstanceOf[java.util.Map[String, _]]) assertTrue(reader.keySeparator == "#") assertTrue(reader.parseKey) } @@ -158,7 +165,7 @@ class ConsoleProducerTest { ) val config = new ConsoleProducer.ProducerConfig(args) val reader = Class.forName(config.readerClass).getDeclaredConstructor().newInstance().asInstanceOf[LineMessageReader] - reader.init(System.in, ConsoleProducer.getReaderProps(config)) + reader.configure(ConsoleProducer.getReaderProps(config).asInstanceOf[java.util.Map[String, _]]) assertEquals(";", reader.keySeparator) assertTrue(reader.parseKey) assertTrue(reader.parseHeaders) @@ -220,4 +227,61 @@ class ConsoleProducerTest { producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG)) } + @Test + def testNewReader(): Unit = { + ConsoleProducerTest.configureCount = 0 + ConsoleProducerTest.closeCount = 0 + val reader = ConsoleProducer.newReader(classOf[ConsoleProducerTest.TestMessageReader].getName, new Properties()) + // the deprecated MessageReader get configured when creating records + assertEquals(0, ConsoleProducerTest.configureCount) + reader.readRecords(System.in) + assertEquals(1, ConsoleProducerTest.configureCount) + assertEquals(0, ConsoleProducerTest.closeCount) + assertThrows(classOf[IllegalStateException], () => reader.readRecords(System.in)) + reader.close() + assertEquals(1, ConsoleProducerTest.closeCount) + + ConsoleProducerTest.configureCount = 0 + ConsoleProducerTest.closeCount = 0 + + val reader1 = ConsoleProducer.newReader(classOf[ConsoleProducerTest.TestRecordReader].getName, new Properties()) + assertEquals(1, ConsoleProducerTest.configureCount) + assertEquals(0, ConsoleProducerTest.closeCount) + reader1.close() + assertEquals(1, ConsoleProducerTest.closeCount) + } + + @Test + def testLoopReader(): Unit = { + ConsoleProducerTest.configureCount = 0 + ConsoleProducerTest.closeCount = 0 + val reader = ConsoleProducer.newReader(classOf[ConsoleProducerTest.TestRecordReader].getName, new Properties()) + + ConsoleProducer.loopReader(Mockito.mock(classOf[Producer[Array[Byte], Array[Byte]]]), + reader, System.in, false) + + assertEquals(1, ConsoleProducerTest.configureCount) + assertEquals(1, ConsoleProducerTest.closeCount) + } +} + +@nowarn("cat=deprecation") +object ConsoleProducerTest { + var configureCount = 0 + var closeCount = 0 + class TestMessageReader extends MessageReader { + override def init(inputStream: InputStream, props: Properties): Unit = configureCount += 1 + override def readMessage(): ProducerRecord[Array[Byte], Array[Byte]] = null + + override def close(): Unit = closeCount += 1 + + } + + class TestRecordReader extends RecordReader { + override def configure(configs: util.Map[String, _]): Unit = configureCount += 1 + override def readRecords(inputStream: InputStream): java.util.Iterator[ProducerRecord[Array[Byte], Array[Byte]]] = + java.util.Collections.emptyIterator() + + override def close(): Unit = closeCount += 1 + } } diff --git a/core/src/test/scala/unit/kafka/tools/LineMessageReaderTest.scala b/core/src/test/scala/unit/kafka/tools/LineMessageReaderTest.scala index 0582abc8a13c1..740f42762de1f 100644 --- a/core/src/test/scala/unit/kafka/tools/LineMessageReaderTest.scala +++ b/core/src/test/scala/unit/kafka/tools/LineMessageReaderTest.scala @@ -20,7 +20,7 @@ package kafka.tools import kafka.tools.ConsoleProducer.LineMessageReader import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.common.KafkaException -import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows} import org.junit.jupiter.api.Test import java.io.ByteArrayInputStream @@ -124,10 +124,11 @@ class LineMessageReaderTest { "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), defaultTestProps) - lineReader.readMessage() + lineReader.configure(defaultTestProps.asInstanceOf[java.util.Map[String, _]]) + val iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes)) + iter.next() - val expectedException = assertThrows(classOf[KafkaException], () => lineReader.readMessage()) + val expectedException = assertThrows(classOf[KafkaException], () => iter.next()) assertEquals( "No key separator found on line number 2: 'headerKey1.0:headerValue1.0\tkey1[MISSING-DELIMITER]value1'", @@ -139,9 +140,10 @@ class LineMessageReaderTest { def testMissingHeaderKeySeparator(): Unit = { val lineReader = new LineMessageReader() val input = "key[MISSING-DELIMITER]val\tkey0\tvalue0\n" - lineReader.init(new ByteArrayInputStream(input.getBytes), defaultTestProps) + lineReader.configure(defaultTestProps.asInstanceOf[java.util.Map[String, _]]) + val iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes)) - val expectedException = assertThrows(classOf[KafkaException], () => lineReader.readMessage()) + val expectedException = assertThrows(classOf[KafkaException], () => iter.next()) assertEquals( "No header key separator found in pair 'key[MISSING-DELIMITER]val' on line number 1", @@ -166,7 +168,7 @@ class LineMessageReaderTest { } private def assertThrowsOnInvalidPatternConfig(props: Properties, expectedMessage: String): Unit = { - val exception = assertThrows(classOf[KafkaException], () => new LineMessageReader().init(null, props)) + val exception = assertThrows(classOf[KafkaException], () => new LineMessageReader().configure(props.asInstanceOf[java.util.Map[String, _]])) assertEquals( expectedMessage, exception.getMessage @@ -273,17 +275,18 @@ class LineMessageReaderTest { // If the null marker is not set val lineReader = new LineMessageReader() props.remove("null.marker") - lineReader.init(new ByteArrayInputStream(input.getBytes), props) - assertRecordEquals(record("", "value", List("h0" -> "v0", header)), lineReader.readMessage()) + lineReader.configure(props.asInstanceOf[java.util.Map[String, _]]) + val iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes)) + assertRecordEquals(record("", "value", List("h0" -> "v0", header)), iter.next()) // line 2 is not valid anymore - val expectedException = assertThrows(classOf[KafkaException], () => lineReader.readMessage()) + val expectedException = assertThrows(classOf[KafkaException], () => iter.next()) assertEquals( "No header key separator found in pair '' on line number 2", expectedException.getMessage ) - assertRecordEquals(record("", "", List("h0" -> "", header)), lineReader.readMessage()) - assertRecordEquals(record("key", "", List("h0" -> "", header)), lineReader.readMessage()) - assertRecordEquals(record("key", "", List("h0" -> "", "h1" -> "value")), lineReader.readMessage()) + assertRecordEquals(record("", "", List("h0" -> "", header)), iter.next()) + assertRecordEquals(record("key", "", List("h0" -> "", header)), iter.next()) + assertRecordEquals(record("key", "", List("h0" -> "", "h1" -> "value")), iter.next()) } @Test @@ -293,8 +296,9 @@ class LineMessageReaderTest { val props = defaultTestProps props.put("null.marker", "") val lineReader = new LineMessageReader() - lineReader.init(new ByteArrayInputStream(input.getBytes), props) - val expectedException = assertThrows(classOf[KafkaException], () => lineReader.readMessage()) + lineReader.configure(props.asInstanceOf[java.util.Map[String, _]]) + val iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes)) + val expectedException = assertThrows(classOf[KafkaException], () => iter.next()) assertEquals( "Header keys should not be equal to the null marker '' as they can't be null", expectedException.getMessage @@ -324,8 +328,11 @@ class LineMessageReaderTest { def runTest(props: Properties, input: String, expectedRecords: ProducerRecord[String, String]*): Unit = { val lineReader = new LineMessageReader - lineReader.init(new ByteArrayInputStream(input.getBytes), props) - expectedRecords.foreach(r => assertRecordEquals(r, lineReader.readMessage())) + lineReader.configure(props.asInstanceOf[java.util.Map[String, _]]) + val iter = lineReader.readRecords(new ByteArrayInputStream(input.getBytes)) + expectedRecords.foreach(r => assertRecordEquals(r, iter.next())) + assertFalse(iter.hasNext) + assertThrows(classOf[NoSuchElementException], () => iter.next()) } // The equality method of ProducerRecord compares memory references for the header iterator, this is why this custom equality check is used.