(kafka) branch trunk updated: KAFKA-15964: fix flaky StreamsAssignmentScaleTest (#15485)

[mj...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mjsax 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 d8dd068a626 KAFKA-15964: fix flaky StreamsAssignmentScaleTest (#15485)
d8dd068a626 is described below

commit d8dd068a626dcab538c2b234ffd8799a94b2f0ed
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Thu Mar 7 09:17:52 2024 -0800

    KAFKA-15964: fix flaky StreamsAssignmentScaleTest (#15485)
    
    This PR bumps some timeouts due to slow Jenkins builds.
    
    Reviewers: Bruno Cadonna <br...@confluent.io>
---
 .../internals/StreamsPartitionAssignor.java        |  2 +-
 .../internals/StreamsAssignmentScaleTest.java      | 26 +++++++++++-----------
 2 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
index 3f16838dfc2..fb4e45cdfb1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
@@ -628,7 +628,7 @@ public class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Conf
 
         log.info("{} client nodes and {} consumers participating in this rebalance: \n{}.",
                  clientStates.size(),
-                 clientStates.values().stream().map(ClientState::capacity).reduce(Integer::sum),
+                 clientStates.values().stream().map(ClientState::capacity).reduce(Integer::sum).orElse(0),
                  clientStates.entrySet().stream()
                      .sorted(comparingByKey())
                      .map(entry -> entry.getKey() + ": " + entry.getValue().consumers())
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsAssignmentScaleTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsAssignmentScaleTest.java
index 834b6242194..e14e0333634 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsAssignmentScaleTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsAssignmentScaleTest.java
@@ -68,73 +68,73 @@ import static org.mockito.Mockito.when;
 @Category({IntegrationTest.class})
 @RunWith(MockitoJUnitRunner.StrictStubs.class)
 public class StreamsAssignmentScaleTest {
-    final static long MAX_ASSIGNMENT_DURATION = 60 * 1000L; //each individual assignment should complete within 20s
+    final static long MAX_ASSIGNMENT_DURATION = 120 * 1000L; // we should stay below `max.poll.interval.ms`
     final static String APPLICATION_ID = "streams-assignment-scale-test";
 
     private final Logger log = LoggerFactory.getLogger(StreamsAssignmentScaleTest.class);
 
     /* HighAvailabilityTaskAssignor tests */
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testHighAvailabilityTaskAssignorLargePartitionCount() {
         completeLargeAssignment(6_000, 2, 1, 1, HighAvailabilityTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testHighAvailabilityTaskAssignorLargeNumConsumers() {
         completeLargeAssignment(1_000, 1_000, 1, 1, HighAvailabilityTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testHighAvailabilityTaskAssignorManyStandbys() {
         completeLargeAssignment(1_000, 100, 1, 50, HighAvailabilityTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testHighAvailabilityTaskAssignorManyThreadsPerClient() {
         completeLargeAssignment(1_000, 10, 1000, 1, HighAvailabilityTaskAssignor.class);
     }
 
     /* StickyTaskAssignor tests */
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testStickyTaskAssignorLargePartitionCount() {
         completeLargeAssignment(2_000, 2, 1, 1, StickyTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testStickyTaskAssignorLargeNumConsumers() {
         completeLargeAssignment(1_000, 1_000, 1, 1, StickyTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testStickyTaskAssignorManyStandbys() {
         completeLargeAssignment(1_000, 100, 1, 20, StickyTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testStickyTaskAssignorManyThreadsPerClient() {
         completeLargeAssignment(1_000, 10, 1000, 1, StickyTaskAssignor.class);
     }
 
     /* FallbackPriorTaskAssignor tests */
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testFallbackPriorTaskAssignorLargePartitionCount() {
         completeLargeAssignment(2_000, 2, 1, 1, FallbackPriorTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testFallbackPriorTaskAssignorLargeNumConsumers() {
         completeLargeAssignment(1_000, 1_000, 1, 1, FallbackPriorTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testFallbackPriorTaskAssignorManyStandbys() {
         completeLargeAssignment(1_000, 100, 1, 20, FallbackPriorTaskAssignor.class);
     }
 
-    @Test(timeout = 120 * 1000)
+    @Test(timeout = 300 * 1000)
     public void testFallbackPriorTaskAssignorManyThreadsPerClient() {
         completeLargeAssignment(1_000, 10, 1000, 1, FallbackPriorTaskAssignor.class);
     }