bryanck commented on code in PR #12372:
URL: https://github.com/apache/iceberg/pull/12372#discussion_r1967898907
##########
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)) {
Review Comment:
We should move this logic into the committer implementation.
--
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
