diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala index d246e7b9007b3..c40ae1019304c 100755 --- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala @@ -223,16 +223,22 @@ object ConsoleConsumer extends Logging { .ofType(classOf[String]) .defaultsTo(classOf[DefaultMessageFormatter].getName) val messageFormatterArgOpt = parser.accepts("property", - "The properties to initialize the message formatter. Default properties include:\n" + - "\tprint.timestamp=true|false\n" + - "\tprint.key=true|false\n" + - "\tprint.value=true|false\n" + - "\tkey.separator=\n" + - "\tline.separator=\n" + - "\tkey.deserializer=\n" + - "\tvalue.deserializer=\n" + - "\nUsers can also pass in customized properties for their formatter; more specifically, users " + - "can pass in properties keyed with \'key.deserializer.\' and \'value.deserializer.\' prefixes to configure their deserializers.") + """The properties to initialize the message formatter. Default properties include: + | print.timestamp=true|false + | print.key=true|false + | print.offset=true|false + | print.partition=true|false + | print.headers=true|false + | print.value=true|false + | key.separator= + | line.separator= + | headers.separator= + | key.deserializer= + | value.deserializer= + | header.deserializer= + | + |Users can also pass in customized properties for their formatter; more specifically, users can pass in properties keyed with 'key.deserializer.', 'value.deserializer.' and 'headers.deserializer.' prefixes to configure their deserializers.""" + .stripMargin) .withRequiredArg .describedAs("prop") .ofType(classOf[String]) @@ -457,47 +463,34 @@ object ConsoleConsumer extends Logging { } class DefaultMessageFormatter extends MessageFormatter { + var printTimestamp = false var printKey = false + var printOffset = false + var printPartition = false + var printHeaders = false var printValue = true - var printTimestamp = false - var keySeparator = "\t".getBytes(StandardCharsets.UTF_8) - var lineSeparator = "\n".getBytes(StandardCharsets.UTF_8) + var keySeparator = utfBytes("\t") + var headersSeparator = utfBytes(",") + var lineSeparator = utfBytes("\n") var keyDeserializer: Option[Deserializer[_]] = None var valueDeserializer: Option[Deserializer[_]] = None + var headersDeserializer: Option[Deserializer[_]] = None override def init(props: Properties) { - if (props.containsKey("print.timestamp")) - printTimestamp = props.getProperty("print.timestamp").trim.equalsIgnoreCase("true") - if (props.containsKey("print.key")) - printKey = props.getProperty("print.key").trim.equalsIgnoreCase("true") - if (props.containsKey("print.value")) - printValue = props.getProperty("print.value").trim.equalsIgnoreCase("true") - if (props.containsKey("key.separator")) - keySeparator = props.getProperty("key.separator").getBytes(StandardCharsets.UTF_8) - if (props.containsKey("line.separator")) - lineSeparator = props.getProperty("line.separator").getBytes(StandardCharsets.UTF_8) - // Note that `toString` will be called on the instance returned by `Deserializer.deserialize` - if (props.containsKey("key.deserializer")) { - keyDeserializer = Some(Class.forName(props.getProperty("key.deserializer")).getDeclaredConstructor() - .newInstance().asInstanceOf[Deserializer[_]]) - keyDeserializer.get.configure(propertiesWithKeyPrefixStripped("key.deserializer.", props).asScala.asJava, true) - } - // Note that `toString` will be called on the instance returned by `Deserializer.deserialize` - if (props.containsKey("value.deserializer")) { - valueDeserializer = Some(Class.forName(props.getProperty("value.deserializer")).getDeclaredConstructor() - .newInstance().asInstanceOf[Deserializer[_]]) - valueDeserializer.get.configure(propertiesWithKeyPrefixStripped("value.deserializer.", props).asScala.asJava, false) - } - } - - private def propertiesWithKeyPrefixStripped(prefix: String, props: Properties): Properties = { - val newProps = new Properties() - props.asScala.foreach { case (key, value) => - if (key.startsWith(prefix) && key.length > prefix.length) - newProps.put(key.substring(prefix.length), value) - } - newProps + getPropertyIfExists(props, "print.timestamp", getBoolProperty).foreach(printTimestamp = _) + getPropertyIfExists(props, "print.key", getBoolProperty).foreach(printKey = _) + getPropertyIfExists(props, "print.offset", getBoolProperty).foreach(printOffset = _) + getPropertyIfExists(props, "print.partition", getBoolProperty).foreach(printPartition = _) + getPropertyIfExists(props, "print.headers", getBoolProperty).foreach(printHeaders = _) + getPropertyIfExists(props, "print.value", getBoolProperty).foreach(printValue = _) + getPropertyIfExists(props, "key.separator", getByteProperty).foreach(keySeparator = _) + getPropertyIfExists(props, "line.separator", getByteProperty).foreach(lineSeparator = _) + getPropertyIfExists(props, "headers.separator", getByteProperty).foreach(headersSeparator = _) + + keyDeserializer = getPropertyIfExists(props, "key.deserializer", getDeserializerProperty(true)) + valueDeserializer = getPropertyIfExists(props, "value.deserializer", getDeserializerProperty(false)) + headersDeserializer = getPropertyIfExists(props, "headers.deserializer", getDeserializerProperty(false)) } def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]], output: PrintStream) { @@ -510,25 +503,48 @@ class DefaultMessageFormatter extends MessageFormatter { } def write(deserializer: Option[Deserializer[_]], sourceBytes: Array[Byte], topic: String) { - val nonNullBytes = Option(sourceBytes).getOrElse("null".getBytes(StandardCharsets.UTF_8)) - val convertedBytes = deserializer.map(_.deserialize(topic, nonNullBytes).toString. - getBytes(StandardCharsets.UTF_8)).getOrElse(nonNullBytes) - output.write(convertedBytes) + output.write(deserialize(deserializer, sourceBytes, topic)) } import consumerRecord._ if (printTimestamp) { if (timestampType != TimestampType.NO_TIMESTAMP_TYPE) - output.write(s"$timestampType:$timestamp".getBytes(StandardCharsets.UTF_8)) + output.write(utfBytes(s"$timestampType:$timestamp")) else - output.write(s"NO_TIMESTAMP".getBytes(StandardCharsets.UTF_8)) - writeSeparator(printKey || printValue) + output.write(utfBytes("NO_TIMESTAMP")) + writeSeparator(columnSeparator = printKey || printOffset || printPartition || printHeaders || printValue) } if (printKey) { write(keyDeserializer, key, topic) - writeSeparator(printValue) + writeSeparator(columnSeparator = printOffset || printPartition || printHeaders || printValue) + } + + if (printOffset) { + output.write(utfBytes(offset().toString)) + writeSeparator(columnSeparator = printPartition || printHeaders || printValue) + } + + if (printPartition) { + output.write(utfBytes(partition().toString)) + writeSeparator(columnSeparator = printHeaders || printValue) + } + + if (printHeaders) { + val headersIt = headers().iterator.asScala + if(headersIt.hasNext) { + headersIt.foreach { header => + output.write(utfBytes(header.key() + ":")) + output.write(deserialize(headersDeserializer, header.value(), topic)) + if(headersIt.hasNext) { + output.write(headersSeparator) + } + } + } else { + output.write(utfBytes("NO_HEADERS")) + } + writeSeparator(columnSeparator = printValue) } if (printValue) { @@ -536,6 +552,49 @@ class DefaultMessageFormatter extends MessageFormatter { output.write(lineSeparator) } } + + private def propertiesWithKeyPrefixStripped(prefix: String, props: Properties): Properties = { + val newProps = new Properties() + props.asScala.foreach { case (key, value) => + if (key.startsWith(prefix) && key.length > prefix.length) + newProps.put(key.substring(prefix.length), value) + } + newProps + } + + private def deserialize(deserializer: Option[Deserializer[_]], sourceBytes: Array[Byte], topic: String) = { + val nonNullBytes = Option(sourceBytes).getOrElse(utfBytes("null")) + val convertedBytes = deserializer + .map(d => utfBytes(d.deserialize(topic, nonNullBytes).toString)) + .getOrElse(nonNullBytes) + convertedBytes + } + + private def utfBytes(str: String) = str.getBytes(StandardCharsets.UTF_8) + + private def getByteProperty(props: Properties, key: String): Array[Byte] = { + utfBytes(props.getProperty(key)) + } + + private def getBoolProperty(props: Properties, key: String): Boolean = { + props.getProperty(key).trim.equalsIgnoreCase("true") + } + + private def getDeserializerProperty(isKey: Boolean)(props: Properties, propertyName: String): Deserializer[_] = { + val deserializer = Class.forName(props.getProperty(propertyName)).newInstance().asInstanceOf[Deserializer[_]] + val deserializerConfig = propertiesWithKeyPrefixStripped(propertyName + ".", props) + .asScala + .asJava + deserializer.configure(deserializerConfig, isKey) + deserializer + } + + private def getPropertyIfExists[T](props: Properties, key: String, getter: (Properties, String) => T): Option[T] = { + if (props.containsKey(key)) + Some(getter(props, key)) + else + None + } } class LoggingMessageFormatter extends MessageFormatter with LazyLogging { diff --git a/core/src/test/scala/unit/kafka/tools/DefaultMessageFormatterTest.scala b/core/src/test/scala/unit/kafka/tools/DefaultMessageFormatterTest.scala new file mode 100644 index 0000000000000..47283b9cfdaac --- /dev/null +++ b/core/src/test/scala/unit/kafka/tools/DefaultMessageFormatterTest.scala @@ -0,0 +1,226 @@ +/** + * 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 unit.kafka.tools + +import java.io.{ByteArrayOutputStream, Closeable, PrintStream} +import java.nio.charset.StandardCharsets +import java.util +import java.util.Properties + +import kafka.tools.DefaultMessageFormatter +import org.apache.kafka.clients.consumer.ConsumerRecord +import org.apache.kafka.common.header.Header +import org.apache.kafka.common.header.internals.{RecordHeader, RecordHeaders} +import org.apache.kafka.common.record.TimestampType +import org.apache.kafka.common.serialization.Deserializer +import org.junit.Assert._ +import org.junit.Test +import org.junit.runner.RunWith +import org.junit.runners.Parameterized +import org.junit.runners.Parameterized.Parameters + +import scala.collection.JavaConverters._ + +@RunWith(value = classOf[Parameterized]) +class DefaultMessageFormatterTest(name: String, record: ConsumerRecord[Array[Byte], Array[Byte]], properties: Map[String, String], expected: String) { + import DefaultMessageFormatterTest._ + + @Test + def testWriteRecord() { + withResource(new ByteArrayOutputStream()) { baos => + withResource(new PrintStream(baos)) { ps => + val formatter = buildFormatter(properties) + formatter.writeTo(record, ps) + val actual = new String(baos.toByteArray(), StandardCharsets.UTF_8) + assertEquals(expected, actual) + + } + } + } +} + +object DefaultMessageFormatterTest { + @Parameters(name = "Test {index} - {0}") + def parameters: java.util.Collection[Array[Object]] = { + Seq( + Array( + "print nothing", + consumerRecord(), + Map("print.value" -> "false"), + ""), + Array( + "print key", + consumerRecord(), + Map("print.key" -> "true", "print.value" -> "false"), + "someKey\n"), + Array( + "print value", + consumerRecord(), + Map(), + "someValue\n"), + Array( + "print empty timestamp", + consumerRecord(timestampType = TimestampType.NO_TIMESTAMP_TYPE), + Map("print.timestamp" -> "true", "print.value" -> "false"), + "NO_TIMESTAMP\n"), + Array( + "print log append time timestamp", + consumerRecord(timestampType = TimestampType.LOG_APPEND_TIME), + Map("print.timestamp" -> "true", "print.value" -> "false"), + "LogAppendTime:1234\n"), + Array( + "print create time timestamp", + consumerRecord(timestampType = TimestampType.CREATE_TIME), + Map("print.timestamp" -> "true", "print.value" -> "false"), + "CreateTime:1234\n"), + Array( + "print partition", + consumerRecord(), + Map("print.partition" -> "true", "print.value" -> "false"), + "9\n"), + Array( + "print offset", + consumerRecord(), + Map("print.offset" -> "true", "print.value" -> "false"), + "9876\n"), + Array( + "print headers", + consumerRecord(), + Map("print.headers" -> "true", "print.value" -> "false"), + "h1:v1,h2:v2\n"), + Array( + "print empty headers", + consumerRecord(headers = Nil), + Map("print.headers" -> "true", "print.value" -> "false"), + "NO_HEADERS\n"), + Array( + "print all possible fields with default delimiters", + consumerRecord(), + Map("print.key" -> "true", + "print.timestamp" -> "true", + "print.partition" -> "true", + "print.offset" -> "true", + "print.headers" -> "true", + "print.value" -> "true"), + "CreateTime:1234\tsomeKey\t9876\t9\th1:v1,h2:v2\tsomeValue\n"), + Array( + "print all possible fields with custom delimiters", + consumerRecord(), + Map( + "key.separator" -> "|", + "line.separator" -> "^", + "headers.separator" -> "#", + "print.key" -> "true", + "print.timestamp" -> "true", + "print.partition" -> "true", + "print.offset" -> "true", + "print.headers" -> "true", + "print.value" -> "true"), + "CreateTime:1234|someKey|9876|9|h1:v1#h2:v2|someValue^"), + Array( + "print key with custom deserializer", + consumerRecord(), + Map( + "print.key" -> "true", + "print.headers" -> "true", + "print.value" -> "true", + "key.deserializer" -> "unit.kafka.tools.UpperCaseDeserializer"), + "SOMEKEY\th1:v1,h2:v2\tsomeValue\n"), + Array( + "print value with custom deserializer", + consumerRecord(), + Map( + "print.key" -> "true", + "print.headers" -> "true", + "print.value" -> "true", + "value.deserializer" -> "unit.kafka.tools.UpperCaseDeserializer"), + "someKey\th1:v1,h2:v2\tSOMEVALUE\n"), + Array( + "print headers with custom deserializer", + consumerRecord(), + Map( + "print.key" -> "true", + "print.headers" -> "true", + "print.value" -> "true", + "headers.deserializer" -> "unit.kafka.tools.UpperCaseDeserializer"), + "someKey\th1:V1,h2:V2\tsomeValue\n"), + Array( + "print key and value", + consumerRecord(), + Map("print.key" -> "true", "print.value" -> "true"), + "someKey\tsomeValue\n"), + Array( + "print fields in the beginning, middle and the end", + consumerRecord(), + Map("print.key" -> "true", "print.value" -> "true", "print.partition" -> "true"), + "someKey\t9\tsomeValue\n") + ).asJava + } + + private def buildFormatter(propsToSet: Map[String, String]): DefaultMessageFormatter = { + val properties = new Properties() + //putAll doesn't work on java 9 - https://github.com/scala/bug/issues/10418 + propsToSet.foreach { case (k, v) => + properties.setProperty(k, v) + } + val formatter = new DefaultMessageFormatter() + formatter.init(properties) + formatter + } + + + private def header(key: String, value: String) = { + new RecordHeader(key, value.getBytes(StandardCharsets.UTF_8)) + } + + private def consumerRecord(key: String = "someKey", + value: String = "someValue", + headers: Iterable[Header] = Seq(header("h1", "v1"), header("h2", "v2")), + partition: Int = 9, + offset: Long = 9876, + timestamp: Long = 1234, + timestampType: TimestampType = TimestampType.CREATE_TIME) = { + new ConsumerRecord[Array[Byte], Array[Byte]]( + "someTopic", + partition, + offset, + timestamp, + timestampType, + 0L, + 0, + 0, + key.getBytes(StandardCharsets.UTF_8), + value.getBytes(StandardCharsets.UTF_8), + new RecordHeaders(headers.asJava)) + + } + + private def withResource[Resource <: Closeable, Result](resource: Resource)(handler: Resource => Result): Result = { + try { + handler(resource) + } finally { + resource.close() + } + } +} + +class UpperCaseDeserializer extends Deserializer[String] { + override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {} + override def deserialize(topic: String, data: Array[Byte]): String = new String(data, StandardCharsets.UTF_8).toUpperCase + override def close(): Unit = {} +} \ No newline at end of file