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

[C0urante]

Implements support for per-connector offsets topics as described in KIP-618.

Relies on changes from:

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

[C0urante]

Converting to draft until upstream PRs are reviewed.

[C0urante]

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

[tombentley]

Thanks @C0urante! This is a first pass. I haven't looked at tests yet.

[tombentley]

I think this call to get could result in a leaked TopicAdmin, and thus Admin, because offsetStoreForRegularSourceTask doesn't guarantee to invoke topicAdminCreator (it calls it topicAdminSupplier) (when it does, the Admin will eventually we closed via ConnectorOffsetBackingStore.stop).

In any case I think the ownership of the Admin here (responsibility for ultimately closing it) is pretty unclear, but it's not clear to me why it needs to be. So a comment to explain would be great, if a clean up of the logic is no possible.

[C0urante]

This call won't result in a leak because the AbstractWorkerSourceTask class will always close its TopicAdmin during shutdown; see AbstractWorkerSourceTask::close. And, given the idempotent nature of Admin::close, it's fine to close it from both the AbstractWorkerSourceTask and ConnectorOffsetBackingStore classes.

The underlying point about ownership being unclear is fair, though, and the control flow that creates the topic admin in this method is especially convoluted. I've added a couple comments that hopefully clear things up. I wish the logic itself could be better but as it is, things are easy to test and that's taken advantage of by adding a ton of coverage in the WorkerTest suite. Happy to refactor if there's an easily-testable approach that's clearer, though.

[tombentley]

The conditions under which we actually end up calling topicAdminCreator.get() are:

sourceConfig.offsetsTopic() != null && config.connectorOffsetsTopicsPermitted() // connector-specific store
|| config.topicCreationEnable() && sourceConfig.usesTopicCreation() // task needs for topic creation

These are the only circumstances in which topicAdmin.get() would return non-null.

So I wonder if we should just put that logic directly in doBuild(), and have offsetStoreForRegularSourceTask() accept a nullable topicAdminSupplier (which will be guaranteed non-null in the case where it's actually called).

Wdyt?

[tombentley]

Looking at the tests it doesn't seem we have any that assert that the Admin actually gets closed. WorkerSourceTaskTest asserts this, but for this code we're reliant that every invocation of doBuild always returns a WorkerSourceTask which is always stopped. I can see this is pretty hard to test though.

[C0urante]

👍 I added the logic for checking those conditions to doBuild. It's still a little clunky since we're checking regularSourceTaskUsesConnectorSpecificOffsetsStore in two places now (doBuild and offsetStoreForRegularSourceTask), but for testing purposes it's easier to work with since we can continue to unit test offsetStoreForRegularSourceTask independently without having to go through doBuild. And it's certainly more readable than the caching Supplier logic that it replaced.

Good point about ensuring that we stop WorkerSourceTask instances, we definitely do a poor job of that when we fail to build a task. In fact, this has already been documented as a bug with an accompanying fix PR. If that's worth pursuing at some point let me know and I can rebase onto the latest trunk.

[tombentley]

Thanks! When you have time it would be great if you could rebase that PR.

[C0urante]

👍 done.

[tombentley]

Is it correct that if workerGetFuture.cancel was already cancelled, and returned true, then we won't try to cancel connectorGetFuture?

[C0urante]

Ah yes, good point. Switched to | instead of ||, which disables short-circuiting so that it's guaranteed that cancel is called on both futures.

[C0urante]

Also added a comment to this part just in case someone sees that and things that a single pipe instead of a double pipe is just a typo.

[tombentley]

It's quite confusing that timeout is initially in unit, but in this reassignment becomes ms. I think maybe a new timeoutRemainingMs would be clearer.

[C0urante]

👍 done.

[showuon]

I've made a pass, left some comments. Thanks.

[showuon]

nit: indent

[C0urante]

Ack, addressed

[showuon]

nit: indent

[C0urante]

Ack, addressed

[showuon]

Should we set admin = null in this case, like the comment said:
// Forget the reference to the admin so that we won't even try to use the admin the next time this method is called ?

[C0urante]

We want to keep throwing the exception here if the method is called again; nullifying the admin field would stop that from happening. That does remind me though, we should probably copy the same error message from here into the code path for starting connector offset stores. I've done that in my latest push; LMKWYT.

[showuon]

SGTM. Thanks.

[showuon]

Could user overrides the transactional.id to null? Should we add this test case?

[C0urante]

I don't think that can happen without causing other issues, since the user-supplied configs have come from either the JSON REST API or the Java properties worker file. The former doesn't deserializes things into a literal null and although the latter does, as of #11333, we fail validation for connector configs in that case.

But, given how little additional effort it is, I don't see a problem with adding that check. At the very least, it helps prevent a 500 instead of a 400 response during config validation.

[tombentley]

This constructor kinda confuses the ownership of the Admin client. I think things are cleaner when the TopicAdmin instantiates (and thus owns the admin). Note, it looks like there are no callers for TopicAdmin.admin. It seems that the call sites in doBuild could simply pass the map of configs (and the bootstrapServers looked up from that), rather than instantiating the admin and then passing it to the TopicAdmin.

Obviously the test code has slightly different requirements, meaning we still need this constructor. I did also wonder whether we could also get rid of bootstrapServers by defining toString on KafkaAdminClient and using that for the logging and exceptions here in TopicAdmin. Perhaps that's worth a followup PR at some point, (though perhaps there are benefits to hiding bootstrap servers from receivers of clients).

[C0urante]

Good points. I've:

• Demoted the visibility of the constructor that accepts an Admin instance back to package-private
• Documented that it's the caller's responsibility to close the topic admin in order to release the resources allocated for its Admin
• Removed the unused admin method
• Switched over to the TopicAdmin(Map<String, Object>) constructor in the Worker class

I originally did this refactoring because I was managing the lifecycle of the Admin differently, but kept part of it because it made testing much easier when all you had to pass in was an Admin and a String instead of an Admin and an entire config map. It also seemed like an anti-pattern to have a constructor that accepted both an already-instantiated client object, and the configuration for that client object.

[C0urante]

Hey guys--thanks for the reviews, really appreciate the rapid responses here. I found a bug that's been a bit trickier to solve than expected and have had little time to work on it this week. I plan to push the next draft by Friday at the very latest.

If it matters, the bug is that the offset stores for regular (non-exactly-once) source tasks, and source connectors, are never started. I'm planning on fixing that first, then adding an integration test case to #11782 to simulate a soft downgrade where someone disables exactly-once support on their worker after creating a connector and letting it run for a bit, and finally, manually auditing the changes for KIP-618 to catch any other potential bugs related to improper initialization or cleanup of resources.

[C0urante]

Pushed the next draft which should fix the offset store startup bug and address outstanding review comments.

[C0urante]

I've completed the remaining follow-up tasks.

Audit of resource allocation and cleanup:

• Documented ownership expectations more clearly with the KafkaBasedLog and KafkaOffsetBackingStore classes, especially with regards to their underlying Kafka clients
• Found and fixed a small bug in ExactlyOnceWorkerSourceTask where its transactionMetrics field was never closed
• Found a long-lived bug where HeaderConverter instances are never closed; given that this was not introduced by any of the changes for KIP-618, I've filed https://issues.apache.org/jira/browse/KAFKA-14007 and would like to address it separately
• Found no other resource allocation, initialization, or cleanup bugs in the code paths related to KIP-618

Integration testing

• Added a new test to KAFKA-10000: Integration tests (KIP-618) #11782 (or rather, augmented an existing one) for running a source connector with a separate offsets topic with exactly-once support disabled

[C0urante]

Given that we close the TopicAdmin for source tasks in AbstractWorkerSourceTask::close here, this may seem redundant. However, I've chosen to keep it in for two reasons:

1. The WorkerConnector class has no access to the TopicAdmin used for its offset store; we'd have to add another constructor parameter to that class and manually close the TopicAdmin during shutdown if we removed this line
2. It seems safer to leave this in. Although Autocloseable::close is not required to be idempotent, it is strongly recommended, and since all current calls to ConnectorOffsetBackingStore::stop are wrapped in Utils::closeQuietly anyways, the risks are low. On the other hand, by leaving this out, we would make it easier to introduce a resource leak in the future, which would have more serious consequences.

[tombentley]

Thanks @C0urante, I left a few more comment, but they're nits at this stage.

[tombentley]

I guess we could improve this slightly by not comparing as strings, but as Set<String>. That way at least the order wouldn't matter, which it looks like it does right now.

I've wondered whether we should add some kind of cluster.id to the client configs which could be used to assert that the client was connected to the expected cluster. If we did do that, and they were specified in the configs here, then you could safely know at this point whether they were the same cluster without needing to connect at this point.

[C0urante]

True, we can make this order-independent by comparing sets. Not too much work, so I've done that.

I have a long-term vision of relocating this logic into the WorkerTask class (or its subclasses) so that we can ping the remote Kafka cluster, grab its ID there, and decide at that point whether or not we need to construct an additional offsets store, all without having to worry about blocking up the herder tick thread. This would obviate the need for users to do any additional work at all, but it would require a significant refactoring of this part of the code base, so I've held off on it for now.

[tombentley]

Might newTopicSpecification be a better name? It's just the "description" is a bit ambiguous: it could be describing an existing topic.

[C0urante]

Ack, done

[tombentley]

Great doc, but maybe we should also have some on those constructors themselves?

[C0urante]

👍 good point, done

[C0urante]

Discovered this bug while working on #12307 for https://issues.apache.org/jira/browse/KAFKA-14006.

[showuon]

Made another pass, left some minor comments. Overall LGTM. During the review, I found a bug KAFKA-14012. But it's not related to this change and can be fixed in another PR. Thanks.

[showuon]

We've add all bootstrap servers into producerBootstrapServers set in L1672, why should we re-create another hashSet here?

[C0urante]

Good catch, left in accidentally. Removed

[showuon]

OffsetBackingStore#start() -> OffsetBackingStore#stop() ?

[C0urante]

Good catch, done

[showuon]

and [close] the {@link TopicAdmin} used by that store.

[C0urante]

Ack, done

[showuon]

Why did we log trace to this primary store error, but warn to the secondary store error in L290?

[C0urante]

The failure to write to the primary store will probably be logged elsewhere at a higher level, since we report it to the callback. We swallow failures to write to the secondary store, so this is the only chance we get to log those.

[showuon]

I see. Thanks.

[showuon]

unused variable?

[C0urante]

Used here.

Added this. on writes to this field and others in the configure method to make it easier to identify instance vs. function variables.

[showuon]

LGTM!

[showuon]

Failed tests are unrelated:

Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testSnapshotOnlyAfterConfiguredMinBytes()