[GitHub] [kafka] florin-akermann commented on a change in pull request #11456: KAFKA-13351: Add possibility to write kafka headers in Kafka Console Producer

[GitBox <gi...@apache.org>]

florin-akermann commented on a change in pull request #11456:
URL: https://github.com/apache/kafka/pull/11456#discussion_r762368709



##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -274,28 +296,74 @@ object ConsoleProducer {
         parseKey = props.getProperty("parse.key").trim.equalsIgnoreCase("true")
       if (props.containsKey("key.separator"))
         keySeparator = props.getProperty("key.separator")
+      if (props.containsKey("parse.headers"))
+        parseHeader = 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")
+      if (props.containsKey("headers.key.separator"))
+        headerKeySeparator = props.getProperty("headers.key.separator")
       if (props.containsKey("ignore.error"))
         ignoreError = props.getProperty("ignore.error").trim.equalsIgnoreCase("true")
       reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))
     }
 
     override def readMessage() = {
       lineNumber += 1
-      if (printPrompt)
-        print(">")
-      (reader.readLine(), parseKey) match {
-        case (null, _) => null
-        case (line, true) =>
-          line.indexOf(keySeparator) match {
-            case -1 =>
-              if (ignoreError) new ProducerRecord(topic, line.getBytes(StandardCharsets.UTF_8))
-              else throw new KafkaException(s"No key found on line $lineNumber: $line")
-            case n =>
-              val value = (if (n + keySeparator.size > line.size) "" else line.substring(n + keySeparator.size)).getBytes(StandardCharsets.UTF_8)
-              new ProducerRecord(topic, line.substring(0, n).getBytes(StandardCharsets.UTF_8), value)
-          }
-        case (line, false) =>
-          new ProducerRecord(topic, line.getBytes(StandardCharsets.UTF_8))
+      if (printPrompt) print(">")
+      val line = reader.readLine()
+      try{
+        (line, parseKey, parseHeader) match {
+          case (null, _, _) => null
+          case (line, true, true) =>
+            val Array(headers, key, value) = line.split("[" + headersDelimiter + keySeparator + "]")

Review comment:
       Good point. I pushed an alternative based on your suggestion.
   For the mapHeaders method I compile two regex patterns on init as the length of the headers varies.
   
   What are your thoughs on compiling the patterns for the split between headers keys and values instead of using `indexOf ` and `substring`? In other words, applying the same approach as for `mapHeaders`.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org