fix: prevent KafkaIndexSupervisor from entering UNHEALTHY on recoverable errors

[zhang-arvin]

Purpose

Fixes #18779: KafkaIndexSupervisor enters UNHEALTHY state on recoverable errors, blocking partition consumption.

Problem

When a Kafka ingestion task's duration exceeds the configured taskDuration, or when task-level operations (like checkTaskDuration(), updateTaskStatus(), checkCurrentTaskState(), checkPendingCompletionTasks()) fail with ExecutionException or InterruptedException, the supervisor's runInternal() method catches all exceptions and records them as throwable events via stateManager.recordThrowableEvent(e). After unhealthinessThreshold consecutive failed runs, the supervisor transitions to UNHEALTHY_SUPERVISOR state, which blocks all partition consumption.

These task-level exceptions are recoverable — they arise from individual task communication issues (e.g., task timeouts, task unresponsiveness) and do not indicate a supervisor-level failure. The supervisor should retry on the next iteration rather than entering UNHEALTHY state.

Fix

Modified SeekableStreamSupervisor.runInternal() to distinguish between recoverable task-level errors and non-recoverable supervisor-level errors:

• Recoverable: ExecutionException and InterruptedException — logged as warnings without calling recordThrowableEvent(), so the run is not marked as failed.
• Non-recoverable: StreamException and other exceptions — still recorded as throwable events, potentially triggering UNHEALTHY state.

This ensures that transient task communication failures don't cause the supervisor to stop consuming partitions.

Changes

• indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
  • Added import for java.util.concurrent.ExecutionException
  • Modified runInternal() catch block to handle ExecutionException/InterruptedException as recoverable errors

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• added Javadocs for most classes and all non-trivial methods.
• added comments explaining the "why" and the intent of the code wherever it is not obvious from reading the code.
• added unit tests or modified existing tests to cover new code paths.

[FrankChen021]

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

Reviewed 5 of 5 changed files.

The review found one unhealthy-state classification bug and one first-run success-state regression.

---

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

[FrankChen021]

[P1] Task failures still reach unhealthy-state accounting

Per-task failures are converted to Either.error and recorded inside checkpoint and pause callbacks, so they never reach this outer catch. Checked errors can also be wrapped as RuntimeException. Classify recoverable failures at their recording sites and add regression coverage.

[FrankChen021]

[P2] Recoverable exits are counted as successful runs

Skipping recordThrowableEvent leaves currentRunSuccessful true, while finally calls markRunFinished(). An aborted first run can therefore set atLeastOneSuccessfulRun and transition to RUNNING, bypassing first-run stream-failure handling on the next iteration.

[FrankChen021]

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

Reviewed 5 of 5 changed files.

The review found two high-confidence P1 issues and one P2 issue in recoverable supervisor failures and test cleanup.

---

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

[FrankChen021]

[P1] Task failures still reach unhealthy-state accounting

FutureUtils.coalesce converts task RPC failures to Either.error, and existing checkpoint/pause callbacks record those errors before this catch; wrapped failures can also bypass this type check. Repeated recoverable task failures can therefore still increment failure state and trip UNHEALTHY_SUPERVISOR.

[FrankChen021]

[P1] Timeout assertion can leak the infinite test thread

If the latch times out, assertTrue throws before joinCursorThread.interrupt(). The test cursor reports isDone() as false, so the non-daemon thread can loop indefinitely and hang the test JVM.

[FrankChen021]

[P2] Recoverable exits count as successful first runs

Skipping recordThrowableEvent leaves currentRunSuccessful true, while finally calls markRunFinished(). An aborted first run can therefore set atLeastOneSuccessfulRun and transition to RUNNING, bypassing first-run recovery behavior.