KAFKA-14869: Ignore unknown record types for coordinators (KIP-915, Part-1)

clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java

@@ -231,4 +231,38 @@ public static byte[] toVersionPrefixedBytes(final short version, final Message m
                 buffer.limit() == buffer.array().length) return buffer.array();
         else return Utils.toArray(buffer);
     }
+
+    // Should only be used for testing
+    public static byte[] messageWithUnknownVersion() {
+        return MessageUtil.toVersionPrefixedBytes(Short.MAX_VALUE, new Message() {
+            @Override
+            public short lowestSupportedVersion() {
+                return Short.MAX_VALUE;
+            }
+
+            @Override
+            public short highestSupportedVersion() {
+                return Short.MAX_VALUE;
+            }
+
+            @Override
+            public void addSize(MessageSizeAccumulator size, ObjectSerializationCache cache, short version) {}
+
+            @Override
+            public void write(Writable writable, ObjectSerializationCache cache, short version) {}
+
+            @Override
+            public void read(Readable readable, short version) {}
+
+            @Override
+            public List<RawTaggedField> unknownTaggedFields() {
+                return null;
+            }
+
+            @Override
+            public Message duplicate() {
+                return null;
+            }
+        });
+    }
 }

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -576,7 +576,10 @@ class GroupMetadataManager(brokerId: Int,
     }
   }
 
-  private def doLoadGroupsAndOffsets(topicPartition: TopicPartition, onGroupLoaded: GroupMetadata => Unit): Unit = {
+  // Visible for testing
+  private[group] def doLoadGroupsAndOffsets(topicPartition: TopicPartition,
+                                            onGroupLoaded: GroupMetadata => Unit): Unit = {
+
     def logEndOffset: Long = replicaManager.getLogEndOffset(topicPartition).getOrElse(-1L)
 
     replicaManager.getLog(topicPartition) match {
@@ -651,40 +654,42 @@ class GroupMetadataManager(brokerId: Int,
                 if (batchBaseOffset.isEmpty)
                   batchBaseOffset = Some(record.offset)
                 GroupMetadataManager.readMessageKey(record.key) match {
+                  case Some(key) => key match {
+                    case offsetKey: OffsetKey =>
+                      if (isTxnOffsetCommit && !pendingOffsets.contains(batch.producerId))
+                        pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]())
+
+                      // load offset
+                      val groupTopicPartition = offsetKey.key
+                      if (!record.hasValue) {
+                        if (isTxnOffsetCommit)
+                          pendingOffsets(batch.producerId).remove(groupTopicPartition)
+                        else
+                          loadedOffsets.remove(groupTopicPartition)
+                      } else {
+                        val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(record.value)
+                        if (isTxnOffsetCommit)
+                          pendingOffsets(batch.producerId).put(groupTopicPartition, CommitRecordMetadataAndOffset(batchBaseOffset, offsetAndMetadata))
+                        else
+                          loadedOffsets.put(groupTopicPartition, CommitRecordMetadataAndOffset(batchBaseOffset, offsetAndMetadata))
+                      }
+
+                    case groupMetadataKey: GroupMetadataKey =>
+                      // load group metadata
+                      val groupId = groupMetadataKey.key
+                      val groupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, record.value, time)
+                      if (groupMetadata != null) {
+                        removedGroups.remove(groupId)
+                        loadedGroups.put(groupId, groupMetadata)
+                      } else {
+                        loadedGroups.remove(groupId)
+                        removedGroups.add(groupId)
+                      }
+
+                    case _ => // do nothing
+                  }
 
-                  case offsetKey: OffsetKey =>
-                    if (isTxnOffsetCommit && !pendingOffsets.contains(batch.producerId))
-                      pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition, CommitRecordMetadataAndOffset]())
-
-                    // load offset
-                    val groupTopicPartition = offsetKey.key
-                    if (!record.hasValue) {
-                      if (isTxnOffsetCommit)
-                        pendingOffsets(batch.producerId).remove(groupTopicPartition)
-                      else
-                        loadedOffsets.remove(groupTopicPartition)
-                    } else {
-                      val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(record.value)
-                      if (isTxnOffsetCommit)
-                        pendingOffsets(batch.producerId).put(groupTopicPartition, CommitRecordMetadataAndOffset(batchBaseOffset, offsetAndMetadata))
-                      else
-                        loadedOffsets.put(groupTopicPartition, CommitRecordMetadataAndOffset(batchBaseOffset, offsetAndMetadata))
-                    }
-
-                  case groupMetadataKey: GroupMetadataKey =>
-                    // load group metadata
-                    val groupId = groupMetadataKey.key
-                    val groupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, record.value, time)
-                    if (groupMetadata != null) {
-                      removedGroups.remove(groupId)
-                      loadedGroups.put(groupId, groupMetadata)
-                    } else {
-                      loadedGroups.remove(groupId)
-                      removedGroups.add(groupId)
-                    }
-
-                  case unknownKey =>
-                    throw new IllegalStateException(s"Unexpected message key $unknownKey while loading offsets and group metadata")
+                  case None => // ignore unknown keys
                 }
               }
             }
@@ -1041,7 +1046,7 @@ class GroupMetadataManager(brokerId: Int,
  * key version 2:       group metadata
  *    -> value version 0:       [protocol_type, generation, protocol, leader, members]
  */
-object GroupMetadataManager {
+object GroupMetadataManager extends Logging {
   // Metrics names
   val MetricsGroup: String = "group-coordinator-metrics"
   val LoadTimeSensor: String = "GroupPartitionLoadTime"
@@ -1145,17 +1150,22 @@ object GroupMetadataManager {
    * @param buffer input byte-buffer
    * @return an OffsetKey or GroupMetadataKey object from the message
    */
-  def readMessageKey(buffer: ByteBuffer): BaseKey = {
+  def readMessageKey(buffer: ByteBuffer): Option[BaseKey] = {
     val version = buffer.getShort
     if (version >= OffsetCommitKey.LOWEST_SUPPORTED_VERSION && version <= OffsetCommitKey.HIGHEST_SUPPORTED_VERSION) {
       // version 0 and 1 refer to offset
       val key = new OffsetCommitKey(new ByteBufferAccessor(buffer), version)
-      OffsetKey(version, GroupTopicPartition(key.group, new TopicPartition(key.topic, key.partition)))
+      Some(OffsetKey(version, GroupTopicPartition(key.group, new TopicPartition(key.topic, key.partition))))
     } else if (version >= GroupMetadataKeyData.LOWEST_SUPPORTED_VERSION && version <= GroupMetadataKeyData.HIGHEST_SUPPORTED_VERSION) {
       // version 2 refers to group metadata
       val key = new GroupMetadataKeyData(new ByteBufferAccessor(buffer), version)
-      GroupMetadataKey(version, key.group)
-    } else throw new IllegalStateException(s"Unknown group metadata message version: $version")
+      Some(GroupMetadataKey(version, key.group))
+    } else {
+      // Unknown versions may exist when a downgraded coordinator is reading records from the log.
+      warn(s"Found unknown message key version: $version." +
+        s" The downgraded coordinator will ignore this key and corresponding value.")
+      None
+    }
   }
 
   /**
@@ -1229,17 +1239,21 @@ object GroupMetadataManager {
       Option(consumerRecord.key).map(key => GroupMetadataManager.readMessageKey(ByteBuffer.wrap(key))).foreach {
         // Only print if the message is an offset record.
         // We ignore the timestamp of the message because GroupMetadataMessage has its own timestamp.
-        case offsetKey: OffsetKey =>
-          val groupTopicPartition = offsetKey.key
-          val value = consumerRecord.value
-          val formattedValue =
-            if (value == null) "NULL"
-            else GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(value)).toString
-          output.write(groupTopicPartition.toString.getBytes(StandardCharsets.UTF_8))
-          output.write("::".getBytes(StandardCharsets.UTF_8))
-          output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
-          output.write("\n".getBytes(StandardCharsets.UTF_8))
-        case _ => // no-op
+        case Some(key) => key match {
+          case offsetKey: OffsetKey =>
+            val groupTopicPartition = offsetKey.key
+            val value = consumerRecord.value
+            val formattedValue =
+              if (value == null) "NULL"
+              else GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(value)).toString
+            output.write(groupTopicPartition.toString.getBytes(StandardCharsets.UTF_8))
+            output.write("::".getBytes(StandardCharsets.UTF_8))
+            output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
+            output.write("\n".getBytes(StandardCharsets.UTF_8))
+          case _ => // no-op
+        }
+
+        case None => // no-op
       }
     }
   }
@@ -1250,17 +1264,20 @@ object GroupMetadataManager {
       Option(consumerRecord.key).map(key => GroupMetadataManager.readMessageKey(ByteBuffer.wrap(key))).foreach {
         // Only print if the message is a group metadata record.
         // We ignore the timestamp of the message because GroupMetadataMessage has its own timestamp.
-        case groupMetadataKey: GroupMetadataKey =>
-          val groupId = groupMetadataKey.key
-          val value = consumerRecord.value
-          val formattedValue =
-            if (value == null) "NULL"
-            else GroupMetadataManager.readGroupMessageValue(groupId, ByteBuffer.wrap(value), Time.SYSTEM).toString
-          output.write(groupId.getBytes(StandardCharsets.UTF_8))
-          output.write("::".getBytes(StandardCharsets.UTF_8))
-          output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
-          output.write("\n".getBytes(StandardCharsets.UTF_8))
-        case _ => // no-op
+        case Some(key) => key match {
+          case groupMetadataKey: GroupMetadataKey =>
+            val groupId = groupMetadataKey.key
+            val value = consumerRecord.value
+            val formattedValue =
+              if (value == null) "NULL"
+              else GroupMetadataManager.readGroupMessageValue(groupId, ByteBuffer.wrap(value), Time.SYSTEM).toString
+            output.write(groupId.getBytes(StandardCharsets.UTF_8))
+            output.write("::".getBytes(StandardCharsets.UTF_8))
+            output.write(formattedValue.getBytes(StandardCharsets.UTF_8))
+            output.write("\n".getBytes(StandardCharsets.UTF_8))
+          case _ => // no-op
+        }
+        case None => // no-op
       }
     }
   }
@@ -1273,9 +1290,13 @@ object GroupMetadataManager {
       throw new KafkaException("Failed to decode message using offset topic decoder (message had a missing key)")
     } else {
       GroupMetadataManager.readMessageKey(record.key) match {
-        case offsetKey: OffsetKey => parseOffsets(offsetKey, record.value)
-        case groupMetadataKey: GroupMetadataKey => parseGroupMetadata(groupMetadataKey, record.value)
-        case _ => throw new KafkaException("Failed to decode message using offset topic decoder (message had an invalid key)")
+        case Some(key) => key match {
+          case offsetKey: OffsetKey => parseOffsets(offsetKey, record.value)
+          case groupMetadataKey: GroupMetadataKey => parseGroupMetadata(groupMetadataKey, record.value)
+          case _ => throw new KafkaException("Failed to decode message using offset topic decoder (message had an invalid key)")
+        }
+        case None =>
+          (Some("<UNKNOWN>"), Some("<UNKNOWN>"))
       }
     }
   }

core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala

@@ -19,8 +19,8 @@ package kafka.coordinator.transaction
 import java.io.PrintStream
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
-
 import kafka.internals.generated.{TransactionLogKey, TransactionLogValue}
+import kafka.utils.Logging
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.common.protocol.{ByteBufferAccessor, MessageUtil}
 import org.apache.kafka.common.record.{CompressionType, Record, RecordBatch}
@@ -37,7 +37,7 @@ import scala.jdk.CollectionConverters._
  * key version 0:               [transactionalId]
  *    -> value version 0:       [producer_id, producer_epoch, expire_timestamp, status, [topic, [partition] ], timestamp]
  */
-object TransactionLog {
+object TransactionLog extends Logging {
 
   // log-level config default values and enforced values
   val DefaultNumPartitions: Int = 50
@@ -98,15 +98,19 @@ object TransactionLog {
     *
     * @return the key
     */
-  def readTxnRecordKey(buffer: ByteBuffer): TxnKey = {
+  def readTxnRecordKey(buffer: ByteBuffer): Option[TxnKey] = {
     val version = buffer.getShort
     if (version >= TransactionLogKey.LOWEST_SUPPORTED_VERSION && version <= TransactionLogKey.HIGHEST_SUPPORTED_VERSION) {
       val value = new TransactionLogKey(new ByteBufferAccessor(buffer), version)
-      TxnKey(
+      Some(TxnKey(
         version = version,
         transactionalId = value.transactionalId
-      )
-    } else throw new IllegalStateException(s"Unknown version $version from the transaction log message")
+      ))
+    } else {
+      warn(s"Unknown version $version from the transaction log message." +
+        s" The downgraded coordinator will ignore this key and corresponding value.")
+      None
+    }
   }
 
   /**
@@ -148,17 +152,20 @@ 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 Some(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 None => // Only print if this message is a transaction record
       }
     }
   }
@@ -167,21 +174,26 @@ 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 Some(txnKey) =>
+        val keyString = s"transaction_metadata::transactionalId=${txnKey.transactionalId}"
+
+        val valueString = TransactionLog.readTxnRecordValue(txnKey.transactionalId, record.value) match {
+          case None => "<DELETE>"
 
-    (Some(keyString), Some(valueString))
+          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 None =>
+        (Some("<UNKNOWN>"), Some("<UNKNOWN>"))
+    }
   }
 
 }

core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala

@@ -467,16 +467,20 @@ class TransactionStateManager(brokerId: Int,
             memRecords.batches.forEach { batch =>
               for (record <- batch.asScala) {
                 require(record.hasKey, "Transaction state log's key should not be null")
-                val txnKey = TransactionLog.readTxnRecordKey(record.key)
-                // load transaction metadata along with transaction state
-                val transactionalId = txnKey.transactionalId
-                TransactionLog.readTxnRecordValue(transactionalId, record.value) match {
-                  case None =>
-                    loadedTransactions.remove(transactionalId)
-                  case Some(txnMetadata) =>
-                    loadedTransactions.put(transactionalId, txnMetadata)
+                TransactionLog.readTxnRecordKey(record.key) match {
+                  case Some(txnKey) =>
+                    // load transaction metadata along with transaction state
+                    val transactionalId = txnKey.transactionalId
+                    TransactionLog.readTxnRecordValue(transactionalId, record.value) match {
+                      case None =>
+                        loadedTransactions.remove(transactionalId)
+                      case Some(txnMetadata) =>
+                        loadedTransactions.put(transactionalId, txnMetadata)
+                    }
+                    currOffset = batch.nextOffset
+
+                  case None => // ignore unknown keys
                 }
-                currOffset = batch.nextOffset
               }
             }
           }