Hangleton commented on code in PR #13511:
URL: https://github.com/apache/kafka/pull/13511#discussion_r1165769118
##########
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
Review Comment:
Should we log or print it to notify the user of the presence of the key?
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -167,25 +177,43 @@ object TransactionLog {
* Exposed for printing records using [[kafka.tools.DumpLogSegments]]
*/
def formatRecordKeyAndValue(record: Record): (Option[String],
Option[String]) = {
- val txnKey = TransactionLog.readTxnRecordKey(record.key)
- val keyString =
s"transaction_metadata::transactionalId=${txnKey.transactionalId}"
-
- val valueString =
TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value) match {
- case None => "<DELETE>"
-
- case Some(txnMetadata) => s"producerId:${txnMetadata.producerId}," +
- s"producerEpoch:${txnMetadata.producerEpoch}," +
- s"state=${txnMetadata.state}," +
- s"partitions=${txnMetadata.topicPartitions.mkString("[", ",", "]")}," +
- s"txnLastUpdateTimestamp=${txnMetadata.txnLastUpdateTimestamp}," +
- s"txnTimeoutMs=${txnMetadata.txnTimeoutMs}"
- }
+ TransactionLog.readTxnRecordKey(record.key) match {
+ case txnKey: TxnKey =>
+ val keyString =
s"transaction_metadata::transactionalId=${txnKey.transactionalId}"
+
+ val valueString =
TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value) match {
+ case None => "<DELETE>"
+
+ case Some(txnMetadata) => s"producerId:${txnMetadata.producerId}," +
+ s"producerEpoch:${txnMetadata.producerEpoch}," +
+ s"state=${txnMetadata.state}," +
+ s"partitions=${txnMetadata.topicPartitions.mkString("[", ",",
"]")}," +
+ s"txnLastUpdateTimestamp=${txnMetadata.txnLastUpdateTimestamp}," +
+ s"txnTimeoutMs=${txnMetadata.txnTimeoutMs}"
+ }
+
+ (Some(keyString), Some(valueString))
+
+ case _: UnknownKey =>
+ (Some("<UNKNOWN>"), Some("<UNKNOWN>"))
- (Some(keyString), Some(valueString))
+ case unexpectedKey =>
+ throw new IllegalStateException(s"Found unexpected key $unexpectedKey
while formatting transaction log.")
+ }
}
}
-case class TxnKey(version: Short, transactionalId: String) {
+trait BaseKey{
+ def version: Short
+ def transactionalId: String
+}
+
+case class TxnKey(version: Short, transactionalId: String) extends BaseKey {
override def toString: String = transactionalId
}
+
+case class UnknownKey(version: Short, transactionalId: String = null) extends
BaseKey {
Review Comment:
Is the `transactionalId` going to be populated?
##########
core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala:
##########
@@ -1368,3 +1378,8 @@ case class GroupMetadataKey(version: Short, key: String)
extends BaseKey {
override def toString: String = key
}
+case class UnknownKey(version: Short, key: String = null) extends BaseKey {
Review Comment:
Is the `key` going to be populated?
--
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
