Kafka Connect: Commit coordination #10351

bryanck · 2024-05-18T14:25:55Z

This PR is the next stage in submitting the Kafka Connect Iceberg sink connector, and is a follow up to #8701, #9466, and #9641. It includes the commit coordinator and related tests.

Still not included for the sink are the integration tests, distribution build, or docs, which will be added in follow up PRs. For reference, the current sink implementation can be found at https://github.com/tabular-io/iceberg-kafka-connect, and you can read some existing docs at https://github.com/tabular-io/iceberg-kafka-connect/tree/main/docs.

...nect/kafka-connect-events/src/test/java/org/apache/iceberg/connect/events/EventTestUtil.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/CatalogUtils.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/CommitterFactory.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java

ajantha-bhat · 2024-05-21T10:43:09Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

+  public void put(Collection<SinkRecord> sinkRecords) {
+    if (committer != null) {
+      committer.save(sinkRecords);
+    }


should throw an exception when committer is null else the producer will assume it has been put?

This should never happen, so I changed this to a precondition check. I'll revisit this if needed when I add the integration tests.

I can tell you that it does happen and there's nothing we can do about it other than throw an error and have the task retry. Precondition check makes sense 👍

ajantha-bhat · 2024-05-21T12:12:24Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+            .filter(distinctByKey(dataFile -> dataFile.path().toString()))
+            .collect(Collectors.toList());
+
+    List<DeleteFile> deleteFiles =


Since we are only supporting append, do we need this code?

Similar comments for below RowDelta block.

I wanted to keep the coordinator capable of handling delete files, so when we do add in delta support, it should be fairly straightforward, and won't require changes to the control message data model.

ajantha-bhat · 2024-05-21T12:36:35Z

...-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/CoordinatorTest.java

+  }
+
+  @Test
+  public void testCommitDelta() {


maybe delete file tests can be added when we have delete writers feature added.

See my comment above, I'd prefer to add this into the coordinator now, rather than later.

fqaiser94 · 2024-05-27T14:19:05Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

+  @Override
+  public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
+    committer.save(null);
+  }


Why are we overriding the flush method when we don't have any flush-specific code path in Committer?

The put method will be called on a regular basis (potentially with an empty collection of sink records) so this feels redundant.

Also, I'm fairly certain that this flush method will never actually be called since we are overriding the preCommit method. The default preCommit implementation is the only place where flush is called by the Kafka Connect runtime. Unless you call flush yourself in the preCommit method you've defined (or anywhere else), flush will never actually be called.

Overall, I would recommend you just omit this flush method definition from this class.

This is here in case future committer implementations want to perform an action on flush. The current committer doesn't do anything when the record collection is null.

This is here in case future committer implementations want to perform an action on flush.

What I'm trying to say is that flush will never be called since you're overriding the default preCommit implementation: https://github.com/apache/kafka/blob/c97d4ce026fdb75f80760bcce00b239db951a481/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkTask.java#L139

So no committer implementation will ever be able to "perform an action on flush" unless you call flush yourself somewhere.

I felt it is safer to implement this, given it is a public method, rather than rely on default behavior.

fqaiser94 · 2024-05-27T14:21:28Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

+  @Override
+  public void close(Collection<TopicPartition> partitions) {
+    close();
+  }
+
+  private void close() {
+    if (committer != null) {
+      committer.stop();
+      committer = null;
+    }
+
+    if (catalog != null) {
+      if (catalog instanceof AutoCloseable) {
+        try {
+          ((AutoCloseable) catalog).close();
+        } catch (Exception e) {
+          LOG.warn("An error occurred closing catalog instance, ignoring...", e);
+        }
+      }
+      catalog = null;
+    }
+  }


nit: can you move these close methods so they're after preCommit but before stop? Just so these methods are arranged in life-cycle order.

I'm not sure I follow your suggestion. When the KC close() or stop() lifecycle methods are called, we close the committer and the catalog.

I think he means just rearrange the code, keep close method just before stop.

I have this order to prevent a running committer from having reference to a closed catalog.

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

fqaiser94 · 2024-05-27T15:30:03Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

+
+  @Override
+  public void open(Collection<TopicPartition> partitions) {
+    catalog = CatalogUtils.loadCatalog(config);


open method is called with only the newly assigned partitions. Is there a strong reason to pass just the newly assigned partitions to the Committer.start method when the Committer can just retrieve all partitions assigned to this task via context.assignment anyway?

I'm also worried we might have a bug here. The Committer implementation uses this partitions argument to check if partition 0 of the first topic is assigned to this task and if so, it spawns a Coordinator process. I'm worried that if there was a rebalance where the partition 0 of the first topic doesn't move between tasks, then it would not be included in the partitions argument for any Task and thus we could potentially end up with a Connector that doesn't have any Coordinator process running on any Task. Thoughts?

That's not my understanding, open() will be called with the new assignment, i.e. all assigned topic partitions. See the javadoc: "The list of partitions that are now assigned to the task (may include partitions previously assigned to the task)"

I did read the javadoc and interpreted it differently but I can totally see how you reached your interpretation as well.

The reason I'm inclined towards my interpretation is because I know SinkTask.open is ultimately called by the ConsumerRebalanceListener.onPartitionsAssigned method (here in the Kafka source code) and if you look at the javadocs for that method, it says The list of partitions that are now assigned to the consumer (previously owned partitions will NOT be included, i.e. this list will only include newly added partitions).

I could still be wrong so feel free to test things out or ask in the kafka community.

Thanks, it does seem like you're right. I changed this to get the consumer assignment instead.

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java

fqaiser94 · 2024-05-30T19:32:14Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java

+
+  private void routeRecordDynamically(SinkRecord record) {
+    String routeField = config.tablesRouteField();
+    Preconditions.checkNotNull(routeField, "Route field cannot be null with dynamic routing");


This should really be checked at config parsing time or at SinkWriter construction time?
Instead of on every record.
To be clear; I'm not worried about this from a performance perspective (I'm confident the JVM will optimize this away) but it just seems awkward.

We already check this in the config, so I'll just remove this.

fqaiser94 · 2024-05-30T20:11:17Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java

+        new TopicPartition(record.topic(), record.kafkaPartition()),
+        new Offset(record.kafkaOffset() + 1, timestamp));
+
+    if (config.dynamicTablesEnabled()) {


Same here, we really don't need to check this on every record?

Feels like we're missing an abstraction here, something like the concept of a Router which has a StaticRouter implementation and a DynamicRouter implementation and only one of those is constructed for the lifetime of a SinkWriter based on the config.dynamicTablesEnabled() setting.

Agree. But IMO we should avoid too much internal refactoring which makes extracting code from the Tabular implementation harder.

We can always refactor once the complete code is in.

I think this is a good suggestion, and I looked into adding an abstraction here. Ultimately I felt it added complexity, so I agree we should leave it as-is for now.

fqaiser94 · 2024-05-30T20:13:23Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java

+    Preconditions.checkNotNull(routeField, "Route field cannot be null with dynamic routing");
+
+    String routeValue = extractRouteValue(record.value(), routeField);
+    if (routeValue != null) {


IMO we should throw an error instead of dropping the record.
Users can filter out messages easily using an SMT to get the same behaviour, if necessary.
In the future, I imagine we can allow users to supply custom exception-handler implementations which could also allow users to drop records on error.

I feel skipping is better than getting into a state where the sink can no longer progress. When we add DLQ support then we could route to that.

+1, we can add a TODO about DLQ here or an issue for tracking it.

fqaiser94 · 2024-05-30T20:14:00Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java

+    String routeValue = extractRouteValue(record.value(), routeField);
+    if (routeValue != null) {
+      String tableName = routeValue.toLowerCase();
+      writerForTable(tableName, record, true).write(record);
+    }


Can we support writing to multiple tables if this is a comma separate list of table names? Like we do in "static" mode.

Unless there are other concerns, I think this is an important building block for more advanced functionality e.g. we can remove route-value-regex-table-routing-mode as it could be implemented entirely as an SMT + dynamic mode.

I think that could be a useful feature, but I don't think it should hold up this PR. If you wanted to open an issue that would be great.

fqaiser94 · 2024-05-30T20:14:17Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SinkWriter.java

+      String routeValue = extractRouteValue(record.value(), routeField);
+      if (routeValue != null) {
+        config
+            .tables()
+            .forEach(
+                tableName -> {
+                  Pattern regex = config.tableConfig(tableName).routeRegex();
+                  if (regex != null && regex.matcher(routeValue).matches()) {
+                    writerForTable(tableName, record, false).write(record);
+                  }
+                });
+      }


nit: is this really "static" routing? I guess it's a static list of tables, but the table/s each message is written to is determined dynamically based on the route value ...

More importantly, is this an important enough use-case to support within the connector? I would strongly prefer if we didn't support this within the connector itself (users can easily implement this by writing an SMT + dynamic mode).

Static in the sense that the list of tables is fixed and doesn't change, rather than deriving it from the record. This feature is in use by some.

ajantha-bhat

LGTM.

The required basic functionality of commit coordinator looks good to me.

bryanck · 2024-06-16T16:22:30Z

@ajantha-bhat @fqaiser94 Thanks for the reviews! I think I have addressed comments in case you want to take another look. @Fokko @nastra If either of you have time to take a look, that would be very appreciated.

jbonofre

LGTM, thanks !

ajantha-bhat · 2024-06-28T10:44:26Z

@rdblue, @danielcweeks, @Fokko, @nastra: Please take a look. It is needed for 1.6.0 milestone.

fqaiser94

Sorry for the delay, this has been on my radar for a couple of weeks but i don't have a lot of time for reviews currently.

fqaiser94 · 2024-07-03T02:08:44Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+        AppendFiles appendOp = table.newAppend();
+        if (branch != null) {
+          appendOp.toBranch(branch);
+        }
+        appendOp.set(snapshotOffsetsProp, offsetsJson);
+        appendOp.set(COMMIT_ID_SNAPSHOT_PROP, commitState.currentCommitId().toString());
+        if (vtts != null) {
+          appendOp.set(VTTS_SNAPSHOT_PROP, vtts.toString());
+        }
+        dataFiles.forEach(appendOp::appendFile);
+        appendOp.commit();


I think you forgot to port over the the code for handling multiple partition-specs
tabular-io/iceberg-kafka-connect@cdd54f3

Otherwise, connectors can become hard-stuck when users evolve the partition whilst the connector is running, which I don't think is an entirely unreasonable expectation.

Alternatively, we could try getting this PR into iceberg-core: #9860
That would remove the need to do anything special here.

I feel this should be addressed in Iceberg core

fqaiser94 · 2024-07-03T02:20:34Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+            .filter(
+                envelope -> {
+                  Long minOffset = committedOffsets.get(envelope.partition());
+                  return minOffset == null || envelope.offset() >= minOffset;
+                })


Just a note for posterity's sake (not requesting any changes):
While this is a good check, this does not guarantee truly exactly-once.
There is an unlikely edge-case where if we have two coordinator zombies running, even this check will be insufficient to prevent duplicates.
Unless we add support for conditional commits to iceberg, there's not much that can be done about it unfortunately (well, nothing simple anyway).

fqaiser94 · 2024-07-03T02:23:28Z

...nnect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaClientFactory.java

+    consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+    consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
+    consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());


Suggested change

consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());

I would avoid randomly generating client ID config since they're used for rate limiting in Kafka.
I would just remove this line entirely (if users wish to set a ClientID they can do so via the connector kafka configs).

I changed this to putIfAbsent so it can be overridden.

fqaiser94 · 2024-07-03T02:23:37Z

...nnect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaClientFactory.java

+  Producer<String, byte[]> createProducer(String transactionalId) {
+    Map<String, Object> producerProps = Maps.newHashMap(kafkaProps);
+    producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
+    producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());


Suggested change

producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());

Same here

Changed to putIfAbsent here also.

fqaiser94 · 2024-07-03T02:25:37Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java

+    // pass transient consumer group ID to which we never commit offsets
+    super(
+        "worker",
+        IcebergSinkConfig.DEFAULT_CONTROL_GROUP_PREFIX + UUID.randomUUID(),


Can we put the connect consumer group id in here somewhere?
Otherwise, it's impossible for users to map a random UUIDs back to a specific connector if necessary.

fqaiser94 · 2024-07-03T02:30:12Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java

+    events.add(readyEvent);
+
+    send(events, results.sourceOffsets());
+    context.requestCommit();


Suggested change

context.requestCommit();

Why are we doing this at all?
The previous line already commits the offsets for the connector consumer group.
Plus, Kafka Connect runtime won't commit anything anyway since IcebergSinkTask.preCommit returns an empty map.

I removed this, it is no longer necessary.

fqaiser94 · 2024-07-03T02:34:19Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

+    this.snapshotOffsetsProp =
+        String.format(
+            "kafka.connect.offsets.%s.%s", config.controlTopic(), config.connectGroupId());
+    this.exec = ThreadPools.newWorkerPool("iceberg-committer", config.commitThreads());


we never shut this thread pool down. Is that an issue?

Thanks, I added the explicit shutdown call to ensure tasks are terminated before starting a new coordinator.

fqaiser94 · 2024-07-03T02:34:59Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java

+    }
+
+    if (System.currentTimeMillis() - startTime > config.commitTimeoutMs()) {
+      LOG.info("Commit timeout reached");


Please interpolate the commit id in here as well.

fqaiser94 · 2024-07-03T02:39:53Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java

+        recordList.forEach(producer::send);
+        if (!sourceOffsets.isEmpty()) {
+          producer.sendOffsetsToTransaction(
+              offsetsToCommit, KafkaUtils.consumerGroupMetadata(context, connectGroupId));


Suggested change

offsetsToCommit, KafkaUtils.consumerGroupMetadata(context, connectGroupId));

offsetsToCommit, this.consumerGroupMetadata);

We only need to extract consumer-group-metadata once, we should just do that in the constructor and make it a class field.

The consumer-group-metadata contains the consumer-generation which is used by Kafka brokers for zombie fencing. I'm probably over-thinking this massively but I'm worried that the internal KafkaConsumer might rebalance, get the new generation, and then a zombie task could end up successfully committing a transaction resulting in duplicates.

Regardless, I would move this to a class field as that is definitely safe.

We need to update the metadata to get the latest generation ID.

fqaiser94 · 2024-07-03T02:41:00Z

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/KafkaUtils.java

+              DynFields.builder().hiddenImpl(CONTEXT_CLASS_NAME, "consumer").build(context).get())
+          .groupMetadata();
+    }
+    return new ConsumerGroupMetadata(connectGroupId);


Are we sure we want to fall back to the non-zombie-fencing consumer-group-metadata?
IMO we should just hard fail, this shouldn't happen and if it does, we're risking duplicates.
If not, we should at least throw a warning log if we hit this case.

I removed the fallback, so it hard fails now.

danielcweeks

Some minor nits and questions, but overall LGTM.

I think there are still a couple open questions from others (particularly the one on IcebergSinkTask.open(), which we probably want to address.

I did see that we renamed a few classes and reduced the visibility in others. I'm ok with that for now since this isn't really deployable, but we should probably avoid that going forward.

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/CommitterFactory.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java

bryanck · 2024-07-11T05:03:32Z

I think I've addressed the latest feedback @fqaiser94 @danielcweeks LMK if there is anything else.

danielcweeks · 2024-07-11T15:22:02Z

Thanks @bryanck and @fqaiser94. It's really great to get this one in.

* Kafka Connect: Commit coordination * PR feedback * Get partition assignment from consumer * test fix

github-actions bot added the KAFKACONNECT label May 18, 2024

ajantha-bhat reviewed May 21, 2024

View reviewed changes

ajantha-bhat added this to the Iceberg 1.6.0 milestone May 21, 2024

fqaiser94 reviewed May 27, 2024

View reviewed changes

fqaiser94 reviewed May 30, 2024

View reviewed changes

bryanck force-pushed the kc-coord branch 3 times, most recently from 9242cd1 to 731fdb8 Compare June 11, 2024 14:55

ajantha-bhat approved these changes Jun 12, 2024

View reviewed changes

bryanck force-pushed the kc-coord branch from 731fdb8 to 579bc97 Compare June 16, 2024 15:54

jbonofre approved these changes Jun 25, 2024

View reviewed changes

bryanck force-pushed the kc-coord branch from 579bc97 to 50ff989 Compare June 29, 2024 15:24

danielcweeks requested review from fqaiser94 and danielcweeks and removed request for fqaiser94 June 30, 2024 04:41

fqaiser94 reviewed Jul 3, 2024

View reviewed changes

bryanck force-pushed the kc-coord branch 3 times, most recently from a025e04 to f3ba4c0 Compare July 8, 2024 15:34

Fokko requested a review from fqaiser94 July 10, 2024 10:41

Kafka Connect: Commit coordination

a3b98d7

bryanck force-pushed the kc-coord branch from f3ba4c0 to a3b98d7 Compare July 10, 2024 16:58

danielcweeks reviewed Jul 11, 2024

View reviewed changes

PR feedback

de3739f

bryanck force-pushed the kc-coord branch from 0d40d99 to de3739f Compare July 11, 2024 03:08

Get partition assignment from consumer

2ec253d

bryanck force-pushed the kc-coord branch from 6b2beec to 2ec253d Compare July 11, 2024 04:26

test fix

e276a3d

danielcweeks approved these changes Jul 11, 2024

View reviewed changes

danielcweeks merged commit d8e7642 into apache:main Jul 11, 2024
49 checks passed

amogh-jahagirdar mentioned this pull request Jul 16, 2024

Core: Support appending files with different specs #9860

Merged

bryanck mentioned this pull request Jul 21, 2024

Kafka Connect: Runtime distribution with integration tests #10739

Merged

jasonf20 pushed a commit to jasonf20/iceberg that referenced this pull request Aug 4, 2024

Kafka Connect: Commit coordination (apache#10351)

a692cb8

* Kafka Connect: Commit coordination * PR feedback * Get partition assignment from consumer * test fix

	offsetsToCommit, KafkaUtils.consumerGroupMetadata(context, connectGroupId));
	offsetsToCommit, this.consumerGroupMetadata);

Kafka Connect: Commit coordination #10351

Kafka Connect: Commit coordination #10351

Conversation

bryanck commented May 18, 2024

Choose a reason for hiding this comment

bryanck Jun 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fqaiser94 May 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajantha-bhat left a comment

Choose a reason for hiding this comment

bryanck commented Jun 16, 2024

jbonofre left a comment

Choose a reason for hiding this comment

ajantha-bhat commented Jun 28, 2024

fqaiser94 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bryanck Jul 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fqaiser94 Jul 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

danielcweeks left a comment • edited Loading

Choose a reason for hiding this comment

bryanck commented Jul 11, 2024

danielcweeks commented Jul 11, 2024

bryanck Jun 9, 2024 •

edited

Loading

fqaiser94 May 30, 2024 •

edited

Loading

bryanck Jul 6, 2024 •

edited

Loading

fqaiser94 Jul 3, 2024 •

edited

Loading

danielcweeks left a comment •

edited

Loading