dajac commented on code in PR #13511:
URL: https://github.com/apache/kafka/pull/13511#discussion_r1165986528
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -148,17 +152,23 @@ object TransactionLog {
// Formatter for use with tools to read transaction log messages
class TransactionLogMessageFormatter extends MessageFormatter {
def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]],
output: PrintStream): Unit = {
- Option(consumerRecord.key).map(key =>
readTxnRecordKey(ByteBuffer.wrap(key))).foreach { txnKey =>
- val transactionalId = txnKey.transactionalId
- val value = consumerRecord.value
- val producerIdMetadata = if (value == null)
- None
- else
- readTxnRecordValue(transactionalId, ByteBuffer.wrap(value))
- output.write(transactionalId.getBytes(StandardCharsets.UTF_8))
- output.write("::".getBytes(StandardCharsets.UTF_8))
-
output.write(producerIdMetadata.getOrElse("NULL").toString.getBytes(StandardCharsets.UTF_8))
- output.write("\n".getBytes(StandardCharsets.UTF_8))
+ Option(consumerRecord.key).map(key =>
readTxnRecordKey(ByteBuffer.wrap(key))).foreach {
+ case txnKey: TxnKey =>
+ val transactionalId = txnKey.transactionalId
+ val value = consumerRecord.value
+ val producerIdMetadata = if (value == null)
+ None
+ else
+ readTxnRecordValue(transactionalId, ByteBuffer.wrap(value))
+ output.write(transactionalId.getBytes(StandardCharsets.UTF_8))
+ output.write("::".getBytes(StandardCharsets.UTF_8))
+
output.write(producerIdMetadata.getOrElse("NULL").toString.getBytes(StandardCharsets.UTF_8))
+ output.write("\n".getBytes(StandardCharsets.UTF_8))
+
+ case _: UnknownKey => // Only print if this message is a transaction
record
+
+ case unexpectedKey =>
+ throw new IllegalStateException(s"Found unexpected key
$unexpectedKey while reading transaction log.")
Review Comment:
Does it help if we use a « sealed trait »?
--
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
