hudi-agent commented on code in PR #20271:
URL: https://github.com/apache/hudi/pull/20271#discussion_r4245894893
##########
hudi-kafka-connect/src/main/java/org/apache/hudi/connect/transaction/ConnectTransactionCoordinator.java:
##########
@@ -173,14 +171,60 @@ public void processControlEvent(ControlMessage message) {
@Override
public void run() {
- while (true) {
- try {
- CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- if (event != null) {
- processCoordinatorEvent(event);
+ try {
+ while (hasStarted.get()) {
+ try {
+ CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ if (event != null && hasStarted.get()) {
+ processCoordinatorEvent(event);
+ }
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ log.debug("Transaction coordinator event loop interrupted for topic
{} partition {}",
+ partition.topic(), partition.partition());
+ break;
}
- } catch (InterruptedException exception) {
- log.warn("Error received while polling the event loop in Partition
Coordinator", exception);
+ }
+ } finally {
+ closeTransactionServices();
Review Comment:
🤖 non-blocking: on the usual stop path the loop is parked in
`events.poll()`, so line 182 re-sets the interrupt flag right before this
`finally` runs `javaClient.close()`, and every blocking wait inside that close
fails immediately (the heartbeat scheduler's `awaitTermination`, and Jetty's
`QueuedThreadPool.doStop` `join` when the embedded timeline server stops, which
logs "Failed to stop the Javalin app"); could you clear the flag around the
close and restore it afterwards?
```java
} finally {
boolean interrupted = Thread.interrupted();
try {
closeTransactionServices();
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
```
<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,14 +171,60 @@ public void processControlEvent(ControlMessage message) {
@Override
public void run() {
- while (true) {
- try {
- CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- if (event != null) {
- processCoordinatorEvent(event);
+ try {
+ while (hasStarted.get()) {
+ try {
+ CoordinatorEvent event =
events.poll(COORDINATOR_EVENT_LOOP_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ if (event != null && hasStarted.get()) {
+ processCoordinatorEvent(event);
+ }
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ log.debug("Transaction coordinator event loop interrupted for topic
{} partition {}",
+ partition.topic(), partition.partition());
+ break;
}
- } catch (InterruptedException exception) {
- log.warn("Error received while polling the event loop in Partition
Coordinator", exception);
+ }
+ } finally {
+ closeTransactionServices();
+ }
+ }
+
+ @VisibleForTesting
+ boolean isEventLoopTerminated() {
+ return executorService.isTerminated();
+ }
+
+ private void shutdownEventLoop() {
+ scheduler.shutdownNow();
+ events.clear();
+ executorService.shutdownNow();
+ try {
+ if
(!executorService.awaitTermination(COORDINATOR_EVENT_LOOP_SHUTDOWN_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ log.warn("Transaction coordinator event loop did not terminate within
{} ms for topic {} partition {}",
+ COORDINATOR_EVENT_LOOP_SHUTDOWN_TIMEOUT_MS, partition.topic(),
partition.partition());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted while waiting for transaction coordinator event
loop to terminate for topic {} partition {}",
+ partition.topic(), partition.partition());
+ }
+
+ // The submitted event-loop task may have been cancelled before it
started, in which case
+ // run() cannot close the transaction services itself.
+ if (executorService.isTerminated()) {
+ closeTransactionServices();
+ }
+ }
+
+ private void closeTransactionServices() {
+ if (transactionServicesClosed.compareAndSet(false, true)) {
+ try {
+ transactionServices.close();
+ } catch (RuntimeException e) {
+ transactionServicesClosed.set(false);
+ log.warn("Failed to close transaction services for topic {} partition
{}",
Review Comment:
🤖 nit: could you add a short comment here explaining that resetting the flag
lets a later `stop()` or `shutdownEventLoop()` retry the close? Without it, the
reset looks like a mistake.
<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]
