cshuo commented on code in PR #20271:
URL: https://github.com/apache/hudi/pull/20271#discussion_r4244190176
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -132,24 +135,25 @@ public void start() {
@Override
public void stop() {
- kafkaControlClient.deregisterTransactionCoordinator(this);
- scheduler.shutdownNow();
- hasStarted.set(false);
- if (executorService != null) {
- boolean terminated = false;
+ if (!hasStarted.compareAndSet(true, false)) {
+ return;
+ }
+
+ try {
+ kafkaControlClient.deregisterTransactionCoordinator(this);
+ } finally {
+ scheduler.shutdownNow();
Review Comment:
Stopping mid-commit with lazy cleanup leaves the write client’s heartbeat
running, preventing the abandoned commit from expiring. Could we expose cleanup
on ConnectTransactionServices and close the owned write client after the event
loop terminates?
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -132,24 +135,25 @@ public void start() {
@Override
public void stop() {
- kafkaControlClient.deregisterTransactionCoordinator(this);
- scheduler.shutdownNow();
- hasStarted.set(false);
- if (executorService != null) {
- boolean terminated = false;
+ if (!hasStarted.compareAndSet(true, false)) {
Review Comment:
If deregistration throws before removing the coordinator, later stop() calls
return immediately because hasStarted is already false. The stale registration
then blocks a replacement coordinator. Could we track successful deregistration
separately so cleanup can be retried?
--
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]
