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 number
of the key value pairs in the header varies.
--
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
