hudi-agent commented on code in PR #20271:
URL: https://github.com/apache/hudi/pull/20271#discussion_r4235270132
##########
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();
+ events.clear();
+ executorService.shutdownNow();
Review Comment:
🤖 non-blocking: now that `run()` exits by itself once `hasStarted` is false
(within the 1s poll timeout), would it make sense to call `shutdown()` and do a
short `awaitTermination` before `shutdownNow()`? That way an in-flight
`endCommit` (commit, compaction/clustering scheduling, meta sync) isn't
interrupted partway through on every rebalance.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 non-blocking: if `initializeGlobalCommittedKafkaOffsets()` throws inside
`start()`, `HoodieSinkTask.bootstrap` catches the `HoodieException` without
ever putting the coordinator into `transactionCoordinators`, so this `stop()`
is never called and the event-loop thread stays alive, still registered with
the control agent. Could you call `stop()` from `start()` on failure so this
leak is covered too?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -173,26 +177,47 @@ public void processControlEvent(ControlMessage message) {
@Override
public void run() {
- while (true) {
+ while (hasStarted.get()) {
try {
CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- if (event != null) {
+ if (event != null && hasStarted.get()) {
processCoordinatorEvent(event);
}
} catch (InterruptedException exception) {
- log.warn("Error received while polling the event loop in Partition
Coordinator", exception);
+ Thread.currentThread().interrupt();
+ log.debug("Transaction coordinator event loop interrupted for topic {}
partition {}",
+ partition.topic(), partition.partition());
+ break;
}
}
}
+ @VisibleForTesting
+ boolean isEventLoopTerminated() {
+ return executorService.isTerminated();
+ }
+
private void submitEvent(CoordinatorEvent event) {
this.submitEvent(event, 0, TimeUnit.SECONDS);
}
private void submitEvent(CoordinatorEvent event, long delay, TimeUnit unit) {
Review Comment:
🤖 nit: could you collapse the `hasStarted` check and the try/catch in
`submitEvent` into a single guard? The nested `hasStarted` checks in the lambda
and catch block are hard to follow.
<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]
