KAFKA-16517 Do not decode metadata records in the internal kraft partition listener

checkstyle/suppressions.xml

@@ -155,7 +155,7 @@
 
     <!-- Raft -->
     <suppress checks="NPathComplexity"
-              files="(DynamicVoter|RecordsIterator).java"/>
+              files="(DynamicVoter|DecodingStrategy).java"/>
 
     <suppress checks="JavaNCSS"
               files="(KafkaRaftClientTest).java"/>

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -37,6 +37,7 @@ import org.apache.kafka.common.utils.Utils
 import org.apache.kafka.common.utils.internals.ConfigUtils
 import org.apache.kafka.network.SocketServer
 import org.apache.kafka.raft.KafkaRaftClient
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy
 import org.apache.kafka.server.{DynamicThreadPool, ProcessRole}
 import org.apache.kafka.server.common.{ApiMessageAndVersion, DirectoryEventHandler}
 import org.apache.kafka.server.config.{BrokerReconfigurable => JBrokerReconfigurable, DynamicConfig, DynamicProducerStateManagerConfig, ServerConfigs, ServerLogConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
@@ -106,7 +107,7 @@ object DynamicBrokerConfig {
         Using.resource(
           RecordsSnapshotReader.of(
             rawSnapshotReader,
-            raftManager.recordSerde,
+            RecordsDecodingStrategy.dataAndControl(raftManager.recordSerde),
             BufferSupplier.create(),
             KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
             true,

metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java

@@ -33,6 +33,7 @@
 import org.apache.kafka.raft.errors.BufferAllocationException;
 import org.apache.kafka.raft.errors.NotLeaderException;
 import org.apache.kafka.raft.internals.MemoryBatchReader;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.common.KRaftVersion;
 import org.apache.kafka.server.common.OffsetAndEpoch;
@@ -479,7 +480,7 @@ private void scheduleLogCheck() {
                                 listenerData.handleLoadSnapshot(
                                     RecordsSnapshotReader.of(
                                         snapshot.get(),
-                                        MetadataRecordSerde.INSTANCE,
+                                        RecordsDecodingStrategy.dataAndControl(MetadataRecordSerde.INSTANCE),
                                         BufferSupplier.create(),
                                         Integer.MAX_VALUE,
                                         true,

raft/src/main/java/org/apache/kafka/raft/Batch.java

@@ -221,4 +221,41 @@ public static <T> Batch<T> data(
             List.of()
         );
     }
+
+    /**
+     * Create a batch whose records were not decoded, carrying only the offset information.
+     *
+     * @param baseOffset offset of the first record in the batch
+     * @param epoch epoch of the leader that created this batch
+     * @param appendTimestamp timestamp in milliseconds of when the batch was appended
+     * @param sizeInBytes number of bytes used by this batch
+     * @param numRecords the number of records in this batch
+     */
+    public static <T> Batch<T> notDecoded(
+        long baseOffset,
+        int epoch,
+        long appendTimestamp,
+        int sizeInBytes,
+        int numRecords
+    ) {
+        if (numRecords < 1) {
+            throw new IllegalArgumentException(
+                String.format(
+                    "Batch must contain at least one record; baseOffset = %d; epoch = %d",
+                    baseOffset,
+                    epoch
+                )
+            );
+        }
+
+        return new Batch<>(
+            baseOffset,
+            epoch,
+            appendTimestamp,
+            sizeInBytes,
+            baseOffset + numRecords - 1,
+            List.of(),
+            List.of()
+        );
+    }
 }

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -80,6 +80,7 @@
 import org.apache.kafka.raft.internals.KafkaRaftMetrics;
 import org.apache.kafka.raft.internals.MemoryBatchReader;
 import org.apache.kafka.raft.internals.RecordsBatchReader;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
 import org.apache.kafka.raft.internals.RemoveVoterHandler;
 import org.apache.kafka.raft.internals.RequestSendResult;
 import org.apache.kafka.raft.internals.ThresholdPurgatory;
@@ -452,7 +453,7 @@ private void updateListenersProgress(long highWatermark) {
     private Optional<SnapshotReader<T>> latestSnapshot() {
         return log.latestSnapshot().map(reader ->
             RecordsSnapshotReader.of(reader,
-                serde,
+                RecordsDecodingStrategy.dataAndControl(serde),
                 BufferSupplier.create(),
                 MAX_BATCH_SIZE_BYTES,
                 true, /* Validate batch CRC*/
@@ -498,7 +499,6 @@ public void initialize(
         partitionState = new KRaftControlRecordStateMachine(
             staticVoters,
             log,
-            serde,
             MAX_BATCH_SIZE_BYTES,
             logContext,
             kafkaRaftMetrics,

raft/src/main/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachine.java

@@ -28,7 +28,6 @@
 import org.apache.kafka.raft.RaftLog;
 import org.apache.kafka.raft.VoterSet;
 import org.apache.kafka.server.common.KRaftVersion;
-import org.apache.kafka.server.common.serialization.RecordSerde;
 import org.apache.kafka.snapshot.RawSnapshotReader;
 import org.apache.kafka.snapshot.RecordsSnapshotReader;
 import org.apache.kafka.snapshot.SnapshotReader;
@@ -54,7 +53,6 @@ public final class KRaftControlRecordStateMachine {
 
     private final LogContext logContext;
     private final RaftLog log;
-    private final RecordSerde<?> serde;
     private final Logger logger;
     private final int maxBatchSizeBytes;
 
@@ -80,14 +78,12 @@ public final class KRaftControlRecordStateMachine {
      *
      * @param staticVoterSet the set of voter statically configured
      * @param log the on disk topic partition
-     * @param serde the record decoder for data records
      * @param maxBatchSizeBytes the maximum size of record batch
      * @param logContext the log context
      */
     public KRaftControlRecordStateMachine(
         VoterSet staticVoterSet,
         RaftLog log,
-        RecordSerde<?> serde,
         int maxBatchSizeBytes,
         LogContext logContext,
         KafkaRaftMetrics kafkaRaftMetrics,
@@ -96,7 +92,6 @@ public KRaftControlRecordStateMachine(
         this.logContext = logContext;
         this.log = log;
         this.voterSetHistory = new VoterSetHistory(staticVoterSet, logContext);
-        this.serde = serde;
         this.maxBatchSizeBytes = maxBatchSizeBytes;
         this.logger = logContext.logger(getClass());
         this.kafkaRaftMetrics = kafkaRaftMetrics;
@@ -237,7 +232,7 @@ private void maybeLoadLog() {
                 );
                 try (RecordsIterator<?> iterator = new RecordsIterator<>(
                         info.records,
-                        serde,
+                        RecordsDecodingStrategy.controlOnly(),
                         bufferSupplier,
                         maxBatchSizeBytes,
                         true, // Validate batch CRC
@@ -269,7 +264,7 @@ private void maybeLoadSnapshot() {
             try (BufferSupplier bufferSupplier = BufferSupplier.create();
                  SnapshotReader<?> reader = RecordsSnapshotReader.of(
                     rawSnapshot,
-                    serde,
+                    RecordsDecodingStrategy.controlOnly(),
                     bufferSupplier,
                     maxBatchSizeBytes,
                     true, // Validate batch CRC

raft/src/main/java/org/apache/kafka/raft/internals/RecordsBatchReader.java

@@ -108,7 +108,7 @@ public static <T> RecordsBatchReader<T> of(
     ) {
         return new RecordsBatchReader<>(
             baseOffset,
-            new RecordsIterator<>(records, serde, bufferSupplier, maxBatchSize, doCrcValidation, logContext),
+            new RecordsIterator<>(records, RecordsDecodingStrategy.dataAndControl(serde), bufferSupplier, maxBatchSize, doCrcValidation, logContext),
             closeListener
         );
     }