danny0405 commented on code in PR #20271:
URL: https://github.com/apache/hudi/pull/20271#discussion_r4254045361
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -131,25 +137,17 @@ public void start() {
}
@Override
- public void stop() {
- kafkaControlClient.deregisterTransactionCoordinator(this);
- scheduler.shutdownNow();
+ public synchronized void stop() {
hasStarted.set(false);
- if (executorService != null) {
- boolean terminated = false;
- try {
- log.info("Shutting down executor service.");
- executorService.shutdown();
- log.info("Awaiting termination.");
- terminated = executorService.awaitTermination(100,
TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- // ignored
+ try {
+ if (!deregistered) {
+ kafkaControlClient.deregisterTransactionCoordinator(this);
+ deregistered = true;
}
-
- if (!terminated) {
- log.warn(
- "Unclean Kafka Control Manager executor service shutdown ");
- executorService.shutdownNow();
+ } finally {
+ if (!shutdownInitiated) {
+ shutdownInitiated = true;
+ shutdownEventLoop();
Review Comment:
[P2] Allow subsequent stop() calls to retry failed service cleanup
If an in-flight operation outlasts the five-second wait,
`shutdownEventLoop()` returns before the executor terminates. When the loop
eventually exits, a transient exception from `transactionServices.close()`
resets `transactionServicesClosed` to false, but every later `stop()` skips
this block because `shutdownInitiated` is already true. There is then no caller
left to retry cleanup, so any resources left open by the failed close remain
leaked. I reproduced this with an event handler held past the timeout and a
close mock that fails only its first invocation: after the loop terminates and
`stop()` is called again, the close count is still 1 and the closed flag is
false.
Could we move the `isTerminated()` / `closeTransactionServices()` fallback
out of `shutdownEventLoop()` and place it after this one-time shutdown guard in
`stop()`? That separates one-time executor shutdown from retryable resource
cleanup, preserves the ordering with event processing, and makes resetting the
closed flag effective. Please add a regression test for timeout → delayed loop
exit → close failure → repeated stop().
--
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]
