You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by li...@apache.org on 2026/09/23 16:31:01 UTC
(kafka) branch trunk updated: KAFKA-20632: Extend existing tests (#23135)
This is an automated email from the ASF dual-hosted git repository.
lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 60341d307ee KAFKA-20632: Extend existing tests (#23135)
60341d307ee is described below
commit 60341d307ee307fa9991c11e2762cb53f9a45893
Author: Nikita Shupletsov <ni...@shupletsov.ca>
AuthorDate: Wed Sep 23 09:30:53 2026 -0700
KAFKA-20632: Extend existing tests (#23135)
to verify the correctness of incremental allocation strategy.
Reviewers: Lianet Magrans <lm...@confluent.io>
---
.../producer/internals/RecordAccumulatorTest.java | 330 +++++++++++++--------
.../clients/producer/internals/SenderTest.java | 77 +++--
2 files changed, 255 insertions(+), 152 deletions(-)
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
index 1867aee0a5d..477098d0fce 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
@@ -22,6 +22,7 @@ import org.apache.kafka.clients.MetadataSnapshot;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.producer.BufferExhaustedException;
import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
@@ -55,6 +56,8 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mockito;
import java.nio.ByteBuffer;
@@ -83,6 +86,8 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.Arrays.asList;
+import static org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL;
+import static org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -137,8 +142,9 @@ public class RecordAccumulatorTest {
this.metrics.close();
}
- @Test
- public void testDrainBatches() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testDrainBatches(String allocationStrategy) throws Exception {
// test case: node1(tp1,tp2) , node2(tp3,tp4)
// add tp-4
int partition4 = 3;
@@ -165,8 +171,9 @@ public class RecordAccumulatorTest {
null,
Collections.emptyMap(),
cluster);
+ byte[] value = new byte[512];
long batchSize = value.length + DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
- RecordAccumulator accum = createTestRecordAccumulator((int) batchSize, Integer.MAX_VALUE, Compression.NONE, 10);
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, (int) batchSize, Integer.MAX_VALUE, Compression.NONE, 10);
// initial data
@@ -226,14 +233,15 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testFull() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testFull(String allocationStrategy) throws Exception {
long now = time.milliseconds();
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10L * batchSize, Compression.NONE, 10);
int appends = expectedNumAppends(batchSize);
for (int i = 0; i < appends; i++) {
@@ -271,18 +279,19 @@ public class RecordAccumulatorTest {
@Test
public void testAppendLargeCompressed() throws Exception {
- testAppendLarge(Compression.gzip().build());
+ testAppendLarge(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, Compression.gzip().build());
}
- @Test
- public void testAppendLargeNonCompressed() throws Exception {
- testAppendLarge(Compression.NONE);
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAppendLargeNonCompressed(String allocationStrategy) throws Exception {
+ testAppendLarge(allocationStrategy, Compression.NONE);
}
- private void testAppendLarge(Compression compression) throws Exception {
+ private void testAppendLarge(String allocationStrategy, Compression compression) throws Exception {
int batchSize = 512;
byte[] value = new byte[2 * batchSize];
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, compression, 0);
accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), metadataCache.cluster());
assertEquals(Collections.singleton(node1), accum.ready(metadataCache, time.milliseconds()).readyNodes, "Our partition's leader should be ready");
@@ -305,22 +314,23 @@ public class RecordAccumulatorTest {
@Test
public void testAppendLargeOldMessageFormatCompressed() throws Exception {
- testAppendLargeOldMessageFormat(Compression.gzip().build());
+ testAppendLargeOldMessageFormat(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, Compression.gzip().build());
}
- @Test
- public void testAppendLargeOldMessageFormatNonCompressed() throws Exception {
- testAppendLargeOldMessageFormat(Compression.NONE);
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAppendLargeOldMessageFormatNonCompressed(String allocationStrategy) throws Exception {
+ testAppendLargeOldMessageFormat(allocationStrategy, Compression.NONE);
}
- private void testAppendLargeOldMessageFormat(Compression compression) throws Exception {
+ private void testAppendLargeOldMessageFormat(String allocationStrategy, Compression compression) throws Exception {
int batchSize = 512;
byte[] value = new byte[2 * batchSize];
ApiVersions apiVersions = new ApiVersions();
apiVersions.update(node1.idString(), NodeApiVersions.create(ApiKeys.PRODUCE.id, (short) 0, (short) 2));
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, compression, 0);
accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), metadataCache.cluster());
assertEquals(Collections.singleton(node1), accum.ready(metadataCache, time.milliseconds()).readyNodes, "Our partition's leader should be ready");
@@ -341,10 +351,11 @@ public class RecordAccumulatorTest {
assertEquals(0L, record.timestamp());
}
- @Test
- public void testLinger() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testLinger(String allocationStrategy) throws Exception {
int lingerMs = 10;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, Compression.NONE, lingerMs);
accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), cluster);
assertEquals(0, accum.ready(metadataCache, time.milliseconds()).readyNodes.size(), "No partitions should be ready");
@@ -361,9 +372,10 @@ public class RecordAccumulatorTest {
assertFalse(iter.hasNext(), "No more records");
}
- @Test
- public void testPartialDrain() throws Exception {
- RecordAccumulator accum = createTestRecordAccumulator(
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testPartialDrain(String allocationStrategy) throws Exception {
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, Compression.NONE, 10);
int appends = 1024 / msgSize + 1;
List<TopicPartition> partitions = asList(tp1, tp2);
@@ -378,12 +390,13 @@ public class RecordAccumulatorTest {
}
@SuppressWarnings("unused")
- @Test
- public void testStressfulSituation() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testStressfulSituation(String allocationStrategy) throws Exception {
final int numThreads = 5;
final int msgs = 10000;
final int numParts = 2;
- final RecordAccumulator accum = createTestRecordAccumulator(
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, Compression.NONE, 0);
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < numThreads; i++) {
@@ -418,16 +431,17 @@ public class RecordAccumulatorTest {
}
- @Test
- public void testNextReadyCheckDelay() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testNextReadyCheckDelay(String allocationStrategy) throws Exception {
// Next check time will use lingerMs since this test won't trigger any retries/backoff
int lingerMs = 10;
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD,
- 10 * batchSize, Compression.NONE, lingerMs);
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD,
+ 10 * batchSize, Compression.NONE, lingerMs);
// Just short of going over the limit so we trigger linger time
int appends = expectedNumAppends(batchSize);
@@ -457,8 +471,9 @@ public class RecordAccumulatorTest {
assertTrue(result.nextReadyCheckDelayMs <= lingerMs, "Next check time should be defined by node2, at most linger time");
}
- @Test
- public void testRetryBackoff() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testRetryBackoff(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE / 16;
long retryBackoffMs = Integer.MAX_VALUE / 8;
long retryBackoffMaxMs = retryBackoffMs * 10;
@@ -467,10 +482,8 @@ public class RecordAccumulatorTest {
int batchSize = 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
String metricGrpName = "producer-metrics";
- final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
- deliveryTimeoutMs, metrics, metricGrpName, time, null,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize,
+ lingerMs, retryBackoffMs, retryBackoffMaxMs, deliveryTimeoutMs, metricGrpName);
long now = time.milliseconds();
accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), cluster);
@@ -522,8 +535,9 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testExponentialRetryBackoff() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExponentialRetryBackoff(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE / 16;
long retryBackoffMs = 100;
long retryBackoffMaxMs = 1000;
@@ -532,10 +546,8 @@ public class RecordAccumulatorTest {
int batchSize = 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
String metricGrpName = "producer-metrics";
- final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
- deliveryTimeoutMs, metrics, metricGrpName, time, null,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize,
+ lingerMs, retryBackoffMs, retryBackoffMaxMs, deliveryTimeoutMs, metricGrpName);
long now = time.milliseconds();
long initial = now;
@@ -560,8 +572,9 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testExponentialRetryBackoffLeaderChange() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExponentialRetryBackoffLeaderChange(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE / 16;
long retryBackoffMs = 100;
long retryBackoffMaxMs = 1000;
@@ -593,10 +606,8 @@ public class RecordAccumulatorTest {
null,
Collections.emptyMap());
- final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
- deliveryTimeoutMs, metrics, metricGrpName, time, null,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize,
+ lingerMs, retryBackoffMs, retryBackoffMaxMs, deliveryTimeoutMs, metricGrpName);
long now = time.milliseconds();
long initial = now;
@@ -650,10 +661,11 @@ public class RecordAccumulatorTest {
drainAndCheckBatchAmount(metadataCacheChange, node2, accum, initial + upperBoundBackoffMs + 1, 1);
}
- @Test
- public void testFlush() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testFlush(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE;
- final RecordAccumulator accum = createTestRecordAccumulator(
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
4 * 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, Compression.NONE, lingerMs);
for (int i = 0; i < 100; i++) {
@@ -689,9 +701,10 @@ public class RecordAccumulatorTest {
t.start();
}
- @Test
- public void testAwaitFlushComplete() throws Exception {
- RecordAccumulator accum = createTestRecordAccumulator(
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAwaitFlushComplete(String allocationStrategy) throws Exception {
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
4 * 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, Compression.NONE, Integer.MAX_VALUE);
accum.append(topic, 0, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), cluster);
@@ -702,13 +715,14 @@ public class RecordAccumulatorTest {
assertFalse(accum.flushInProgress(), "flushInProgress count should be decremented even if thread is interrupted");
}
- @Test
- public void testAbortIncompleteBatches() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAbortIncompleteBatches(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE;
int numRecords = 100;
final AtomicInteger numExceptionReceivedInCallback = new AtomicInteger(0);
- final RecordAccumulator accum = createTestRecordAccumulator(
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
128 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, Compression.NONE, lingerMs);
class TestCallback implements RecordAccumulator.AppendCallbacks {
@Override
@@ -745,13 +759,14 @@ public class RecordAccumulatorTest {
assertFalse(accum.hasIncomplete());
}
- @Test
- public void testAbortUnsentBatches() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAbortUnsentBatches(String allocationStrategy) throws Exception {
int lingerMs = Integer.MAX_VALUE;
int numRecords = 100;
final AtomicInteger numExceptionReceivedInCallback = new AtomicInteger(0);
- final RecordAccumulator accum = createTestRecordAccumulator(
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
128 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, Compression.NONE, lingerMs);
final KafkaException cause = new KafkaException();
@@ -792,14 +807,14 @@ public class RecordAccumulatorTest {
assertTrue(accum.hasIncomplete());
}
- private void doExpireBatchSingle(int deliveryTimeoutMs) throws InterruptedException {
+ private void doExpireBatchSingle(String allocationStrategy, int deliveryTimeoutMs) throws InterruptedException {
int lingerMs = 300;
List<Boolean> muteStates = Arrays.asList(false, true);
Set<Node> readyNodes;
List<ProducerBatch> expiredBatches;
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(deliveryTimeoutMs,
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, deliveryTimeoutMs,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, Compression.NONE, lingerMs);
// Make the batches ready due to linger. These batches are not in retry
@@ -829,18 +844,21 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testExpiredBatchSingle() throws InterruptedException {
- doExpireBatchSingle(3200);
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExpiredBatchSingle(String allocationStrategy) throws InterruptedException {
+ doExpireBatchSingle(allocationStrategy, 3200);
}
- @Test
- public void testExpiredBatchSingleMaxValue() throws InterruptedException {
- doExpireBatchSingle(Integer.MAX_VALUE);
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExpiredBatchSingleMaxValue(String allocationStrategy) throws InterruptedException {
+ doExpireBatchSingle(allocationStrategy, Integer.MAX_VALUE);
}
- @Test
- public void testExpiredBatches() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExpiredBatches(String allocationStrategy) throws InterruptedException {
long retryBackoffMs = 100L;
int lingerMs = 30;
int requestTimeout = 60;
@@ -849,7 +867,7 @@ public class RecordAccumulatorTest {
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
deliveryTimeoutMs, batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, Compression.NONE, lingerMs);
int appends = expectedNumAppends(batchSize);
@@ -936,13 +954,14 @@ public class RecordAccumulatorTest {
assertEquals(1, accum.ready(metadataCache, time.milliseconds()).readyNodes.size(), "No partitions should be ready.");
}
- @Test
- public void testMutedPartitions() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testMutedPartitions(String allocationStrategy) throws InterruptedException {
long now = time.milliseconds();
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, Compression.NONE, 10);
int appends = expectedNumAppends(batchSize);
for (int i = 0; i < appends; i++) {
@@ -972,16 +991,17 @@ public class RecordAccumulatorTest {
assertFalse(drained.get(node1.id()).isEmpty(), "The batch should have been drained.");
}
- @Test
- public void testRecordsDrainedWhenTransactionCompleting() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testRecordsDrainedWhenTransactionCompleting(String allocationStrategy) throws Exception {
int batchSize = 1025;
int deliveryTimeoutMs = 3200;
int lingerMs = 10;
long totalSize = 10 * batchSize;
TransactionManager transactionManager = Mockito.mock(TransactionManager.class);
- RecordAccumulator accumulator = createTestRecordAccumulator(transactionManager, deliveryTimeoutMs,
- batchSize, totalSize, Compression.NONE, lingerMs);
+ RecordAccumulator accumulator = createTestRecordAccumulator(allocationStrategy, transactionManager,
+ deliveryTimeoutMs, batchSize, totalSize, Compression.NONE, lingerMs);
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(12345L, (short) 5);
Mockito.when(transactionManager.producerIdAndEpoch()).thenReturn(producerIdAndEpoch);
@@ -1021,7 +1041,7 @@ public class RecordAccumulatorTest {
@Test
public void testSplitAndReenqueue() throws ExecutionException, InterruptedException {
long now = time.milliseconds();
- RecordAccumulator accum = createTestRecordAccumulator(1024, 10 * 1024, Compression.gzip().build(), 10);
+ RecordAccumulator accum = createTestRecordAccumulator(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, 1024, 10 * 1024, Compression.gzip().build(), 10);
// Create a big batch
ByteBuffer buffer = ByteBuffer.allocate(4096);
@@ -1111,7 +1131,7 @@ public class RecordAccumulatorTest {
// First set the compression ratio estimation to be good.
CompressionRatioEstimator.setEstimation(tp1.topic(), CompressionType.GZIP, 0.1f);
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, bufferCapacity, Compression.gzip().build(), 0);
+ RecordAccumulator accum = createTestRecordAccumulator(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, batchSize, bufferCapacity, Compression.gzip().build(), 0);
int numSplitBatches = prepareSplitBatches(accum, seed, 100, 20);
assertTrue(numSplitBatches > 0, "There should be some split batches");
// Drain all the split batches.
@@ -1135,7 +1155,7 @@ public class RecordAccumulatorTest {
final int batchSize = 1024;
final int numMessages = 1000;
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, 3 * 1024, Compression.gzip().build(), 10);
+ RecordAccumulator accum = createTestRecordAccumulator(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, batchSize, 3 * 1024, Compression.gzip().build(), 10);
// Adjust the high and low compression ratio message percentage
for (int goodCompRatioPercentage = 1; goodCompRatioPercentage < 100; goodCompRatioPercentage++) {
int numSplit = 0;
@@ -1160,12 +1180,13 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testSoonToExpireBatchesArePickedUpForExpiry() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testSoonToExpireBatchesArePickedUpForExpiry(String allocationStrategy) throws InterruptedException {
int lingerMs = 500;
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, Compression.NONE, lingerMs);
accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, time.milliseconds(), cluster);
@@ -1192,8 +1213,9 @@ public class RecordAccumulatorTest {
assertEquals(1, drained.size(), "A batch did not drain after linger");
}
- @Test
- public void testExpiredBatchesRetry() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testExpiredBatchesRetry(String allocationStrategy) throws InterruptedException {
int lingerMs = 3000;
int rtt = 1000;
int deliveryTimeoutMs = 3200;
@@ -1203,7 +1225,7 @@ public class RecordAccumulatorTest {
// test case assumes that the records do not fill the batch completely
int batchSize = 1025;
- RecordAccumulator accum = createTestRecordAccumulator(
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy,
batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, Compression.NONE, lingerMs);
// Test batches in retry.
@@ -1230,14 +1252,15 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testUniformBuiltInPartitioner() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testUniformBuiltInPartitioner(String allocationStrategy) throws Exception {
mockRandom = new AtomicInteger();
long totalSize = 1024 * 1024;
int batchSize = 1024; // note that this is also a "sticky" limit for the partitioner
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, 0);
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize, Compression.NONE, 0);
// Set up callbacks so that we know what partition is chosen.
final AtomicInteger partition = new AtomicInteger(RecordMetadata.UNKNOWN_PARTITION);
@@ -1296,8 +1319,9 @@ public class RecordAccumulatorTest {
assertEquals(4, mockRandom.get());
}
- @Test
- public void testAdaptiveBuiltInPartitioner() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testAdaptiveBuiltInPartitioner(String allocationStrategy) throws Exception {
// Mock random number generator with just sequential integer.
mockRandom = new AtomicInteger();
@@ -1305,15 +1329,8 @@ public class RecordAccumulatorTest {
RecordAccumulator.PartitionerConfig config = new RecordAccumulator.PartitionerConfig(true, 100, false, "");
long totalSize = 1024 * 1024;
int batchSize = 128;
- RecordAccumulator accum = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
- 3200, config, metrics, "producer-metrics", time, null,
- new BufferPool(totalSize, batchSize, metrics, time, "producer-internal-metrics")) {
- @Override
- BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
- int stickyBatchSize, boolean rackAware, String rack) {
- return new SequentialPartitioner(logContext, topic, stickyBatchSize, rackAware, rack);
- }
- };
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, null, 3200, batchSize, totalSize,
+ Compression.NONE, 0, config);
byte[] largeValue = new byte[batchSize];
int[] queueSizes = {1, 7, 2};
@@ -1387,14 +1404,15 @@ public class RecordAccumulatorTest {
}
- @Test
- public void testBuiltInPartitionerFractionalBatches() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testBuiltInPartitionerFractionalBatches(String allocationStrategy) throws Exception {
// Test how we avoid creating fractional batches with high linger.ms (see
// BuiltInPartitioner.updatePartitionInfo).
long totalSize = 1024 * 1024;
int batchSize = 512; // note that this is also a "sticky" limit for the partitioner
int valSize = 32;
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, 10);
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize, Compression.NONE, 10);
byte[] value = new byte[valSize];
for (int c = 10; c-- > 0; ) {
@@ -1423,8 +1441,9 @@ public class RecordAccumulatorTest {
* 1. how long it has waited between retry attempts.
* 2. change in leader hosting the partition.
*/
- @Test
- public void testReadyAndDrainWhenABatchIsBeingRetried() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testReadyAndDrainWhenABatchIsBeingRetried(String allocationStrategy) throws InterruptedException {
int part1LeaderEpoch = 100;
// Create cluster metadata, partition1 being hosted by node1
PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()), Optional.of(part1LeaderEpoch), null, null, null);
@@ -1437,10 +1456,8 @@ public class RecordAccumulatorTest {
int deliveryTimeoutMs = Integer.MAX_VALUE;
long totalSize = 10 * 1024;
String metricGrpName = "producer-metrics";
- final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
- deliveryTimeoutMs, metrics, metricGrpName, time, null,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ final RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize,
+ lingerMs, retryBackoffMs, retryBackoffMaxMs, deliveryTimeoutMs, metricGrpName);
// Create 1 batch(batchA) to be produced to partition1.
long now = time.milliseconds();
@@ -1538,12 +1555,13 @@ public class RecordAccumulatorTest {
}
}
- @Test
- public void testDrainWithANodeThatDoesntHostAnyPartitions() {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testDrainWithANodeThatDoesntHostAnyPartitions(String allocationStrategy) {
int batchSize = 10;
int lingerMs = 10;
long totalSize = 10 * 1024;
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, lingerMs);
+ RecordAccumulator accum = createTestRecordAccumulator(allocationStrategy, batchSize, totalSize, Compression.NONE, lingerMs);
// Create cluster metadata, node2 doesn't host any partitions.
PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()), Optional.empty(), null, null, null);
@@ -1557,7 +1575,7 @@ public class RecordAccumulatorTest {
@Test
public void testRetryPolicy() {
- RecordAccumulator accum = createTestRecordAccumulator(1024, 10 * 1024, Compression.NONE, 0);
+ RecordAccumulator accum = createTestRecordAccumulator(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, 1024, 10 * 1024, Compression.NONE, 0);
try {
long spent = time.milliseconds(); // a deadline that is up: reached, so no time is left
long open = time.milliseconds() + 1000; // and one that is not
@@ -1697,29 +1715,79 @@ public class RecordAccumulatorTest {
}
}
- private RecordAccumulator createTestRecordAccumulator(int batchSize, long totalSize, Compression compression, int lingerMs) {
+ private static Stream<String> allocationStrategies() {
+ return Stream.of(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL);
+ }
+
+ private BufferPool createTestBufferPool(String allocationStrategy, long totalSize, int batchSize, String metricGrpName) {
+ return allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)
+ ? new BufferPool(totalSize, 128, metrics, time, metricGrpName, BufferPool.AllocationMode.INCREMENTAL)
+ : new BufferPool(totalSize, batchSize, metrics, time, metricGrpName);
+ }
+
+ private RecordAccumulator createTestRecordAccumulator(String allocationStrategy, int batchSize, long totalSize, Compression compression, int lingerMs) {
int deliveryTimeoutMs = 3200;
- return createTestRecordAccumulator(deliveryTimeoutMs, batchSize, totalSize, compression, lingerMs);
+ return createTestRecordAccumulator(allocationStrategy, deliveryTimeoutMs, batchSize, totalSize, compression, lingerMs);
}
- private RecordAccumulator createTestRecordAccumulator(int deliveryTimeoutMs, int batchSize, long totalSize, Compression compression, int lingerMs) {
- return createTestRecordAccumulator(null, deliveryTimeoutMs, batchSize, totalSize, compression, lingerMs);
+ private RecordAccumulator createTestRecordAccumulator(String allocationStrategy, int deliveryTimeoutMs, int batchSize, long totalSize, Compression compression, int lingerMs) {
+ return createTestRecordAccumulator(allocationStrategy, null, deliveryTimeoutMs, batchSize, totalSize, compression, lingerMs);
}
/**
- * Return a test RecordAccumulator instance
+ * Return a test RecordAccumulator instance serving the given allocation strategy, with the
+ * built-in partitioner replaced by a sequential one.
*/
private RecordAccumulator createTestRecordAccumulator(
+ String allocationStrategy,
TransactionManager txnManager,
int deliveryTimeoutMs,
int batchSize,
long totalSize,
Compression compression,
int lingerMs
+ ) {
+ return createTestRecordAccumulator(allocationStrategy, txnManager, deliveryTimeoutMs, batchSize, totalSize,
+ compression, lingerMs, new RecordAccumulator.PartitionerConfig());
+ }
+
+ private RecordAccumulator createTestRecordAccumulator(
+ String allocationStrategy,
+ TransactionManager txnManager,
+ int deliveryTimeoutMs,
+ int batchSize,
+ long totalSize,
+ Compression compression,
+ int lingerMs,
+ RecordAccumulator.PartitionerConfig partitionerConfig
) {
long retryBackoffMs = 100L;
long retryBackoffMaxMs = 1000L;
String metricGrpName = "producer-metrics";
+ BufferPool pool = createTestBufferPool(allocationStrategy, totalSize, batchSize, metricGrpName);
+
+ if (allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)) {
+ return new ChunkedRecordAccumulator(
+ logContext,
+ batchSize,
+ compression,
+ lingerMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ deliveryTimeoutMs,
+ partitionerConfig,
+ metrics,
+ metricGrpName,
+ time,
+ txnManager,
+ pool) {
+ @Override
+ BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
+ int stickyBatchSize, boolean rackAware, String rack) {
+ return new SequentialPartitioner(logContext, topic, stickyBatchSize, rackAware, rack);
+ }
+ };
+ }
return new RecordAccumulator(
logContext,
@@ -1729,11 +1797,12 @@ public class RecordAccumulatorTest {
retryBackoffMs,
retryBackoffMaxMs,
deliveryTimeoutMs,
+ partitionerConfig,
metrics,
metricGrpName,
time,
txnManager,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
+ pool) {
@Override
BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
int stickyBatchSize, boolean rackAware, String rack) {
@@ -1742,6 +1811,21 @@ public class RecordAccumulatorTest {
};
}
+ /**
+ * Return an uncompressed test RecordAccumulator instance serving the given allocation strategy,
+ * with an explicit retry/timeout configuration and the production built-in partitioner.
+ */
+ private RecordAccumulator createTestRecordAccumulator(String allocationStrategy, int batchSize, long totalSize,
+ int lingerMs, long retryBackoffMs, long retryBackoffMaxMs,
+ int deliveryTimeoutMs, String metricGrpName) {
+ BufferPool pool = createTestBufferPool(allocationStrategy, totalSize, batchSize, metricGrpName);
+ return allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)
+ ? new ChunkedRecordAccumulator(logContext, batchSize, Compression.NONE, lingerMs, retryBackoffMs,
+ retryBackoffMaxMs, deliveryTimeoutMs, metrics, metricGrpName, time, null, pool)
+ : new RecordAccumulator(logContext, batchSize, Compression.NONE, lingerMs, retryBackoffMs,
+ retryBackoffMaxMs, deliveryTimeoutMs, metrics, metricGrpName, time, null, pool);
+ }
+
private class SequentialPartitioner extends BuiltInPartitioner {
public SequentialPartitioner(LogContext logContext, String topic, int stickyBatchSize, boolean rackAware, String rack) {
@@ -1768,8 +1852,8 @@ public class RecordAccumulatorTest {
// Initialize test environment with a large batch size
long now = time.milliseconds();
int batchSize = 1024 * 1024; // 1MB batch size
- RecordAccumulator accum = createTestRecordAccumulator(batchSize, 10 * batchSize, Compression.gzip().build(),
- 10);
+ RecordAccumulator accum = createTestRecordAccumulator(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, batchSize,
+ 10 * batchSize, Compression.gzip().build(), 10);
// Create a large producer batch manually (bypassing the accumulator's normal
// append process)
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
index cb3e56f2b08..35acdad3d67 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
@@ -26,6 +26,7 @@ import org.apache.kafka.clients.MetadataSnapshot;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.InvalidRecordException;
@@ -98,6 +99,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.InOrder;
import java.nio.ByteBuffer;
@@ -119,7 +121,10 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+import static org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL;
+import static org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL;
import static org.apache.kafka.clients.producer.internals.ProducerTestUtils.runUntil;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -546,16 +551,16 @@ public class SenderTest {
}
}
- @Test
- public void testNodeLatencyStats() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testNodeLatencyStats(String allocationStrategy) throws Exception {
try (Metrics m = new Metrics()) {
// Create a new record accumulator with non-0 partitionAvailabilityTimeoutMs
// otherwise it wouldn't update the stats.
RecordAccumulator.PartitionerConfig config = new RecordAccumulator.PartitionerConfig(false, 42, false, "");
long totalSize = 1024 * 1024;
- accumulator = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
- DELIVERY_TIMEOUT_MS, config, m, "producer-metrics", time, null,
- new BufferPool(totalSize, batchSize, m, time, "producer-internal-metrics"));
+ accumulator = createAccumulator(allocationStrategy, config, 0L, 0L, m, "producer-metrics",
+ createBufferPool(allocationStrategy, totalSize, m, "producer-internal-metrics"));
SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
apiVersions.update("0", NodeApiVersions.create(ApiKeys.PRODUCE.id, ApiKeys.PRODUCE.oldestVersion(), ApiKeys.PRODUCE.latestVersion()));
@@ -3385,21 +3390,19 @@ public class SenderTest {
assertEquals(KafkaException.class, abortResult.error().getClass());
}
- @Test
- public void testProducerBatchRetriesWhenPartitionLeaderChanges() throws Exception {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testProducerBatchRetriesWhenPartitionLeaderChanges(String allocationStrategy) throws Exception {
Metrics m = new Metrics();
SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
try {
// SETUP
String metricGrpName = "producer-metrics-test-stats-1";
long totalSize = 1024 * 1024;
- BufferPool pool = new BufferPool(totalSize, batchSize, metrics, time,
- metricGrpName);
+ BufferPool pool = createBufferPool(allocationStrategy, totalSize, metrics, metricGrpName);
long retryBackoffMaxMs = 100L;
- // lingerMs is 0 to send batch as soon as any records are available on it.
- this.accumulator = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, 0, 10L, retryBackoffMaxMs,
- DELIVERY_TIMEOUT_MS, metrics, metricGrpName, time, null, pool);
+ this.accumulator = createAccumulator(allocationStrategy, new RecordAccumulator.PartitionerConfig(),
+ 10L, retryBackoffMaxMs, metrics, metricGrpName, pool);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false,
MAX_REQUEST_SIZE, ACKS_ALL,
10, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null);
@@ -3498,8 +3501,9 @@ public class SenderTest {
* Test the scenario that FetchResponse returns NOT_LEADER_OR_FOLLOWER, indicating change in leadership, but it
* does not contain new leader info(defined in KIP-951).
*/
- @Test
- public void testWhenProduceResponseReturnsWithALeaderShipChangeErrorButNoNewLeaderInformation()
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testWhenProduceResponseReturnsWithALeaderShipChangeErrorButNoNewLeaderInformation(String allocationStrategy)
throws InterruptedException {
// Setup 3 partitions, tp0 & tp1 return with NOT_LEADER_OR_FOLLOWER, tp2 doesn't return an error.
Metrics m = new Metrics();
@@ -3508,13 +3512,10 @@ public class SenderTest {
// SETUP
String metricGrpName = "producer-metrics-test-stats-1";
long totalSize = 1024 * 1024;
- BufferPool pool = new BufferPool(totalSize, batchSize, metrics, time,
- metricGrpName);
+ BufferPool pool = createBufferPool(allocationStrategy, totalSize, metrics, metricGrpName);
long retryBackoffMaxMs = 100L;
- // lingerMs is 0 to send batch as soon as any records are available on it.
- this.accumulator = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, 0, 10L, retryBackoffMaxMs,
- DELIVERY_TIMEOUT_MS, metrics, metricGrpName, time, null, pool);
+ this.accumulator = createAccumulator(allocationStrategy, new RecordAccumulator.PartitionerConfig(),
+ 10L, retryBackoffMaxMs, metrics, metricGrpName, pool);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false,
MAX_REQUEST_SIZE, ACKS_ALL,
10, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null);
@@ -3578,8 +3579,9 @@ public class SenderTest {
* Test the scenario that FetchResponse returns NOT_LEADER_OR_FOLLOWER, indicating change in leadership, along with
* new leader info(defined in KIP-951).
*/
- @Test
- public void testWhenProduceResponseReturnsWithALeaderShipChangeErrorAndNewLeaderInformation()
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testWhenProduceResponseReturnsWithALeaderShipChangeErrorAndNewLeaderInformation(String allocationStrategy)
throws InterruptedException {
// Setup 3 partitions, tp0 & tp1 return with NOT_LEADER_OR_FOLLOWER, tp2 doesn't return an error.
Metrics m = new Metrics();
@@ -3588,13 +3590,10 @@ public class SenderTest {
// SETUP
String metricGrpName = "producer-metrics-test-stats-1";
long totalSize = 1024 * 1024;
- BufferPool pool = new BufferPool(totalSize, batchSize, metrics, time,
- metricGrpName);
+ BufferPool pool = createBufferPool(allocationStrategy, totalSize, metrics, metricGrpName);
long retryBackoffMaxMs = 100L;
- // lingerMs is 0 to send batch as soon as any records are available on it.
- this.accumulator = new RecordAccumulator(logContext, batchSize,
- Compression.NONE, 0, 10L, retryBackoffMaxMs,
- DELIVERY_TIMEOUT_MS, metrics, metricGrpName, time, null, pool);
+ this.accumulator = createAccumulator(allocationStrategy, new RecordAccumulator.PartitionerConfig(),
+ 10L, retryBackoffMaxMs, metrics, metricGrpName, pool);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false,
MAX_REQUEST_SIZE, ACKS_ALL,
10, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null);
@@ -3895,6 +3894,26 @@ public class SenderTest {
return produceResponse(tp, offset, error, throttleTimeMs, -1L, null);
}
+ private static Stream<String> allocationStrategies() {
+ return Stream.of(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL, BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL);
+ }
+
+ private BufferPool createBufferPool(String allocationStrategy, long totalSize, Metrics metrics, String metricGrpName) {
+ return allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)
+ ? new BufferPool(totalSize, 128, metrics, time, metricGrpName, BufferPool.AllocationMode.INCREMENTAL)
+ : new BufferPool(totalSize, batchSize, metrics, time, metricGrpName);
+ }
+
+ private RecordAccumulator createAccumulator(String allocationStrategy, RecordAccumulator.PartitionerConfig partitionerConfig,
+ long retryBackoffMs, long retryBackoffMaxMs, Metrics metrics,
+ String metricGrpName, BufferPool pool) {
+ return allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)
+ ? new ChunkedRecordAccumulator(logContext, batchSize, Compression.NONE, 0, retryBackoffMs, retryBackoffMaxMs,
+ DELIVERY_TIMEOUT_MS, partitionerConfig, metrics, metricGrpName, time, null, pool)
+ : new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, retryBackoffMs, retryBackoffMaxMs,
+ DELIVERY_TIMEOUT_MS, partitionerConfig, metrics, metricGrpName, time, null, pool);
+ }
+
private TransactionManager createTransactionManager() {
return new TransactionManager(new LogContext(), null, 0, RETRY_BACKOFF_MS, new ApiVersions(), metadata, false);
}