amaechler opened a new pull request, #20620: URL: https://github.com/apache/druid/pull/20620
### Description `EmbeddedKafkaSupervisorTest.testSupervisorIngestsOnlySelectedPartitions` (added in #20474) fails intermittently at its first SQL assertion with `Object 'datasource_xxx' not found`. In one downstream CI job it failed all 4 surefire attempts at line 130. Two other tests in the class also run SQL after a wait that does not check the Broker, so this PR fixes them too. #### Root cause `waitForSelectedRows` waited only for the indexer's `ingest/events/processed` metric. That metric means the task has read the row, not that the Broker can query it. The Broker adds a datasource to its SQL schema only after it discovers the realtime segment and refreshes the segment metadata, so the `SELECT COUNT(*)` right after the wait could reach the Broker first. The second half of the test has a related race. Claude found it by analyzing the code but I have not reproduced it. When the expanded spec replaces the supervisor, tasks that no longer match the partition allocation are stopped without publishing (`SeekableStreamSupervisor` calls `stopTask(id, false)`). If the first task had not published yet, its partition 1 offset was not committed, and the new task re-reads partition 1 from the earliest offset. The cumulative `events/processed >= 2` wait can then pass on the re-read alone, before partition 3 is ingested. Meanwhile the stopped task's segment can disappear from the Broker, which removes the datasource from the schema until the new segment is refreshed. #### Fix - **`EmbeddedClusterApis.waitForSegmentsToBeQueryable(dataSource, numSegments, timeoutMillis)`** is a new overload of `waitForAllSegmentsToBeQueryable`. It takes the segment count instead of reading it from the metadata store, so it also covers unpublished realtime segments, which `sys.segments` reports as available. `waitForAllSegmentsToBeQueryable` now delegates to it with unchanged behavior. - **`testSupervisorIngestsOnlySelectedPartitions`** waits through a new `waitForHandedOffRows` helper. It waits for `ingest/handoff/count` instead of `ingest/events/processed`, then calls `waitForAllSegmentsToBeQueryable`. With `maxRowsPerSegment = 1`, each row is its own segment, and the 500 ms task duration publishes the segments without suspending the supervisor. Handoff follows publish, which commits the partition 1 offset before the spec is replaced. The test also submits both specs through `submitSupervisor`, like the other tests in the class, so the `@AfterEach` teardown suspends the supervisor and cancels its tasks even when an assertion fails. - **`test_runKafkaSupervisorWithHeaderFiltering`** already waited for handoff, but then queried without checking the Broker. It now uses `waitForHandedOffRows` too. - **`test_runKafkaSupervisor`** checks the row count while ingestion is still running, so waiting for handoff would change what it tests. It now replaces its `ingest/events/processed` wait with `waitForSegmentsToBeQueryable(dataSource, 10, ...)`: the Broker must see all 10 one-row segments and have the table in its SQL schema. I used the polling helper instead of `waitForAllSegmentsToBeAvailable` because the second wait in the selected-partitions test must see the partition 3 segment. Events from the first half of that test can satisfy both conditions of the metric-based helper: the cumulative `segment/schemaCache/refresh/count` sum and the row-signature event. The races in the two sibling tests come from reading the code; I did not reproduce them. `test_runKafkaSupervisor` still has a small window: with 500 ms task durations, a segment can be handed off from the task to the Historical between the `sys.segments` poll and the `COUNT(*)` query. Waiting for handoff makes the selected-partitions test take about 5 to 7 seconds locally, instead of about 1 second. #### Related This is the fourth flaky-test fix in this class, and it builds on the earlier ones: - #19581 added the `ingest/events/processed` wait before the row count in `test_runKafkaSupervisor`. That wait shows that the Indexer read the rows but not that the Broker can query them, and `testSupervisorIngestsOnlySelectedPartitions` used the same wait. This PR replaces it in both tests. - #20479 (for #20478) added the `@AfterEach` teardown that the selected-partitions test now uses through `submitSupervisor`. - #20519 (for #20518) changed header filtering to wait for handoff without suspending the supervisor. The new helper keeps that wait. <hr> This PR has: - [x] been self-reviewed. - [x] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader. - [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
