(druid) branch master updated: test: harden bounded Kafka supervisor timeouts (#20124)

[fr...@apache.org]

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

FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git


The following commit(s) were added to refs/heads/master by this push:
     new f98a2cd00ef test: harden bounded Kafka supervisor timeouts (#20124)
f98a2cd00ef is described below

commit f98a2cd00ef1738eb2a982b1a38a2cbcce704091
Author: Frank Chen <fr...@outlook.com>
AuthorDate: Mon Aug 24 16:53:27 2026 +0800

    test: harden bounded Kafka supervisor timeouts (#20124)
    
    * test: harden bounded Kafka supervisor timeouts
    
    * test: reduce bounded Kafka segment rollovers
    
    * test: apply bounded segment limit to all supervisors
    
    * docs: explain bounded supervisor test constants
    
    * test: inline bounded segment size
---
 .../indexing/KafkaBoundedSupervisorTest.java        | 21 ++++++++++++++++++---
 1 file changed, 18 insertions(+), 3 deletions(-)

diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
index ad3a3f3fff1..9a38fe5d2e4 100644
--- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
@@ -23,6 +23,7 @@ import org.apache.druid.common.utils.IdUtils;
 import org.apache.druid.data.input.impl.JsonInputFormat;
 import org.apache.druid.indexing.kafka.simulate.KafkaResource;
 import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpecBuilder;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
 import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
 import org.apache.druid.query.DruidMetrics;
@@ -39,6 +40,10 @@ import java.util.Map;
  */
 public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
 {
+  // Allow two minutes for bounded-supervisor cold start, ingestion, and segment publication on CI.
+  // This is a maximum wait, not a fixed delay; successful waits return as soon as the metric is emitted.
+  private static final long BOUNDED_SUPERVISOR_INGESTION_TIMEOUT_MILLIS = 120_000L;
+
   private final KafkaResource kafkaServer = new KafkaResource();
 
   @Override
@@ -47,6 +52,16 @@ public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
     return kafkaServer;
   }
 
+  @Override
+  protected KafkaSupervisorSpecBuilder createKafkaSupervisor(KafkaResource kafkaServer)
+  {
+    // Use a moderate segment size to avoid the shared fixture's one-row segment rollover overhead while
+    // retaining normal segment publication behavior. These tests assert offsets, row counts, and supervisor
+    // state, not segment count, so 100 does not change their semantics.
+    return super.createKafkaSupervisor(kafkaServer)
+        .withTuningConfig(tuningConfig -> tuningConfig.withMaxRowsPerSegment(100));
+  }
+
   @Override
   protected EmbeddedDruidCluster createCluster()
   {
@@ -89,7 +104,7 @@ public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
 
     // Bounded supervisor cold start (post supervisor -> schedule task -> consume -> publish) can exceed
     // the cluster default wait on CI; give it a generous ceiling.
-    waitUntilPublishedRecordsAreIngested(totalRecords, 120_000L);
+    waitUntilPublishedRecordsAreIngested(totalRecords, BOUNDED_SUPERVISOR_INGESTION_TIMEOUT_MILLIS);
 
     // Wait for supervisor to transition to COMPLETED state
     waitForSupervisorToComplete(supervisor.getId());
@@ -203,7 +218,7 @@ public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
     cluster.callApi().postSupervisor(supervisor1);
 
     // Wait for records to be ingested (approximately 200 records total from both partitions)
-    waitUntilPublishedRecordsAreIngested(200);
+    waitUntilPublishedRecordsAreIngested(200, BOUNDED_SUPERVISOR_INGESTION_TIMEOUT_MILLIS);
 
     // Wait for supervisor to transition to COMPLETED state
     waitForSupervisorToComplete(supervisor1.getId());
@@ -265,7 +280,7 @@ public class KafkaBoundedSupervisorTest extends StreamIndexTestBase
     final KafkaSupervisorSpec supervisor1 = createBoundedKafkaSupervisor(kafkaServer, topic, boundedConfig1);
 
     cluster.callApi().postSupervisor(supervisor1);
-    waitUntilPublishedRecordsAreIngested(250);
+    waitUntilPublishedRecordsAreIngested(250, BOUNDED_SUPERVISOR_INGESTION_TIMEOUT_MILLIS);
     waitForSupervisorToComplete(supervisor1.getId());
 
     final SupervisorStatus status1 = cluster.callApi().getSupervisorStatus(supervisor1.getId());


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org