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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
