hudi-agent commented on code in PR #20271:
URL: https://github.com/apache/hudi/pull/20271#discussion_r4255720784
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -131,25 +150,20 @@ 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();
+ }
+ if (executorService.isTerminated()) {
Review Comment:
🤖 non-blocking: it looks like `HoodieSinkTask.close()`/`cleanup()` drops the
coordinator from `transactionCoordinators` as soon as `stop()` returns
normally, so in the timeout → delayed loop exit → close failure case nothing
calls `stop()` again and this retry only runs when deregistration also failed;
is that the coverage you intended, or should the leak case be retried somewhere
the sink task actually reaches (e.g. once more in `run()`'s `finally`)?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
