hudi-agent commented on code in PR #19432:
URL: https://github.com/apache/hudi/pull/19432#discussion_r3690252675
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -254,8 +254,12 @@ public void start() throws Exception {
}
restoreEvents(Long.MAX_VALUE);
} catch (Throwable throwable) {
- log.error("Failed to start operator coordinator.", throwable);
- context.failJob(throwable);
+ // Rethrow instead of context.failJob(): failJob triggers an in-graph
global failover that
+ // keeps this same coordinator instance alive without calling start()
again, leaving the
+ // half-initialized null fields (executor, writeClient, metaClient ...)
to be reused and NPE
+ // later. Rethrowing surfaces the failure as a JobMaster start failure
so the partially
+ // initialized instance is discarded rather than kept serving.
+ throw new HoodieException("Failed to start operator coordinator.",
throwable);
Review Comment:
🤖 I don't think this specific clientIds leak actually holds up.
`initClientIds` builds `ClientIds` via `.conf(conf)`, which uses the fixed
`WRITE_CLIENT_ID` (ClientIds L254) rather than calling `nextId()` — the new-id
allocation only happens in `OptionsInference` (guarded by
`!conf.contains(WRITE_CLIENT_ID)` and wrapped in try-with-resources), so a
fresh coordinator on retry reuses the same configured id instead of allocating
a new one. And since Flink's `startOperatorCoordinators` calls `closeQuietly`
on the failed coordinator (as the author noted above), `close()` runs
`clientIds.close()` → `executor.shutdownNow()`, so the heartbeat thread is
stopped rather than left orphaned. Am I missing a path where `nextId()` gets
re-invoked on failover?
--
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]
