KAFKA-10000: Exactly-once source tasks (KIP-618) #11780

C0urante · 2022-02-17T04:28:07Z

Implements a source task wrapper (ExactlyOnceWorkerSourceTask) that follows the behavior described in KIP-618 for writing source records and their offsets in transactions with user-configurable (and sometimes connector-defined) boundaries.

Relies on changes from:

KAFKA-10000: Add new metrics for source task transactions (KIP-618) #11772
KAFKA-10000: Add all public-facing config properties related to exactly-once source support (KIP-618) #11775

C0urante · 2022-02-17T04:35:11Z

connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java

This is a placeholder implementation. This class is fully implemented in the downstream PR #11781

C0urante · 2022-02-17T22:00:12Z

Jenkins build failures appear to be due to the issue addressed by #10702. I've pushed a change that applies the same fix from that PR to the newly-added failing test classes for this PR.

C0urante · 2022-02-18T06:35:45Z

Converting to draft until upstream PRs are reviewed.

C0urante · 2022-05-10T02:47:58Z

Given that all merge conflicts have been resolved and #11775 has already been approved, marking this as ready for review.

mimaison

Thanks @C0urante for the PR. I've made a first pass over the main (I've not looked at tests yet) and it looks pretty good. I left a few minor comments.

mimaison · 2022-05-14T11:54:17Z

connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java

Can we undo this?

Apologies, will revert and remove similar changes from other PRs.

mimaison · 2022-05-17T10:54:17Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java

Are you planning to address this TODO in one of the other PRs? Same below in ExactlyOnceSourceTaskBuilder

Yes; per-connector offsets topics are implemented in #11781.

Can we add a comment ? so it's clear to anyone that stumbles on this

mimaison · 2022-05-17T11:02:43Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java

Nit: can we move this block above the admin block above? so it follows the same flow than SourceTaskBuilder.doBuild()

Good idea, done.

mimaison · 2022-05-17T11:04:25Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java

Nit: All other messages are capitalized

Good point; given that Utils::closeQuietly uses the second parameter in the middle of a log message I'm inclined to lowercase each of these.

mimaison · 2022-05-17T11:10:33Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java

Nit: Let's keep the new line

mimaison · 2022-05-17T11:18:32Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java

Should we use DistributedConfig.transactionalProducerId() here?

No, the two serve different purposes.

DistributedConfig::transactionalProducerId computes the transactional ID that the leader should use when writing to the config topic

Worker::transactionalId computes the transactional ID that exactly-once source tasks should use when writing to Kafka

Thanks, that makes sense. I thought we wanted to use DistributedConfig::transactionalProducerId as a prefix in Worker::transactionalId but that's not necessary.

Is this worth a comment?

👍 added a comment and updated the method name.

mimaison · 2022-05-17T11:24:27Z

@rhauch @kkonstantine This is a pretty important feature, can you take a look? The first few PRs were relatively simple but in the remaining PRs things are getting a bit more involved so it would be good to get another pair of eyes on it.

C0urante · 2022-05-17T15:48:54Z

Thanks @mimaison. I've addressed your comments and this is ready for another round

mimaison · 2022-05-19T10:16:49Z

Thanks for the updates @C0urante ! I've on PTO next week, I'll take a look at the tests when I'm back.

C0urante · 2022-06-01T15:23:29Z

@mimaison I'm hoping you can take a look sometime this week. There's still a lot left to review for this KIP and it's worrying that the 3.3.0 feature freeze deadline is creeping closer with little progress being made.

tombentley

Thanks @C0urante, I've made an initial pass.

tombentley · 2022-06-02T10:24:02Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java

Why should they be completed by this point? I.e. what in the Java memory model guarantees it?

That's a fair point. Right now this is a result of the threading model of the producer, which blocks on the completion of all in-flight batches and their callbacks before sending a request to the broker to end a transaction. This is not technically guaranteed by the documented behavior for Producer::commitTransaction, though; the only guarantees we get from the Javadocs are:

This method will flush any unsent records before actually committing the transaction. Further, if any of the send(ProducerRecord) calls which were part of the transaction hit irrecoverable errors, this method will throw the last received exception immediately and the transaction will not be committed.

If we don't want to rely on this technically-undocumented behavior, we can add some more bookkeeping to ensure this ourselves. Alternatively, we can rely on this behavior and I can remove the comment above about producer callback guarantees and the bookkeeping we already do on that front (which I've done for now).

@hachikuji @guozhangwang do you have thoughts about whether it should be part of the contract for Producer::commitTransaction that all user-supplied callbacks for records within the transaction are fired before the method returns? The use case is tracking record metadata that's only available after the record is ack'd and then reporting it to source tasks, before clearing a local cache of that metadata.

@C0urante thanks for the explanation. I don't think a revised Producer contract would be sufficient though. It's not the blocking per-se, it's the visibility of mutations to commitableRecords done on those threads being correctly synchronized with reads on this line. Producer can't help with that because it doesn't know about commitableRecords, and so the writes and the read could be re-ordered anyway.

I think all we need is a syncrhonized(commitableRecords) around these accesses.

Ah, fair. I was taking a charitable interpretation of the Javadocs, which state:

If multiple threads access a linked hash map concurrently, and at least one of the threads modifies the map structurally, it must be synchronized externally.

In this case, by the time we get to this line, we don't technically have multiple threads accessing the map concurrently; it's guaranteed that this will be the only thread reading from or writing to the map. But it costs very little to add the synchronization barrier, so I've done that.

Also emailed Jason and Guozhang for their thoughts on callbacks with transactional producers. Hoping to hear back soon 🤞

Yeah, I think it's a documentation gap. It is intended for callbacks to be invoked before EndTxn is sent. It would be a little strange otherwise. Worth noting that the callback is just for notification. Even if the callback raises an exception, we won't stop an active commit.

Thanks Jason, filed https://issues.apache.org/jira/browse/KAFKA-13967 to track the producer docs improvement and will have a PR out for it shortly.

Javadocs PR: #12264

tombentley · 2022-06-02T10:31:31Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java

Is this worth a comment?

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTransactionContext.java

.../runtime/src/test/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTaskTest.java

C0urante · 2022-06-02T14:37:45Z

Thanks Tom. Useful comments RE thread safety and an interesting point about transactional producer callbacks. Hoping we can do things the easy way on that front and let the clients library do some of the lifting for us, but if we can't settle on that by mid next week or actively decide against it before then, can do that manually in the ExactlyOnceWorkerSourceTask class instead.

tombentley

Thanks again @C0urante.

tombentley · 2022-06-09T08:31:34Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java

The Producer has it's own TransactionManager. Maybe we should call this TransactionBoundaryManager to avoid confusion. It's also a better description of what this is seeking to abstract over imho.

Good point, no need to complicate IDE class lookups. Changed to TransactionBoundaryManager

mimaison

LGTM, thanks @C0urante

mimaison · 2022-06-10T12:36:56Z

@C0urante I merged #11779 so this needs a rebase before merging

C0urante · 2022-06-10T16:54:39Z

Thanks Mickael, finished the rebase and handled the TODO from #11779 (comment).

CONFLUENT: Sync from apache/kafka trunk to confluentinc/kafka master (13 Jun 2022) apache/trunk: (7 commits) KAFKA-13891: reset generation when syncgroup failed with REBALANCE_IN…(apache#12140) KAFKA-10000: Exactly-once source tasks (apache#11780) KAFKA-13436: Omitted BrokerTopicMetrics metrics in the documentation (apache#11473) MINOR: Use Exit.addShutdownHook instead of directly adding hooks to R…(apache#12283) KAFKA-13846: Adding overloaded metricOrElseCreate method (apache#12121) KAFKA-13935 Fix static usages of IBP in KRaft mode (apache#12250) HOTFIX: null check keys of ProducerRecord when computing sizeInBytes (apache#12288) Conflicts: None

C0urante changed the title ~~KAFKA-10000: Exactly-once source tasks~~ KAFKA-10000: Exactly-once source tasks (KIP-618) Feb 17, 2022

C0urante mentioned this pull request Feb 17, 2022

KAFKA-10000: Per-connector offsets topics (KIP-618) #11781

Merged

C0urante commented Feb 17, 2022

View reviewed changes

C0urante mentioned this pull request Feb 17, 2022

KAFKA-10000: Exactly-once support for source connectors (KIP-618) #10907

Closed

3 tasks

C0urante force-pushed the kafka-10000-source-task-split branch from b588fa0 to 2d5eddd Compare February 17, 2022 21:59

C0urante marked this pull request as draft February 18, 2022 06:35

C0urante force-pushed the kafka-10000-source-task-split branch 2 times, most recently from ce75140 to 1837704 Compare March 3, 2022 17:09

C0urante force-pushed the kafka-10000-source-task-split branch from 1837704 to 48f01b1 Compare May 10, 2022 02:47

C0urante marked this pull request as ready for review May 10, 2022 02:48

C0urante force-pushed the kafka-10000-source-task-split branch from 48f01b1 to 1d95571 Compare May 10, 2022 02:53

C0urante mentioned this pull request May 14, 2022

KAFKA-12497: Skip periodic offset commits for failed source tasks #10528

Merged

3 tasks

mimaison added the connect label May 17, 2022

mimaison reviewed May 17, 2022

View reviewed changes

C0urante force-pushed the kafka-10000-source-task-split branch 3 times, most recently from a36b039 to ca5e253 Compare May 17, 2022 13:16

C0urante force-pushed the kafka-10000-source-task-split branch from ca5e253 to 5f16a11 Compare May 31, 2022 14:55

tombentley reviewed Jun 2, 2022

View reviewed changes

C0urante force-pushed the kafka-10000-source-task-split branch from 5f16a11 to 0a9f7c1 Compare June 2, 2022 14:34

C0urante force-pushed the kafka-10000-source-task-split branch from 0a9f7c1 to 32e0333 Compare June 7, 2022 04:29

C0urante mentioned this pull request Jun 7, 2022

KAFKA-10000: Zombie fencing (KIP-618) #11779

Merged

C0urante force-pushed the kafka-10000-source-task-split branch from 32e0333 to a0e5a64 Compare June 8, 2022 04:50

tombentley approved these changes Jun 9, 2022

View reviewed changes

C0urante force-pushed the kafka-10000-source-task-split branch from a0e5a64 to 14a107e Compare June 10, 2022 02:51

mimaison approved these changes Jun 10, 2022

View reviewed changes

KAFKA-10000: Exactly-once source tasks

87ccec4

C0urante force-pushed the kafka-10000-source-task-split branch from 14a107e to 87ccec4 Compare June 10, 2022 16:48

mimaison merged commit 9e8ef8b into apache:trunk Jun 13, 2022

C0urante deleted the kafka-10000-source-task-split branch June 13, 2022 15:04

yashmayya mentioned this pull request Jul 28, 2022

MINOR: Update comment on verifyTaskGenerationAndOwnership method in DistributedHerder #12451

Merged

KAFKA-10000: Exactly-once source tasks (KIP-618) #11780

KAFKA-10000: Exactly-once source tasks (KIP-618) #11780

Uh oh!

Conversation

C0urante commented Feb 17, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante commented Feb 17, 2022

Uh oh!

C0urante commented Feb 18, 2022

Uh oh!

C0urante commented May 10, 2022

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mimaison May 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mimaison commented May 17, 2022

Uh oh!

C0urante commented May 17, 2022

Uh oh!

mimaison commented May 19, 2022

Uh oh!

C0urante commented Jun 1, 2022

Uh oh!

tombentley left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

C0urante commented Feb 17, 2022 •

edited

Loading

mimaison May 19, 2022 •

edited

Loading