bryanck commented on code in PR #12372:
URL: https://github.com/apache/iceberg/pull/12372#discussion_r1967903020
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java:
##########
@@ -46,26 +45,38 @@ public String version() {
@Override
public void start(Map<String, String> props) {
this.config = new IcebergSinkConfig(props);
+ /*
+ catalog and committer are global resource and does not depend on the
topic partition,
+ hence we should open this with the start call only and should only close
these if the task is closed by connect
+ framework.
+ */
+ catalog = CatalogUtils.loadCatalog(config);
+ committer = CommitterFactory.createCommitter(catalog, config, context);
}
@Override
public void open(Collection<TopicPartition> partitions) {
- Preconditions.checkArgument(catalog == null, "Catalog already open");
- Preconditions.checkArgument(committer == null, "Committer already open");
-
- catalog = CatalogUtils.loadCatalog(config);
- committer = CommitterFactory.createCommitter(config);
- committer.start(catalog, config, context);
+ // We should be starting co-ordinator only the list of partitions has the
zeroth partition.
+ if(committer.isCoordinator(partitions)) {
+ committer.startCoordinator();
+ }
+ committer.syncLastCommittedOffsets();
}
@Override
public void close(Collection<TopicPartition> partitions) {
- close();
+ // We need to close worker here in every case to ensure exactly once
otherwise this will lead to duplicate records.
+ committer.stopWorker();
+ // Coordinator should only be closed if this received closed partitions
has the partition which elected this task as coordinator.
+ if(committer.isCoordinator(partitions)) {
+ committer.stopCoordinator();
+ }
}
private void close() {
if (committer != null) {
- committer.stop();
+ committer.stopWorker();
Review Comment:
Same as above, we should keep a generic `stop` and let the committer
implementation decide what to do.
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
