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 cde7404de1e test: Fix flaky EmbeddedKafkaSupervisorTest (#20519)
cde7404de1e is described below
commit cde7404de1e6257fce19c9990dd7d94332fca10c
Author: Suraj Goel <[email protected]>
AuthorDate: Thu Oct 8 12:28:37 2026 +0530
test: Fix flaky EmbeddedKafkaSupervisorTest (#20519)
Fixes #20518.
Description
-----------
`EmbeddedKafkaSupervisorTest.test_runKafkaSupervisorWithHeaderFiltering`
(added in #18525)
failed its first attempt in 6 of 7 master runs. Each time, it timed out
waiting for `ingest/handoff/count`
to reach 4. It fails only when it runs after the other tests in the class.
The surefire retry runs it alone,
so the retry passed.
Root cause
-----------
The task does read all 10 records. The race is between the test suspending
the supervisor
and the task's own checkpoint:
The fix suggested in the issue (wait for `ingest/events/processed` >= 4
before suspending) makes
the race less likely but does not remove it. `processed` is incremented
during parsing, a few
milliseconds before the checkpoint request is sent, and the metric is
emitted every 100 ms.
Fix
---
- The header-filter supervisor spec is now built from `newKafkaSupervisor()`
- The test now waits for handoff without suspending the supervisor. The
short task duration publishes
the segments. The existing `@AfterEach` teardown then suspends the
supervisor and cancels the
remaining tasks.
---
.../simulate/EmbeddedKafkaSupervisorTest.java | 25 +++++-----------------
1 file changed, 5 insertions(+), 20 deletions(-)
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
index 6afb70a79a4..ea0a3ca49ff 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
@@ -215,8 +215,9 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
Assertions.assertTrue(supervisorStatus.isHealthy());
Assertions.assertEquals("RUNNING", supervisorStatus.getState());
- // Suspend the supervisor and wait for segment handoff
-
cluster.callApi().postSupervisor(kafkaSupervisorSpec.createSuspendedSpec());
+ // Wait for segment handoff without suspending the supervisor; its short
task duration publishes the segments.
+ // Suspending while ingestion is in progress can replace the supervisor
while the task's checkpoint request is in
+ // flight, which fails the task before it publishes anything.
indexer.latchableEmitter().waitForEventAggregate(
event -> event.hasMetricName("ingest/handoff/count")
.hasDimension(DruidMetrics.DATASOURCE, dataSource),
@@ -345,24 +346,8 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
InDimFilter filter = new InDimFilter("environment",
ImmutableSet.of("production"));
KafkaHeaderBasedFilterConfig headerFilterConfig = new
KafkaHeaderBasedFilterConfig(filter, "UTF-8", 1000);
- return new KafkaSupervisorSpecBuilder()
- .withDataSchema(
- schema -> schema
- .withTimestamp(new TimestampSpec("timestamp", null, null))
- .withDimensions(DimensionsSpec.EMPTY)
- )
- .withTuningConfig(
- tuningConfig -> tuningConfig
- .withMaxRowsPerSegment(1)
- .withReleaseLocksOnHandoff(true)
- )
- .withIoConfig(
- ioConfig -> ioConfig
- .withInputFormat(new CsvInputFormat(List.of("timestamp",
"item"), null, null, false, 0, false))
- .withConsumerProperties(kafkaServer.consumerProperties())
- .withUseEarliestSequenceNumber(true)
- .withHeaderBasedFilterConfig(headerFilterConfig)
- )
+ return newKafkaSupervisor()
+ .withIoConfig(ioConfig ->
ioConfig.withHeaderBasedFilterConfig(headerFilterConfig))
.withId(supervisorId)
.build(dataSource, topic);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
