lianetm commented on code in PR #23135:
URL: https://github.com/apache/kafka/pull/23135#discussion_r3769608296
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3026,8 +3069,10 @@ public void
testTransitionToAbortableErrorOnBatchExpiry() throws InterruptedExce
assertTrue(transactionManager.hasAbortableError());
}
- @Test
- public void testTransitionToAbortableErrorOnMultipleBatchExpiry() throws
InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testTransitionToAbortableErrorOnMultipleBatchExpiry(String
allocationStrategy) throws InterruptedException {
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3082,8 +3127,10 @@ public void
testTransitionToAbortableErrorOnMultipleBatchExpiry() throws Interru
assertTrue(transactionManager.hasAbortableError());
}
- @Test
- public void testDropCommitOnBatchExpiry() throws InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testDropCommitOnBatchExpiry(String allocationStrategy) throws
InterruptedException {
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3141,8 +3188,10 @@ public void testDropCommitOnBatchExpiry() throws
InterruptedException {
assertFalse(transactionManager.transactionContainsPartition(tp0));
}
- @Test
- public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws
InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testTransitionToFatalErrorWhenRetriedBatchIsExpired(String
allocationStrategy) throws InterruptedException {
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -268,18 +276,19 @@ public void testFull() throws Exception {
@Test
public void testAppendLargeCompressed() throws Exception {
- testAppendLarge(Compression.gzip().build());
+ testAppendLarge(BUFFER_MEMORY_ALLOCATION_STRATEGY_FULL,
Compression.gzip().build());
Review Comment:
let's create a jira to re-enable these compression ones for incremental once
it supports it?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1297,24 +1320,34 @@ public void onCompletion(RecordMetadata metadata,
Exception exception) {
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();
// Create accumulator with partitioner config to enable adaptive
partitioning.
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);
+ BufferPool pool = createTestBufferPool(allocationStrategy, totalSize,
batchSize, "producer-internal-metrics");
+ RecordAccumulator accum =
allocationStrategy.equals(ProducerConfig.BUFFER_MEMORY_ALLOCATION_STRATEGY_INCREMENTAL)
Review Comment:
can we reuse the `createTestRecordAccumulator` here?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -2987,8 +3028,10 @@ public void
resendFailedProduceRequestAfterAbortableError() throws Exception {
assertNotNull(responseFuture.get()); // should throw the exception
which caused the transaction to be aborted.
}
- @Test
- public void testTransitionToAbortableErrorOnBatchExpiry() throws
InterruptedException {
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testTransitionToAbortableErrorOnBatchExpiry(String
allocationStrategy) throws InterruptedException {
Review Comment:
I would expect this test at the transaction mgr level is unaffected by
chunks/allocation stategy so no need to parametrize?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3764,10 +3815,10 @@ public void
testBumpTransactionalEpochOnRecoverableAddOffsetsRequestError() thro
}
@ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testHealthyPartitionRetriesDuringEpochBump(boolean
transactionV2Enabled) throws InterruptedException {
+ @MethodSource("transactionV2AndAllocationStrategies")
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1539,12 +1574,14 @@ public void testReadyAndDrainWhenABatchIsBeingRetried()
throws InterruptedExcept
}
}
- @Test
- public void testDrainWithANodeThatDoesntHostAnyPartitions() {
- int batchSize = 10;
+ @ParameterizedTest
+ @MethodSource("allocationStrategies")
+ public void testDrainWithANodeThatDoesntHostAnyPartitions(String
allocationStrategy) {
+ // Nothing is ever appended here, so batch.size only has to stay above
the chunk size.
Review Comment:
why does it need to be above chunk here? (I would expect it if we were
flowing through the KafkaProducer that ignores incremental if batch < chunk,
but not here at the acc level, so why this batch change?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
