This is an automated email from the ASF dual-hosted git repository.
kfaraz 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 004f8eb9f40 test: Fixes for flaky StreamIndexFaultToleranceTest tests.
(#19749)
004f8eb9f40 is described below
commit 004f8eb9f4046f34006666b07d2fa1d203b341e3
Author: Gian Merlino <[email protected]>
AuthorDate: Sat Jul 25 00:29:36 2026 -0700
test: Fixes for flaky StreamIndexFaultToleranceTest tests. (#19749)
Two fixes:
1. StreamIndexTestBase: After each test case, cancel any active tasks
and wait for them to exit. Prevents leftover tasks from interfering
with the next task.
2. StreamIndexFaultToleranceTest: In the early handoff test, wait for
pre-handoff tasks to exit before checking that the handoff completed
successfully.
---
.../indexing/StreamIndexFaultToleranceTest.java | 11 ++++++
.../embedded/indexing/StreamIndexTestBase.java | 43 ++++++++++++++++++++++
2 files changed, 54 insertions(+)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexFaultToleranceTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexFaultToleranceTest.java
index c315969e5f5..2c6a3dd7310 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexFaultToleranceTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexFaultToleranceTest.java
@@ -26,12 +26,14 @@ import org.apache.druid.query.DruidMetrics;
import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.hamcrest.Matchers;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -166,6 +168,15 @@ public abstract class StreamIndexFaultToleranceTest
extends StreamIndexTestBase
.hasDimension("noticeType", "handoff_task_group_notice")
);
+ // Wait for the pre-handoff tasks to actually shut down, since otherwise
getRunningTaskIds() below can
+ // still observe them as running.
+ final Set<Object> shutDownTaskIds = new HashSet<>(taskIdsBeforeHandoff);
+ overlord.latchableEmitter().waitForEventAggregate(
+ event -> event.hasMetricName("task/run/time")
+ .hasDimensionMatching(DruidMetrics.TASK_ID,
Matchers.in(shutDownTaskIds)),
+ agg -> agg.hasCountAtLeast(taskIdsBeforeHandoff.size())
+ );
+
totalRecords += publish1kRecords(topic, useTransactions);
waitUntilPublishedRecordsAreIngested(totalRecords);
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexTestBase.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexTestBase.java
index 2c31e09284f..836fe01f542 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexTestBase.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexTestBase.java
@@ -19,9 +19,11 @@
package org.apache.druid.testing.embedded.indexing;
+import com.google.common.collect.ImmutableList;
import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.JsonInputFormat;
import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.TaskStatusPlus;
import org.apache.druid.indexer.granularity.UniformGranularitySpec;
import org.apache.druid.indexing.kafka.simulate.KafkaResource;
import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpecBuilder;
@@ -31,6 +33,7 @@ import
org.apache.druid.indexing.kinesis.supervisor.KinesisSupervisorSpec;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.testing.embedded.EmbeddedBroker;
import org.apache.druid.testing.embedded.EmbeddedCoordinator;
@@ -47,8 +50,10 @@ import
org.apache.druid.testing.embedded.tools.JsonEventSerializer;
import org.apache.druid.testing.embedded.tools.StreamGenerator;
import
org.apache.druid.testing.embedded.tools.WikipediaStreamEventStreamGenerator;
import org.joda.time.Period;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -81,6 +86,27 @@ public abstract class StreamIndexTestBase extends
EmbeddedClusterTestBase
.addServer(new EmbeddedRouter());
}
+ /**
+ * Terminates every supervisor and cancels every active task. The cluster is
shared across all tests
+ * in the class (see {@link EmbeddedClusterTestBase}), so a task left
running by one test would potentially
+ * interfere with the next test.
+ */
+ @AfterEach
+ protected void terminateLeftoverSupervisorsAndTasks()
+ {
+ final List<SupervisorStatus> supervisors = ImmutableList.copyOf(
+ cluster.callApi().onLeaderOverlord(OverlordClient::supervisorStatuses)
+ );
+ for (SupervisorStatus supervisor : supervisors) {
+ cluster.callApi().onLeaderOverlord(o ->
o.terminateSupervisor(supervisor.getId()));
+ }
+
+ cluster.callApi()
+ .waitForResult(this::cancelAllTasks, count -> count == 0)
+ .withTimeoutMillis(60_000)
+ .go();
+ }
+
protected KafkaSupervisorSpecBuilder createKafkaSupervisor(KafkaResource
kafkaServer)
{
return MoreResources.Supervisor.KAFKA_JSON
@@ -207,4 +233,21 @@ public abstract class StreamIndexTestBase extends
EmbeddedClusterTestBase
}
return records.size();
}
+
+ /**
+ * Cancels every task that has not completed yet and returns how many such
tasks there were.
+ */
+ private int cancelAllTasks()
+ {
+ final List<TaskStatusPlus> tasks = new ArrayList<>();
+ for (String taskState : List.of("running", "pending", "waiting")) {
+ tasks.addAll(cluster.callApi().getTasks(null, taskState));
+ }
+
+ for (TaskStatusPlus task : tasks) {
+ cluster.callApi().onLeaderOverlord(o -> o.cancelTask(task.getId()));
+ }
+
+ return tasks.size();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
