of(
) {
return new RecordsBatchReader<>(
baseOffset,
- new RecordsIterator<>(records, serde, bufferSupplier, maxBatchSize, doCrcValidation, logContext),
+ new RecordsIterator<>(
+ records,
+ RecordsDecodingStrategy.dataAndControl(serde),
+ bufferSupplier,
+ maxBatchSize,
+ doCrcValidation,
+ logContext
+ ),
closeListener
);
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/RecordsDecodingStrategy.java b/raft/src/main/java/org/apache/kafka/raft/internals/RecordsDecodingStrategy.java
new file mode 100644
index 0000000000000..9f81acf6ac2f2
--- /dev/null
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/RecordsDecodingStrategy.java
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.internal.DefaultRecordBatch;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
+import org.apache.kafka.common.utils.internals.ByteUtils;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.raft.ControlRecord;
+import org.apache.kafka.server.common.serialization.RecordSerde;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiFunction;
+
+/**
+ * Decides which records {@link RecordsIterator} decodes when turning a batch into a {@link Batch}.
+ * A batch's records are decoded only when this strategy is interested in them; otherwise the batch
+ * is returned as a {@link Batch#skipped} batch carrying only the offset information.
+ *
+ * Use one of the factory methods to select the behavior:
+ *
+ * - {@link #dataAndControl} decodes both the control and data records.
+ * - {@link #controlOnly} decodes only the control records and skips the data records. Used by
+ * the internal kraft partition listener, which needs no serde.
+ * - {@link #dataOnly} decodes only the data records and skips the control records.
+ * - {@link #none} skips both.
+ *
+ */
+public final class RecordsDecodingStrategy {
+ private final boolean decodeControlRecords;
+ // When present, data records are decoded with this serde; when empty, they are skipped.
+ private final Optional> serde;
+
+ private RecordsDecodingStrategy(boolean decodeControlRecords, Optional> serde) {
+ this.decodeControlRecords = decodeControlRecords;
+ this.serde = serde;
+ }
+
+ /**
+ * Decodes both the control and data records of a batch.
+ */
+ public static RecordsDecodingStrategy dataAndControl(RecordSerde serde) {
+ return new RecordsDecodingStrategy<>(true, Optional.of(serde));
+ }
+
+ /**
+ * Decodes only the data records of a batch and skips the control records.
+ */
+ public static RecordsDecodingStrategy dataOnly(RecordSerde serde) {
+ return new RecordsDecodingStrategy<>(false, Optional.of(serde));
+ }
+
+ /**
+ * Decodes only the control records of a batch and skips the data records.
+ */
+ public static RecordsDecodingStrategy controlOnly() {
+ return new RecordsDecodingStrategy<>(true, Optional.empty());
+ }
+
+ /**
+ * Skips both the control and data records of a batch.
+ */
+ public static RecordsDecodingStrategy none() {
+ return new RecordsDecodingStrategy<>(false, Optional.empty());
+ }
+
+ Batch readBatch(DefaultRecordBatch batch, BufferSupplier bufferSupplier, int numRecords) {
+ if (batch.isControlBatch()) {
+ return decodeControlRecords ? readControlBatch(batch, bufferSupplier, numRecords) : skippedBatch(batch, numRecords);
+ } else {
+ return serde
+ .map(value -> readDataBatch(batch, value, bufferSupplier, numRecords))
+ .orElseGet(() -> skippedBatch(batch, numRecords));
+ }
+ }
+
+ private static Batch readControlBatch(DefaultRecordBatch batch, BufferSupplier bufferSupplier, int numRecords) {
+ InputStream input = batch.recordInputStream(bufferSupplier);
+ try {
+ List records = new ArrayList<>(numRecords);
+ for (int i = 0; i < numRecords; i++) {
+ records.add(readRecord(input, batch.sizeInBytes(), bufferSupplier, RecordsDecodingStrategy::decodeControlRecord));
+ }
+ return Batch.control(
+ batch.baseOffset(),
+ batch.partitionLeaderEpoch(),
+ batch.maxTimestamp(),
+ batch.sizeInBytes(),
+ records
+ );
+ } finally {
+ Utils.closeQuietly(input, "BytesStream for input containing records");
+ }
+ }
+
+ private static Batch readDataBatch(DefaultRecordBatch batch, RecordSerde serde, BufferSupplier bufferSupplier, int numRecords) {
+ InputStream input = batch.recordInputStream(bufferSupplier);
+ try {
+ List records = new ArrayList<>(numRecords);
+ for (int i = 0; i < numRecords; i++) {
+ records.add(readRecord(input, batch.sizeInBytes(), bufferSupplier, (key, value) -> decodeDataRecord(key, value, serde)));
+ }
+ return Batch.data(
+ batch.baseOffset(),
+ batch.partitionLeaderEpoch(),
+ batch.maxTimestamp(),
+ batch.sizeInBytes(),
+ records
+ );
+ } finally {
+ Utils.closeQuietly(input, "BytesStream for input containing records");
+ }
+ }
+
+ private static Batch skippedBatch(DefaultRecordBatch batch, int numRecords) {
+ return Batch.skipped(
+ batch.baseOffset(),
+ batch.partitionLeaderEpoch(),
+ batch.maxTimestamp(),
+ batch.sizeInBytes(),
+ numRecords
+ );
+ }
+
+ @SuppressWarnings("NPathComplexity")
+ private static U readRecord(
+ InputStream stream,
+ int totalBatchSize,
+ BufferSupplier bufferSupplier,
+ BiFunction, Optional, U> decoder
+ ) {
+ // Read size of body in bytes
+ int size;
+ try {
+ size = ByteUtils.readVarint(stream);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Unable to read record size", e);
+ }
+ if (size <= 0) {
+ throw new RuntimeException("Invalid non-positive frame size: " + size);
+ }
+ if (size > totalBatchSize) {
+ throw new RuntimeException("Specified frame size, " + size + ", is larger than the entire size of the " +
+ "batch, which is " + totalBatchSize);
+ }
+ ByteBuffer buf = bufferSupplier.get(size);
+
+ // The last byte of the buffer is reserved for a varint set to the number of record headers, which
+ // must be 0. Therefore, we set the ByteBuffer limit to size - 1.
+ buf.limit(size - 1);
+
+ try {
+ int bytesRead = stream.read(buf.array(), 0, size);
+ if (bytesRead != size) {
+ throw new RuntimeException("Unable to read " + size + " bytes, only read " + bytesRead);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to read record bytes", e);
+ }
+ try {
+ ByteBufferAccessor input = new ByteBufferAccessor(buf);
+
+ // Read unused attributes
+ input.readByte();
+
+ long timestampDelta = input.readVarlong();
+ if (timestampDelta != 0) {
+ throw new IllegalArgumentException("Got timestamp delta of " + timestampDelta + ", but this is invalid because it " +
+ "is not 0 as expected.");
+ }
+
+ // Read offset delta
+ input.readVarint();
+
+ // Read the key
+ int keySize = input.readVarint();
+ Optional key = Optional.empty();
+ if (keySize >= 0) {
+ key = Optional.of(input.readByteBuffer(keySize));
+ }
+
+ // Read the value
+ int valueSize = input.readVarint();
+ Optional value = Optional.empty();
+ if (valueSize >= 0) {
+ value = Optional.of(input.readByteBuffer(valueSize));
+ }
+
+ // Read the record body from the file input reader
+ U record = decoder.apply(key, value);
+
+ // Read the number of headers. Currently, this must be a single byte set to 0.
+ int numHeaders = buf.array()[size - 1];
+ if (numHeaders != 0) {
+ throw new IllegalArgumentException("Got numHeaders of " + numHeaders + ", but this is invalid because " +
+ "it is not 0 as expected.");
+ }
+
+ return record;
+ } finally {
+ bufferSupplier.release(buf);
+ }
+ }
+
+ private static T decodeDataRecord(Optional key, Optional value, RecordSerde serde) {
+ if (key.isPresent()) {
+ throw new IllegalArgumentException("Got key in the record when no key was expected");
+ }
+
+ if (value.isEmpty()) {
+ throw new IllegalArgumentException("Missing value in the record when a value was expected");
+ } else if (value.get().remaining() == 0) {
+ throw new IllegalArgumentException("Got an unexpected empty value in the record");
+ }
+
+ ByteBuffer valueBuffer = value.get();
+
+ return serde.read(new ByteBufferAccessor(valueBuffer), valueBuffer.remaining());
+ }
+
+ private static ControlRecord decodeControlRecord(Optional key, Optional value) {
+ if (key.isEmpty()) {
+ throw new IllegalArgumentException("Missing key in the record when a key was expected");
+ } else if (key.get().remaining() == 0) {
+ throw new IllegalArgumentException("Got an unexpected empty key in the record");
+ }
+
+ if (value.isEmpty()) {
+ throw new IllegalArgumentException("Missing value in the record when a value was expected");
+ } else if (value.get().remaining() == 0) {
+ throw new IllegalArgumentException("Got an unexpected empty value in the record");
+ }
+
+ return ControlRecord.of(key.get(), value.get());
+ }
+}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java b/raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java
index f0c17b267c620..d025babe15f58 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java
@@ -16,38 +16,29 @@
*/
package org.apache.kafka.raft.internals;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.record.internal.DefaultRecordBatch;
import org.apache.kafka.common.record.internal.FileRecords;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.MutableRecordBatch;
import org.apache.kafka.common.record.internal.Records;
-import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.BufferSupplier;
-import org.apache.kafka.common.utils.internals.ByteUtils;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.raft.Batch;
-import org.apache.kafka.raft.ControlRecord;
-import org.apache.kafka.server.common.serialization.RecordSerde;
import org.slf4j.Logger;
import java.io.IOException;
-import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
-import java.util.List;
import java.util.NoSuchElementException;
import java.util.Optional;
-import java.util.function.BiFunction;
public final class RecordsIterator implements Iterator>, AutoCloseable {
private final Logger logger;
private final Records records;
- private final RecordSerde serde;
+ private final RecordsDecodingStrategy decodingStrategy;
private final BufferSupplier bufferSupplier;
private final int batchSize;
// Setting to true will make the RecordsIterator perform a CRC Validation
@@ -65,20 +56,20 @@ public final class RecordsIterator implements Iterator>, AutoCloseab
/**
* This class provides an iterator over records retrieved via the raft client or from a snapshot
* @param records the records
- * @param serde the serde to deserialize records
+ * @param decodingStrategy the strategy deciding which records to decode (control, data, or both)
* @param bufferSupplier the buffer supplier implementation to allocate buffers when reading records. This must return ByteBuffer allocated on the heap
* @param batchSize the maximum batch size
*/
public RecordsIterator(
Records records,
- RecordSerde serde,
+ RecordsDecodingStrategy decodingStrategy,
BufferSupplier bufferSupplier,
int batchSize,
boolean doCrcValidation,
LogContext logContext
) {
this.records = records;
- this.serde = serde;
+ this.decodingStrategy = decodingStrategy;
this.bufferSupplier = bufferSupplier;
this.batchSize = Math.max(batchSize, Records.HEADER_SIZE_UP_TO_MAGIC);
this.doCrcValidation = doCrcValidation;
@@ -218,155 +209,6 @@ private Batch readBatch(DefaultRecordBatch batch) {
throw new IllegalStateException("Expected a record count for the records batch");
}
- InputStream input = batch.recordInputStream(bufferSupplier);
- final Batch result;
- try {
- if (batch.isControlBatch()) {
- List records = new ArrayList<>(numRecords);
- for (int i = 0; i < numRecords; i++) {
- ControlRecord record = readRecord(
- input,
- batch.sizeInBytes(),
- RecordsIterator::decodeControlRecord
- );
- records.add(record);
- }
- result = Batch.control(
- batch.baseOffset(),
- batch.partitionLeaderEpoch(),
- batch.maxTimestamp(),
- batch.sizeInBytes(),
- records
- );
- } else {
- List records = new ArrayList<>(numRecords);
- for (int i = 0; i < numRecords; i++) {
- T record = readRecord(input, batch.sizeInBytes(), this::decodeDataRecord);
- records.add(record);
- }
-
- result = Batch.data(
- batch.baseOffset(),
- batch.partitionLeaderEpoch(),
- batch.maxTimestamp(),
- batch.sizeInBytes(),
- records
- );
- }
- } finally {
- Utils.closeQuietly(input, "BytesStream for input containing records");
- }
-
- return result;
- }
-
- private U readRecord(
- InputStream stream,
- int totalBatchSize,
- BiFunction, Optional, U> decoder
- ) {
- // Read size of body in bytes
- int size;
- try {
- size = ByteUtils.readVarint(stream);
- } catch (IOException e) {
- throw new UncheckedIOException("Unable to read record size", e);
- }
- if (size <= 0) {
- throw new RuntimeException("Invalid non-positive frame size: " + size);
- }
- if (size > totalBatchSize) {
- throw new RuntimeException("Specified frame size, " + size + ", is larger than the entire size of the " +
- "batch, which is " + totalBatchSize);
- }
- ByteBuffer buf = bufferSupplier.get(size);
-
- // The last byte of the buffer is reserved for a varint set to the number of record headers, which
- // must be 0. Therefore, we set the ByteBuffer limit to size - 1.
- buf.limit(size - 1);
-
- try {
- int bytesRead = stream.read(buf.array(), 0, size);
- if (bytesRead != size) {
- throw new RuntimeException("Unable to read " + size + " bytes, only read " + bytesRead);
- }
- } catch (IOException e) {
- throw new UncheckedIOException("Failed to read record bytes", e);
- }
- try {
- ByteBufferAccessor input = new ByteBufferAccessor(buf);
-
- // Read unused attributes
- input.readByte();
-
- long timestampDelta = input.readVarlong();
- if (timestampDelta != 0) {
- throw new IllegalArgumentException("Got timestamp delta of " + timestampDelta + ", but this is invalid because it " +
- "is not 0 as expected.");
- }
-
- // Read offset delta
- input.readVarint();
-
- // Read the key
- int keySize = input.readVarint();
- Optional key = Optional.empty();
- if (keySize >= 0) {
- key = Optional.of(input.readByteBuffer(keySize));
- }
-
- // Read the value
- int valueSize = input.readVarint();
- Optional value = Optional.empty();
- if (valueSize >= 0) {
- value = Optional.of(input.readByteBuffer(valueSize));
- }
-
- // Read the metadata record body from the file input reader
- U record = decoder.apply(key, value);
-
- // Read the number of headers. Currently, this must be a single byte set to 0.
- int numHeaders = buf.array()[size - 1];
- if (numHeaders != 0) {
- throw new IllegalArgumentException("Got numHeaders of " + numHeaders + ", but this is invalid because " +
- "it is not 0 as expected.");
- }
-
- return record;
- } finally {
- bufferSupplier.release(buf);
- }
- }
-
- private T decodeDataRecord(Optional key, Optional value) {
- if (key.isPresent()) {
- throw new IllegalArgumentException("Got key in the record when no key was expected");
- }
-
- if (value.isEmpty()) {
- throw new IllegalArgumentException("Missing value in the record when a value was expected");
- } else if (value.get().remaining() == 0) {
- throw new IllegalArgumentException("Got an unexpected empty value in the record");
- }
-
- ByteBuffer valueBuffer = value.get();
-
- return serde.read(new ByteBufferAccessor(valueBuffer), valueBuffer.remaining());
- }
-
- private static ControlRecord decodeControlRecord(Optional key, Optional value) {
- if (key.isEmpty()) {
- throw new IllegalArgumentException("Missing key in the record when a key was expected");
- } else if (key.get().remaining() == 0) {
- throw new IllegalArgumentException("Got an unexpected empty key in the record");
- }
-
- if (value.isEmpty()) {
- throw new IllegalArgumentException("Missing value in the record when a value was expected");
- } else if (value.get().remaining() == 0) {
- throw new IllegalArgumentException("Got an unexpected empty value in the record");
- }
-
- return ControlRecord.of(key.get(), value.get());
+ return decodingStrategy.readBatch(batch, bufferSupplier, numRecords);
}
}
diff --git a/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotReader.java b/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotReader.java
index 9bff4a1c8fcdb..807107e224085 100644
--- a/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotReader.java
+++ b/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotReader.java
@@ -22,6 +22,7 @@
import org.apache.kafka.common.utils.internals.BufferSupplier;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.raft.Batch;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.raft.internals.RecordsIterator;
import org.apache.kafka.server.common.OffsetAndEpoch;
import org.apache.kafka.server.common.serialization.RecordSerde;
@@ -115,10 +116,36 @@ public static RecordsSnapshotReader of(
int maxBatchSize,
boolean doCrcValidation,
LogContext logContext
+ ) {
+ return ofDecodingStrategy(
+ snapshot,
+ RecordsDecodingStrategy.dataAndControl(serde),
+ bufferSupplier,
+ maxBatchSize,
+ doCrcValidation,
+ logContext
+ );
+ }
+
+ // Used within the raft module to pass an explicit decoding strategy.
+ public static RecordsSnapshotReader ofDecodingStrategy(
+ RawSnapshotReader snapshot,
+ RecordsDecodingStrategy decodingStrategy,
+ BufferSupplier bufferSupplier,
+ int maxBatchSize,
+ boolean doCrcValidation,
+ LogContext logContext
) {
return new RecordsSnapshotReader<>(
snapshot.snapshotId(),
- new RecordsIterator<>(snapshot.records(), serde, bufferSupplier, maxBatchSize, doCrcValidation, logContext)
+ new RecordsIterator<>(
+ snapshot.records(),
+ decodingStrategy,
+ bufferSupplier,
+ maxBatchSize,
+ doCrcValidation,
+ logContext
+ )
);
}
diff --git a/raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java b/raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java
index 65002862e4677..f421cc3b1bdfa 100644
--- a/raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java
+++ b/raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java
@@ -20,7 +20,7 @@
import org.apache.kafka.common.utils.internals.BufferSupplier;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.raft.KafkaRaftClient;
-import org.apache.kafka.raft.internals.IdentitySerde;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.server.common.OffsetAndEpoch;
import org.slf4j.Logger;
@@ -163,9 +163,9 @@ public static Path markForDelete(Path logDir, OffsetAndEpoch snapshotId) {
public static long lastContainedLogTimestamp(RawSnapshotReader reader, LogContext logContext) {
try (var bufferSupplier = new BufferSupplier.GrowableBufferSupplier();
RecordsSnapshotReader recordsSnapshotReader =
- RecordsSnapshotReader.of(
+ RecordsSnapshotReader.ofDecodingStrategy(
reader,
- IdentitySerde.INSTANCE,
+ RecordsDecodingStrategy.controlOnly(),
bufferSupplier,
KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
true,
diff --git a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
index 08f3d63833b68..a5c15c9d097ae 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
@@ -42,6 +42,7 @@
import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.common.utils.internals.BufferSupplier;
import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.server.common.Feature;
import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.snapshot.RecordsSnapshotReader;
@@ -117,9 +118,9 @@ public void testLeaderWritesBootstrapRecords() throws Exception {
// check the bootstrap snapshot exists and contains the expected records
assertEquals(BOOTSTRAP_SNAPSHOT_ID, context.log.latestSnapshotId().get());
- try (SnapshotReader reader = RecordsSnapshotReader.of(
+ try (SnapshotReader reader = RecordsSnapshotReader.ofDecodingStrategy(
context.log.latestSnapshot().get(),
- context.serde,
+ RecordsDecodingStrategy.dataAndControl(context.serde),
BufferSupplier.NO_CACHING,
KafkaRaftClient.MAX_BATCH_SIZE_BYTES,
false,
diff --git a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
index 3852921f7e125..517f70dfc3ddd 100644
--- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
@@ -35,6 +35,7 @@
import org.apache.kafka.raft.MockLog.LogBatch;
import org.apache.kafka.raft.MockLog.LogEntry;
import org.apache.kafka.raft.internals.BatchMemoryPool;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.server.common.Feature;
import org.apache.kafka.server.common.serialization.RecordSerde;
import org.apache.kafka.snapshot.RecordsSnapshotReader;
@@ -1250,9 +1251,9 @@ private void assertCommittedData(RaftNode node) {
assertTrue(snapshotId.offset() <= highWatermark.getAsLong());
startOffset.set(snapshotId.offset());
- try (SnapshotReader snapshot = RecordsSnapshotReader.of(
+ try (SnapshotReader snapshot = RecordsSnapshotReader.ofDecodingStrategy(
log.readSnapshot(snapshotId).get(),
- node.intSerde,
+ RecordsDecodingStrategy.dataAndControl(node.intSerde),
BufferSupplier.create(),
Integer.MAX_VALUE,
true,
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java
index 077b7da5b4d3e..da06d44a680e1 100644
--- a/raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java
@@ -63,7 +63,6 @@ private static KRaftControlRecordStateMachine buildPartitionListener(
return new KRaftControlRecordStateMachine(
staticVoterSet,
log,
- STRING_SERDE,
1024,
new LogContext(),
raftMetrics,
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/RecordsDecodingStrategyTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/RecordsDecodingStrategyTest.java
new file mode 100644
index 0000000000000..c7a6226576eaa
--- /dev/null
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/RecordsDecodingStrategyTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.message.KRaftVersionRecord;
+import org.apache.kafka.common.record.internal.CompressionType;
+import org.apache.kafka.common.record.internal.ControlRecordType;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
+import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.raft.ControlRecord;
+import org.apache.kafka.raft.internals.RecordsIteratorTest.TestBatch;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+final class RecordsDecodingStrategyTest {
+ private static final StringSerde STRING_SERDE = new StringSerde();
+
+ // The records under test: a control batch at offset 0 followed by a data batch at offsets 1,2,3.
+ private static final long CONTROL_BASE_OFFSET = 0L;
+ private static final long DATA_BASE_OFFSET = 1L;
+ private static final List DATA_RECORDS = List.of("a", "b", "c");
+ private static final long DATA_LAST_OFFSET = DATA_BASE_OFFSET + DATA_RECORDS.size() - 1;
+ private static final List CONTROL_RECORDS = List.of(ControlRecord.of(new KRaftVersionRecord()));
+ private static final MemoryRecords RECORDS = controlBatchThenDataBatch();
+
+ private static RecordsIterator iterator(RecordsDecodingStrategy strategy) {
+ return new RecordsIterator<>(
+ RECORDS,
+ strategy,
+ BufferSupplier.NO_CACHING,
+ Integer.MAX_VALUE,
+ true,
+ new LogContext()
+ );
+ }
+
+ @Test
+ void testDataAndControl() {
+ try (RecordsIterator iterator = iterator(RecordsDecodingStrategy.dataAndControl(STRING_SERDE))) {
+ assertControlBatchDecoded(iterator.next());
+ assertDataBatchDecoded(iterator.next());
+ }
+ }
+
+ @Test
+ void testControlOnly() {
+ try (RecordsIterator iterator = iterator(RecordsDecodingStrategy.controlOnly())) {
+ assertControlBatchDecoded(iterator.next());
+ assertDataBatchSkipped(iterator.next());
+ }
+ }
+
+ @Test
+ void testDataOnly() {
+ try (RecordsIterator iterator = iterator(RecordsDecodingStrategy.dataOnly(STRING_SERDE))) {
+ assertControlBatchSkipped(iterator.next());
+ assertDataBatchDecoded(iterator.next());
+ }
+ }
+
+ @Test
+ void testNone() {
+ try (RecordsIterator iterator = iterator(RecordsDecodingStrategy.none())) {
+ assertControlBatchSkipped(iterator.next());
+ assertDataBatchSkipped(iterator.next());
+ }
+ }
+
+ private static void assertControlBatchDecoded(Batch batch) {
+ assertEquals(CONTROL_RECORDS, batch.controlRecords());
+ assertEquals(CONTROL_BASE_OFFSET, batch.baseOffset());
+ assertEquals(CONTROL_BASE_OFFSET, batch.lastOffset());
+ }
+
+ private static void assertControlBatchSkipped(Batch batch) {
+ assertTrue(batch.controlRecords().isEmpty());
+ assertEquals(CONTROL_BASE_OFFSET, batch.baseOffset());
+ assertEquals(CONTROL_BASE_OFFSET, batch.lastOffset());
+ }
+
+ private static void assertDataBatchDecoded(Batch batch) {
+ assertEquals(DATA_RECORDS, batch.records());
+ assertEquals(DATA_BASE_OFFSET, batch.baseOffset());
+ assertEquals(DATA_LAST_OFFSET, batch.lastOffset());
+ }
+
+ private static void assertDataBatchSkipped(Batch batch) {
+ assertTrue(batch.records().isEmpty());
+ assertEquals(DATA_BASE_OFFSET, batch.baseOffset());
+ assertEquals(DATA_LAST_OFFSET, batch.lastOffset());
+ }
+
+ // Builds records containing a control batch at CONTROL_BASE_OFFSET followed by a data batch at DATA_BASE_OFFSET.
+ private static MemoryRecords controlBatchThenDataBatch() {
+ MemoryRecords controlRecords = RecordsIteratorTest.buildControlRecords(ControlRecordType.KRAFT_VERSION);
+ TestBatch dataBatch = new TestBatch<>(DATA_BASE_OFFSET, 1, 100L, DATA_RECORDS);
+ MemoryRecords dataRecords = RecordsIteratorTest.buildRecords(CompressionType.NONE, List.of(dataBatch));
+
+ ByteBuffer combined = ByteBuffer.allocate(controlRecords.sizeInBytes() + dataRecords.sizeInBytes());
+ combined.put(controlRecords.buffer());
+ combined.put(dataRecords.buffer());
+ combined.flip();
+ return MemoryRecords.readableRecords(combined);
+ }
+}
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/RecordsIteratorTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/RecordsIteratorTest.java
index d4c189432383a..97006a7ad8bd7 100644
--- a/raft/src/test/java/org/apache/kafka/raft/internals/RecordsIteratorTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/RecordsIteratorTest.java
@@ -326,7 +326,7 @@ static RecordsIterator createIterator(
) {
return new RecordsIterator<>(
records,
- STRING_SERDE,
+ RecordsDecodingStrategy.dataAndControl(STRING_SERDE),
bufferSupplier,
Records.HEADER_SIZE_UP_TO_MAGIC,
validateCrc,
diff --git a/raft/src/test/java/org/apache/kafka/snapshot/RecordsSnapshotWriterTest.java b/raft/src/test/java/org/apache/kafka/snapshot/RecordsSnapshotWriterTest.java
index eb23728fcabfd..d13cc964b51fd 100644
--- a/raft/src/test/java/org/apache/kafka/snapshot/RecordsSnapshotWriterTest.java
+++ b/raft/src/test/java/org/apache/kafka/snapshot/RecordsSnapshotWriterTest.java
@@ -27,6 +27,7 @@
import org.apache.kafka.raft.Batch;
import org.apache.kafka.raft.VoterSet;
import org.apache.kafka.raft.VoterSetTest;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.raft.internals.StringSerde;
import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.OffsetAndEpoch;
@@ -64,9 +65,9 @@ void testBuilderKRaftVersion0() {
snapshot.freeze();
}
- try (RecordsSnapshotReader reader = RecordsSnapshotReader.of(
+ try (RecordsSnapshotReader reader = RecordsSnapshotReader.ofDecodingStrategy(
new MockRawSnapshotReader(snapshotId, buffer.get()),
- STRING_SERDE,
+ RecordsDecodingStrategy.dataAndControl(STRING_SERDE),
BufferSupplier.NO_CACHING,
maxBatchSizeBytes,
true,
@@ -137,9 +138,9 @@ void testKBuilderRaftVersion1WithVoterSet() {
snapshot.freeze();
}
- try (RecordsSnapshotReader reader = RecordsSnapshotReader.of(
+ try (RecordsSnapshotReader reader = RecordsSnapshotReader.ofDecodingStrategy(
new MockRawSnapshotReader(snapshotId, buffer.get()),
- STRING_SERDE,
+ RecordsDecodingStrategy.dataAndControl(STRING_SERDE),
BufferSupplier.NO_CACHING,
maxBatchSizeBytes,
true,
@@ -195,9 +196,9 @@ void testBuilderKRaftVersion1WithoutVoterSet() {
snapshot.freeze();
}
- try (RecordsSnapshotReader reader = RecordsSnapshotReader.of(
+ try (RecordsSnapshotReader reader = RecordsSnapshotReader.ofDecodingStrategy(
new MockRawSnapshotReader(snapshotId, buffer.get()),
- STRING_SERDE,
+ RecordsDecodingStrategy.dataAndControl(STRING_SERDE),
BufferSupplier.NO_CACHING,
maxBatchSizeBytes,
true,
diff --git a/raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java b/raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java
index c57af8084bf5a..dfec0c0950d93 100644
--- a/raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java
+++ b/raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java
@@ -27,6 +27,7 @@
import org.apache.kafka.raft.Batch;
import org.apache.kafka.raft.ControlRecord;
import org.apache.kafka.raft.RaftClientTestContext;
+import org.apache.kafka.raft.internals.RecordsDecodingStrategy;
import org.apache.kafka.raft.internals.StringSerde;
import org.apache.kafka.server.common.OffsetAndEpoch;
@@ -191,9 +192,9 @@ private SnapshotReader readSnapshot(
OffsetAndEpoch snapshotId,
int maxBatchSize
) {
- return RecordsSnapshotReader.of(
+ return RecordsSnapshotReader.ofDecodingStrategy(
context.log.readSnapshot(snapshotId).get(),
- context.serde,
+ RecordsDecodingStrategy.dataAndControl(context.serde),
BufferSupplier.create(),
maxBatchSize,
true,
@@ -251,9 +252,9 @@ record = records.next();
public static void assertDataSnapshot(List> batches, RawSnapshotReader reader) {
assertDataSnapshot(
batches,
- RecordsSnapshotReader.of(
+ RecordsSnapshotReader.ofDecodingStrategy(
reader,
- new StringSerde(),
+ RecordsDecodingStrategy.dataAndControl(new StringSerde()),
BufferSupplier.create(),
Integer.MAX_VALUE,
true,