wombatu-kun commented on code in PR #16843:
URL: https://github.com/apache/iceberg/pull/16843#discussion_r3426990096
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -159,31 +159,75 @@ public void stop() {
@Override
public void close(Collection<TopicPartition> closedPartitions) {
+ RuntimeException failure = null;
+
// Always try to stop the worker to avoid duplicates.
- stopWorker();
+ try {
+ stopWorker();
+ } catch (RuntimeException e) {
+ failure = appendFailure(failure, e);
+ LOG.warn("Committer {} failed to stop worker, continuing cleanup",
taskId, e);
+ }
// Defensive: close called without prior initialization (should not
happen).
if (!isInitialized.get()) {
LOG.warn("Close unexpectedly called on committer {} without partition
assignment", taskId);
+ if (failure != null) {
+ throw failure;
+ }
return;
}
- // Empty partitions → task was stopped explicitly. Stop coordinator if
running.
+ // Empty partitions: task was stopped explicitly. Stop coordinator if
running.
if (closedPartitions.isEmpty()) {
LOG.info("Committer {} stopped. Closing coordinator.", taskId);
- stopCoordinator();
+ try {
+ stopCoordinator();
+ } catch (RuntimeException e) {
+ failure = appendFailure(failure, e);
+ LOG.warn("Committer {} failed to stop coordinator", taskId, e);
+ }
+ if (failure != null) {
+ throw failure;
+ }
return;
}
// Normal close: if leader partition is lost, stop coordinator.
- if (hasLeaderPartition(closedPartitions)) {
+ boolean stopCoordinator = false;
Review Comment:
This local `boolean stopCoordinator` shares its name with the
`stopCoordinator()` method called a few lines below, so the branch reads as `if
(stopCoordinator) { ... stopCoordinator(); }`. Rename the flag to a predicate
such as `shouldStopCoordinator` (or `leaderPartitionLost`) to separate the
condition from the action it triggers.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -237,4 +281,13 @@ private void stopCoordinator() {
coordinatorThread = null;
}
}
+
+ private RuntimeException appendFailure(RuntimeException failure,
RuntimeException next) {
Review Comment:
This `appendFailure` (set-as-primary if none yet, else `addSuppressed`) is
the same merge logic hand-inlined in the catch blocks of `Channel.stop()`,
`Worker.stop()`, and `CoordinatorThread.terminate()` (two branches). All four
classes live in `org.apache.iceberg.connect.channel`; consider promoting one
package-private static helper and calling it from each so the
suppressed-exception handling stays in one place.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
