Refactor into Writer and Committer #234

fqtab · 2024-04-09T11:53:38Z

Changes summary:

Add Writer and Committer interface
Refactor existing logic to conform to the above interfaces

There should be no breaking changes in this PR.

fqtab

~~Still have to figure out why integration tests are failing in CI (passing on my local machine) but in the meantime,~~
Some comments to hopefully guide reviewers.

fqtab · 2024-04-09T11:59:38Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/IcebergSinkTask.java

+  private Writer writer;
+  private Committer committer;


Moved pretty much all of the logic out of this IcebergSinkTask class into Writer and Committer implementations. All we do here now is manage the lifecycle of a Writer and Committer.

This will make it easier in the future to introduce a pluggable Committer interface.

fqtab · 2024-04-09T12:03:54Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/api/Committer.java

+public interface Committer {
+  void commit(CommittableSupplier committableSupplier);
+}


This is the Committer interface I'm proposing.
I am not making this interface pluggable in this PR, nor do I intend to do so any time soon until the need arises.
This means we can still change it after this PR without having to do a breaking release.
So we don't need to get this interface 100% right here but we should at least try to align directionally on the design of the interface.

As far as the design is concerned, I've taken some inspiration from Flink here (they have a concept of a Committable as well).

Besides that, I've decided to go super simple with the API for now; just a single void commit(CommittableSupplier committableSupplier) method.

You might wonder, why not just void commit(Committable committable)?
The reason is because we want to let Committer implementations decide when to force the writers to flush and produce a Committable which by definition means closing all open files. We want to avoid closing files unnecessarily because that will result in many small files. Hence the Committer takes a CommittableSuppler and when the Committer determines it is a good time to commit, it can call CommittableSupplier.committables to force the Writer to close any open files and produce a Committable.

What is the purpose of having a pluggable committer interface?

Answered offline.

fqtab · 2024-04-09T12:04:52Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/IcebergSinkTask.java

-  public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
-    processControlEvents();
-  }


Removed the flush method entirely.
All the work now happens in put.

fqtab · 2024-04-09T12:05:34Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/IcebergSinkTask.java

+  public Map<TopicPartition, OffsetAndMetadata> preCommit(
+      Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
+    return ImmutableMap.of();


Not committing to the connect consumer group via this method anymore, all consumer-offset-commits are managed manually.

fqtab · 2024-04-09T12:08:46Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java

+      KafkaUtils.commitOffsets(
+          producer, offsetsToCommit, new ConsumerGroupMetadata(config.connectGroupId()));


This is where I commit to the connect- group now.
Right next to where I commit to the (current) source-of-truth group: control-group-id.
Note that committing to the connect- group is best-effort (same as before); the connect group is not currently the source of truth.
In a future PR, we will start committing to the connect- group exclusively (as part of worker zombie fencing) and make that the source of truth, however that is a breaking change which I want to avoid in this PR (we should bundle up breaking changes in another PR).

Re; future PR, if we stop committing to connect- won't we run into the potential for the connect- consumer group to be deleted by Kafka due to inactivity?

I think you've misunderstood. In the future, we will still be committing to the connect- consumer group. The only change is that we will be committing to the connect- group exclusively i.e. I intend to get rid of the connector-managed-consumer-group (but this is all theoretical right now, we can talk about it more when we get there).

fqtab · 2024-04-09T12:26:10Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/IcebergSinkConfig.java

+  public Integer taskId() {
    // this is for internal use and is not part of the config definition...
-    return originalProps.get(INTERNAL_TRANSACTIONAL_SUFFIX_PROP);
+    return Integer.valueOf(originalProps.get(INTERNAL_TASK_ID));


Another change I do not consider breaking as it's clearly documented as internal.

...a-connect/src/main/java/io/tabular/iceberg/connect/channel/CoordinatorThreadFactoryImpl.java

kafka-connect/src/main/java/io/tabular/iceberg/connect/IcebergSinkTask.java

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/Channel.java

fqtab · 2024-04-09T12:41:43Z

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/ConsumerFactoryImpl.java

+    // TODO: why putIfAbsent? why not just put?
    consumerProps.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");


I'm not sure I understand why this is a putIfAbsent rather than just put
I'd like to know why.
To me, the most correct thing here is to always start from latest.

I'm with you, I can't think of a good reason why you would set this to earliest. All this is doing is leaving it to be user configurable w/ a default to latest. Can live with it if we can't figure out the mystery.

tabmatfournier · 2024-04-10T18:14:10Z

...a-connect/src/main/java/io/tabular/iceberg/connect/channel/CoordinatorThreadFactoryImpl.java

+        thread.start();
+        LOG.info("Started commit coordinator");
+      } else {
+        thread = null;


Won't this branch result in a case where we have no coordinator thread running?

Are we making the assumption that we are no stable, therefor we are going to be rebalanced, but are assuming we are going to be eventually stable and we'll (eventually) not enter this branch?

Won't this branch result in a case where we have no coordinator thread running?

Yes, this is what is happening in the code today.

Are we making the assumption that we are no stable, therefor we are going to be rebalanced, but are assuming we are going to be eventually stable and we'll (eventually) not enter this branch?

This is my understanding.
@bryanck would appreciate it you could confirm if this is why you had this logic?

The reasoning for this was to ensure we account for all subscribed topics before ordering for the leader election., e.g. if you have multiple source topics.

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java

kafka-connect/src/main/java/io/tabular/iceberg/connect/data/WriterImpl.java

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java

tabmatfournier · 2024-04-10T20:08:26Z

Mostly questions. The main thrust of this is 👍 .

fqtab · 2024-04-11T19:49:20Z

@bryanck here's the refactor PR I mentioned earlier to enable a pluggable committer interface.
Would be good to get your thoughts/alignment on this before moving forward with the donation process :)

tabmatfournier

LGTM from my end.

fqtab marked this pull request as ready for review April 9, 2024 12:43

fqtab commented Apr 9, 2024

View reviewed changes

fqtab mentioned this pull request Apr 9, 2024

DO NOT MERGE: Refactor plan #230

Closed

tabmatfournier reviewed Apr 10, 2024

View reviewed changes

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java Show resolved Hide resolved

tabmatfournier reviewed Apr 10, 2024

View reviewed changes

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java Outdated Show resolved Hide resolved

tabmatfournier reviewed Apr 10, 2024

View reviewed changes

kafka-connect/src/main/java/io/tabular/iceberg/connect/data/WriterImpl.java Outdated Show resolved Hide resolved

tabmatfournier reviewed Apr 10, 2024

View reviewed changes

kafka-connect/src/main/java/io/tabular/iceberg/connect/channel/CommitterImpl.java Show resolved Hide resolved

fqtab force-pushed the refactor_writer_commiter_interface branch 3 times, most recently from e82981c to 375bc15 Compare April 17, 2024 12:50

fqaiser94 added 4 commits April 17, 2024 16:40

Refactor into Writer and Committer

7d6082e

Undo Worker changes

2743522

Undo KafkaClientFactory changes

a15fbf3

Undo Channel changes

1bd2488

fqtab force-pushed the refactor_writer_commiter_interface branch from 375bc15 to 1bd2488 Compare April 18, 2024 13:51

Add Task (public interface) and TaskImpl

d0de921

tabmatfournier approved these changes Apr 20, 2024

View reviewed changes

fqaiser94 added 7 commits April 23, 2024 14:53

minor

3df51c8

Avoid not-strictly-necessary Catalog initialization change for now

8a623bd

Avoid unnecessary Channel changes

da13bea

Avoid unnecessary Worker changes

b73848f

Clean up tests

dacb536

Avoid unnecessary CoordinatorThread initialization change

2acc71e

Minor: avoid exposing Catalog in TaskImpl constructor

669a79a

fqtab merged commit e3dee36 into main Apr 25, 2024
1 check passed

fqtab deleted the refactor_writer_commiter_interface branch April 25, 2024 13:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor into Writer and Committer #234

Refactor into Writer and Committer #234

fqtab commented Apr 9, 2024 •

edited

Loading

fqtab left a comment •

edited

Loading

fqtab Apr 9, 2024

fqtab Apr 9, 2024 •

edited

Loading

tabmatfournier Apr 9, 2024

tabmatfournier Apr 10, 2024

fqtab Apr 9, 2024

fqtab Apr 9, 2024 •

edited

Loading

fqtab Apr 9, 2024 •

edited

Loading

tabmatfournier Apr 10, 2024

fqtab Apr 12, 2024

fqtab Apr 9, 2024

fqtab Apr 9, 2024

tabmatfournier Apr 10, 2024

tabmatfournier Apr 10, 2024

fqtab Apr 15, 2024

bryanck Apr 15, 2024

tabmatfournier commented Apr 10, 2024

fqtab commented Apr 11, 2024

tabmatfournier left a comment

		KafkaUtils.commitOffsets(
		producer, offsetsToCommit, new ConsumerGroupMetadata(config.connectGroupId()));

		// TODO: why putIfAbsent? why not just put?
		consumerProps.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

Refactor into Writer and Committer #234

Refactor into Writer and Committer #234

Conversation

fqtab commented Apr 9, 2024 • edited Loading

fqtab left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fqtab Apr 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

fqtab Apr 9, 2024 • edited Loading

Choose a reason for hiding this comment

fqtab Apr 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

tabmatfournier commented Apr 10, 2024

fqtab commented Apr 11, 2024

tabmatfournier left a comment

Choose a reason for hiding this comment

fqtab commented Apr 9, 2024 •

edited

Loading

fqtab left a comment •

edited

Loading

fqtab Apr 9, 2024 •

edited

Loading

fqtab Apr 9, 2024 •

edited

Loading

fqtab Apr 9, 2024 •

edited

Loading