Skip to content

KAFKA-13834: add normal test case #12092

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
showuon merged 17 commits into from
Apr 24, 2022
Merged

KAFKA-13834: add normal test case #12092

Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
9a61e49
bugfix for KAFKA-13834
Apr 19, 2022
8e7964b
KAFKA-13834: fix drain batch starving issue
Apr 19, 2022
3dbbefd
change import * to single class
Apr 20, 2022
e213a79
1：make the nodesDrainIndex as private final
Apr 20, 2022
ff88937
1：remove unchecked identification
Apr 20, 2022
8ec5da3
add more context
Apr 20, 2022
e817088
add space between [,] and next argument.
Apr 20, 2022
d628cd1
fix the infinite loop
Apr 21, 2022
d200801
add the drain index test case
Apr 21, 2022
d6232e9
add the continue case
Apr 21, 2022
159b020
Modify comments
Apr 21, 2022
4c76bd0
Code format change
Apr 21, 2022
cdb1432
change test name to testDrainBatches
Apr 21, 2022
c43d3e9
modify code comments
Apr 21, 2022
ba866d6
change code style
Apr 21, 2022
6f7597b
Merge branch 'apache:trunk' into trunk
ruanliang-hualun Apr 24, 2022
fedab82
KAFKA-13834: add normal test case
Apr 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 19 additions & 7 deletions clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Deque;
import java.util.HashSet;
Expand All @@ -61,6 +62,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static java.util.Arrays.asList;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
Expand Down Expand Up @@ -107,7 +109,7 @@ public void testDrainBatches() throws Exception {
PartitionInfo part4 = new PartitionInfo(topic, partition4, node2, null, null);

long batchSize = value.length + DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
RecordAccumulator accum = createTestRecordAccumulator((int) batchSize, 1024, CompressionType.NONE, 10);
RecordAccumulator accum = createTestRecordAccumulator((int) batchSize, Integer.MAX_VALUE, CompressionType.NONE, 10);
Cluster cluster = new Cluster(null, Arrays.asList(node1, node2), Arrays.asList(part1, part2, part3, part4),
Collections.emptySet(), Collections.emptySet());

Expand Down Expand Up @@ -142,15 +144,25 @@ public void testDrainBatches() throws Exception {
// drain batches from 2 nodes: node1 => tp2, node2 => tp3 (because tp4 is muted)
Map<Integer, List<ProducerBatch>> batches4 = accum.drain(cluster, new HashSet<Node>(Arrays.asList(node1, node2)), (int) batchSize, 0);
verifyTopicPartitionInBatches(batches4, tp2, tp3);

// add record for tp1, tp2, tp3, and unmute tp4
accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds());
accum.append(tp2, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds());
accum.append(tp3, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds());
accum.unmutePartition(tp4);
// set maxSize as a max value, so that the all partitions in 2 nodes should be drained: node1 => [tp1, tp2], node2 => [tp3, tp4]
Map<Integer, List<ProducerBatch>> batches5 = accum.drain(cluster, new HashSet<Node>(Arrays.asList(node1, node2)), Integer.MAX_VALUE, 0);
verifyTopicPartitionInBatches(batches5, tp1, tp2, tp3, tp4);
}

private void verifyTopicPartitionInBatches(Map<Integer, List<ProducerBatch>> batches, TopicPartition... tp) {
assertEquals(tp.length, batches.size());
private void verifyTopicPartitionInBatches(Map<Integer, List<ProducerBatch>> nodeBatches, TopicPartition... tp) {
int allTpBatchCount = nodeBatches.values().stream().flatMap(Collection::stream).collect(Collectors.toList()).size();

@ruanliang-hualun ruanliang-hualun Apr 24, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

allTpBatchCount represent the count of the producerbatch from the nodeBatches map

assertEquals(tp.length, allTpBatchCount);
List<TopicPartition> topicPartitionsInBatch = new ArrayList<TopicPartition>();
for (Map.Entry<Integer, List<ProducerBatch>> entry : batches.entrySet()) {
List<ProducerBatch> batchList = entry.getValue();
assertEquals(1, batchList.size());

@ruanliang-hualun ruanliang-hualun Apr 24, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

because the batcheList.size() can be 1 or 2, remove this assert statement

topicPartitionsInBatch.add(batchList.get(0).topicPartition);
for (Map.Entry<Integer, List<ProducerBatch>> entry : nodeBatches.entrySet()) {
List<ProducerBatch> tpBatchList = entry.getValue();
List<TopicPartition> tpList = tpBatchList.stream().map(producerBatch -> producerBatch.topicPartition).collect(Collectors.toList());
topicPartitionsInBatch.addAll(tpList);
}

for (int i = 0; i < tp.length; i++) {
Expand Down