hkwi commented on code in PR #16843:
URL: https://github.com/apache/iceberg/pull/16843#discussion_r3427547644
##########
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:
Thank you for pointing this out. I renamed the local flag to
`shouldStopCoordinator` in c852c178a so the predicate is clearly separated from
the `stopCoordinator()` action.
--
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]
