FrankChen021 commented on code in PR #20072:
URL: https://github.com/apache/druid/pull/20072#discussion_r4219000941
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -2060,13 +2060,25 @@ public void runInternal()
logDebugReport();
}
catch (Exception e) {
- stateManager.recordThrowableEvent(e);
- if (e instanceof StreamException) {
- // When a StreamException is thrown, the error message is more useful
than the stack trace in telling what's wrong.
- log.makeAlert("Exception in supervisor run loop for supervisor[%s] for
dataSource[%s]: [%s]",
- supervisorId, dataSource, e.getMessage()).emit();
+ if (e instanceof ExecutionException || e instanceof
InterruptedException) {
Review Comment:
[P1] Classify task failures where they are recorded
**Finding:** Recoverable pause/checkpoint RPC failures do not reach this
catch as ExecutionException. FutureUtils.coalesce converts failed task futures
into Either.error; checkpointTaskGroup records pauseException directly, and
verifyAndMergeCheckpoints similarly records checkpoint errors. markRunFinished
therefore still counts these runs as failed, so repeated recoverable task
communication failures can still trigger UNHEALTHY_SUPERVISOR despite this
change.
**Suggestion:** Apply the recoverable-error policy at the task callback
recording sites, with regression coverage for failed pause and checkpoint
futures.
##########
processing/src/test/java/org/apache/druid/segment/join/PostJoinCursorTest.java:
##########
@@ -237,7 +237,7 @@ private void testAdvance(boolean withInterruption) throws
IOException, Interrupt
joinCursorThread.setUncaughtExceptionHandler(exceptionHandler);
joinCursorThread.start();
- countDownLatch.await(1, TimeUnit.SECONDS);
+ assertTrue(countDownLatch.await(1, TimeUnit.SECONDS));
Review Comment:
[P1] Interrupt the infinite cursor thread on assertion failure
**Finding:** If a slow or loaded test runner causes await to time out, the
new assertion throws before joinCursorThread.interrupt(). The worker is a
non-daemon thread whose CursorNoAdvance reports isDone() as false and whose
value matcher always rejects rows, so it can continue spinning indefinitely and
prevent the test JVM from exiting. The previous timeout path still reached the
interrupt.
**Suggestion:** Put interruption and bounded thread cleanup in a finally
block so the worker is stopped even when the latch assertion fails.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -2060,13 +2060,25 @@ public void runInternal()
logDebugReport();
}
catch (Exception e) {
- stateManager.recordThrowableEvent(e);
- if (e instanceof StreamException) {
- // When a StreamException is thrown, the error message is more useful
than the stack trace in telling what's wrong.
- log.makeAlert("Exception in supervisor run loop for supervisor[%s] for
dataSource[%s]: [%s]",
- supervisorId, dataSource, e.getMessage()).emit();
+ if (e instanceof ExecutionException || e instanceof
InterruptedException) {
+ // Task-level errors (e.g., task communication timeouts) are
recoverable and should not
+ // cause the supervisor to enter UNHEALTHY state. Log a warning
instead of recording
+ // as a throwable event that would count toward the unhealthiness
threshold.
+ log.noStackTrace().warn(
+ e,
Review Comment:
[P2] Avoid counting an aborted iteration as a successful run
**Finding:** When an ExecutionException or InterruptedException aborts the
first iteration before task discovery/checkpointing completes, this branch
leaves currentRunSuccessful true. The unconditional finally call to
markRunFinished then sets atLeastOneSuccessfulRun, increments the
successful-run counter, and transitions to RUNNING. Subsequent stream
connection failures bypass the first-run early-return guard, and an
already-unhealthy supervisor can also be declared recovered after enough
aborted iterations.
**Suggestion:** Represent recoverable aborted iterations separately from
successful and unhealthy failures, preserving first-run and recovery
bookkeeping until an iteration completes.
--
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]
