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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
