fix: stabilize task queue and ingestion CI tests

[FrankChen021]

Description

Fix three CI failures observed in Dependabot PRs:

• #20281: remove the task-count assertion across independent snapshots (986 vs 1000), and wait for cleanup instead of sleeping a fixed 100 ms.
• #20285: explicitly publish 500 records to each Kafka partition. The failed run reached offsets 100/96 while the test required 100/100.
• #20289: emit the Broker datasource-removal metric from the last-segment callback. The log recorded removal at 01:20:12 UTC, but the test timed out waiting for the refresh-only metric. Add coverage without a background refresh.

Validation

• TaskQueueScaleTest: 2 tests passed.
• BrokerSegmentMetadataCacheTest: 23 tests passed, including the new regression test.
• Checkstyle passed for all four affected modules.
• KafkaBoundedSupervisorTest: compiled; local execution blocked during Testcontainers startup by a Colima Docker socket mount error. CI validation pending.
• Follow-up smoke-test cleanup: all 6 IngestionSmokeTest methods passed locally (retries disabled), plus Checkstyle. Terminate the owned supervisor and drain/cancel datasource-scoped tasks between methods so leftover Kafka tasks cannot occupy the two worker slots; keep assertion timeouts unchanged.

Release note

Emit the Broker datasource-removal metric when the last segment is removed directly from the cache.

Review

• Self-reviewed, including concurrency behavior.
• Added or updated tests.

[Copilot]

🟡 Changes recommended

The new datasource-removal hook can lead to duplicate DATASOURCE_REMOVED emissions due to stale rebuild state unless the rebuild set is cleared when the datasource is removed (see stored review comment).

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR stabilizes several flaky CI tests around task queue state accounting, Kafka bounded ingestion offsets, and Broker datasource-removal metric emission by making the tests and cache behavior less timing- and distribution-sensitive.

Changes:

• Add a datasource-removal hook to the segment metadata cache and emit DATASOURCE_REMOVED from the Broker cache when the last segment is removed (without requiring a refresh cycle).
• Make TaskQueueScaleTest avoid summing non-atomic counters and wait for cleanup completion instead of using a fixed sleep.
• Ensure bounded Kafka supervisor tests publish records to both partitions so end offsets are deterministic, and add a regression test for datasource-removal metric emission.

File summaries

File	Description
sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java	Adds regression coverage ensuring last-segment removal emits the datasource-removed metric without a background refresh.
sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java	Emits DATASOURCE_REMOVED via the new datasource-removal hook when the cache drops the last segment of a datasource.
server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java	Introduces a removeDataSourceAction hook invoked when the last segment of a datasource is removed.
indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java	Removes a flaky assertion based on independent snapshots; waits for cleanup to complete before asserting queue/storage emptiness.
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java	Publishes records explicitly to each partition to stabilize per-partition end-offset assertions.

Review details

• Files reviewed: 5/5 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[FrankChen021]

Review deferred: GitHub currently reports this PR as conflicting with master.

Current head: ee5d92b55d98678655c1bb41819702c50bb60f8a (codex/stabilize-ci-tests). Please resolve the conflicts with master and push the updated head so the changes can be reviewed later.

Reviewed 0 of 0 changed files; review was deferred before preparation.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

Reviewed 6 of 6 changed files.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

🟡 Changes recommended

Reviewed all 6 changed files at the current head. Coverage: embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java; embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java; indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java; server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java; sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java; sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java.

The cleanup, bounded Kafka test stabilization, task-queue synchronization, and stale rebuild-state changes otherwise look consistent. One remaining P2 metric/concurrency issue is called out inline: the empty-row-signature path can still duplicate DATASOURCE_REMOVED when a last-segment callback wins an in-flight refresh.

Static review only. Validation actually run: git diff --check against the PR merge base (passed). No builds, tests, dependency installs, formatters, or broad validations were run.

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

Findings that could not be attached inline:

• sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java:268 - [P2] Guard the empty-signature removal metric

  Finding: The new at-most-once gate covers only the rowSignature == null path. buildDataSourceRowSignature returns an empty signature when the map still contains an unrefreshed segment whose row signature is null; if that last segment is removed while this refresh is in flight, removeSegment removes the table and the new callback emits DATASOURCE_REMOVED, then this branch calls tables.remove (which returns null) but emits the metric unconditionally. Normal segment churn can therefore still produce two removal events for one datasource removal.

  Suggestion: Apply the same successful-removal guard or shared helper to the empty-signature branch, and add a regression test covering an in-flight refresh with an uninitialized segment.

---

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

[FrankChen021]

Re the remaining P2 (BrokerSegmentMetadataCache.java:268, empty-signature removal metric): good catch, addressed in a59f5f5.

buildDataSourceRowSignature returns an empty but non-null signature while a datasource only holds unrefreshed segments, and that branch still emitted dataSource/removed unconditionally. It now uses the same tables.remove(dataSource) != null guard as the null-signature branch and the last-segment callback, so whichever path actually removes the table is the only one that emits. As a side effect, a new datasource whose schema has not been initialised no longer reports a removal on every refresh cycle when no table ever existed.

Tests added:

• testLastSegmentRemovedDuringRefreshWithUninitializedSegmentEmitsRemovalOnce drives the exact interleaving: the table is built from a refreshed segment, an unrefreshed segment is added and the refreshed one removed, a refresh is started on another thread and held (via an overridden buildDataSourceRowSignature) right after computing the empty signature, the last segment is removed on the test thread, then the refresh is released. Asserts exactly one dataSource/removed and no foo table.
• testRefreshOfUninitializedDatasourceDoesNotEmitRemovalMetric covers the never-initialised case.

Both fail on the previous head (expected: <1> but was: <2> and expected: <0> but was: <1>) and pass now; BrokerSegmentMetadataCacheTest is 26/26 locally and checkstyle passes.

[FrankChen021]

🟢 Approval recommended

No actionable issues found in this review. The incremental fix correctly closes the previously reported duplicate dataSource/removed path: both removal branches and the last-segment callback gate emission on successful table removal. I reviewed the complete current diff and surrounding task cleanup, bounded Kafka offset setup, task-queue synchronization, and metadata-cache changes; no remaining merge-blocking correctness issue was found.

Reviewed 6 of 6 changed files.

Static review only. Validation actually run: git diff --check 2fed7ca11d9852412ce76798a3fa150ddae9afdb a59f5f5d4d0856b568dbfe3d31ebd3135b47107d (passed). No builds, tests, dependency installs, formatters, or broad validations were run.

---

This is an automated review by Codex GPT-5.6-Luna(max)