Skip to content

Commit 5568480

Browse files
MINOR: Fix testRackAwareAssignment flake
The last part of testRackAwareAssignment was found to be flaky. This part moves all topic partitions to different racks and waits for consumer assignments to settle. Each of the three consumers is expected to revoke all its partitions and be assigned partitions previously held by another within a 15 second timeout. This timeout is not always sufficient. The consumer heartbeat interval is left at the default of 5,000 ms and each consumer polls every 3,000 ms. In the worst case, it takes a consumer around 7,000 ms to reconcile an assignment change. An additional 3,000 ms round of polling may be required when a consumer needs to auto-commit offsets. Two rounds of reconciliation must happen within 15,000 ms. The timeline of an example failing run looks like: -02.956 Group coordinator computes target assignment at epoch=6 consumer0=[0] consumer1=[1, 2] consumer2=[3, 4, 5] +00.000 15 second timeout starts +03.179 consumer0 heartbeats This is the first heartbeat since the rack reassignments. +03.179 Group coordinator computes target assignment at epoch=7 consumer0=[5] consumer1=[3, 4] consumer2=[0, 1, 2] +03.186 consumer0 heartbeat receives assignment [] +04.151 consumer1 starts poll() +04.877 consumer1 heartbeats +04.878 consumer1 heartbeat receives assignment [] +05.155 consumer1 ends poll() +07.259 consumer1 starts poll() +07.259 consumer1 sends auto-commit with offsets for [1, 2] +07.288 consumer1 receives auto-commit response +08.263 consumer1 ends poll() +10.379 consumer1 starts poll() +10.379 consumer1 calls onPartitionsRevoked with [1, 2] +10.379 consumer1 calls onPartitionsAssigned with [] +10.382 consumer1 heartbeats with owned partitions [] +10.387 consumer1 heartbeat receives assignment [3, 4] +10.483 consumer1 calls onPartitionsAssigned [3, 4] +10.483 consumer1 heartbeats with owned partitions [3, 4] +11.384 consumer1 ends poll() +15.000 15 second timeout elapses and the test fails +15.300 consumer2 heartbeat receives assignment [0, 1, 2]
1 parent 0dad6a7 commit 5568480

1 file changed

Lines changed: 1 addition & 1 deletion

File tree

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerIntegrationTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -352,7 +352,7 @@ public void testRackAwareAssignment(ClusterInstance clusterInstance) throws Exec
352352
return consumer0.assignment().equals(Set.of(new TopicPartition(topic, 5))) &&
353353
consumer1.assignment().equals(Set.of(new TopicPartition(topic, 3), new TopicPartition(topic, 4))) &&
354354
consumer2.assignment().equals(Set.of(new TopicPartition(topic, 0), new TopicPartition(topic, 1), new TopicPartition(topic, 2)));
355-
}, "Consumer with topic partition mapping should be 0 -> 5 | 1 -> 3, 4 | 2 -> 0, 1, 2");
355+
}, 30000, "Consumer with topic partition mapping should be 0 -> 5 | 1 -> 3, 4 | 2 -> 0, 1, 2");
356356
}
357357
}
358358

0 commit comments

Comments
 (0)