Skip to content

test: harden bounded Kafka supervisor timeouts #20124

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

Open
FrankChen021 wants to merge 5 commits into
base: master
Choose a base branch
from
+18 −3
Open

test: harden bounded Kafka supervisor timeouts #20124

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
164a3cf
test: harden bounded Kafka supervisor timeouts
FrankChen021 Aug 21, 2026
73c0c30
test: reduce bounded Kafka segment rollovers
FrankChen021 Aug 21, 2026
645c216
test: apply bounded segment limit to all supervisors
FrankChen021 Aug 21, 2026
7c309cf
docs: explain bounded supervisor test constants
FrankChen021 Aug 22, 2026
ce7a909
test: inline bounded segment size
FrankChen021 Aug 22, 2026
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
21 changes: 18 additions & 3 deletions .../src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
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;
Expand All @@ -39,6 +40,10 @@
*/
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
Expand All @@ -47,6 +52,16 @@ protected StreamIngestResource<?> getStreamIngestResource()
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()
{
Expand Down Expand Up @@ -89,7 +104,7 @@ public void test_boundedSupervisor_ingestsDataAndCompletes()

// 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());
Expand Down Expand Up @@ -203,7 +218,7 @@ public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
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());
Expand Down Expand Up @@ -265,7 +280,7 @@ public void test_boundedSupervisor_doesNotSilentlyCompleteWhenStaleOffsetExceeds
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());
Expand Down
Loading