KAFKA-10000: Zombie fencing (KIP-618)

[C0urante]

Implements the zombie fencing logic described in KIP-618 (except for the portion already covered by #11778).

Relies on changes from:

• KAFKA-10000: Add producer fencing API to admin client (KIP-618) #11777
• KAFKA-10000: Use transactional producer for config topic (KIP-618) #11778

Note that none of the logic here actually causes zombie fencing to take place, it only implements the internal API required to perform zombie fencing. Downstream PRs will actually put this logic into play.

[C0urante]

Converting to draft until upstream PRs are reviewed.

[C0urante]

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

[tombentley]

Thanks @C0urante, made a first pass.

[tombentley]

There's a resource leak if the whenComplete never calls the passed lambda. I think you should be able to call admin.close in a catch(Exception).

[C0urante]

Good catch, done.

[tombentley]

This pattern of swapping out class loaders is pretty common, but also a little verbose. Perhaps Plugins could expose a withClassloader(ClassLoader) method that returned an AutoClosable, so that call sites like this could use try-with-resources and the compiler could warn about leaking resources?

[C0urante]

Ah yeah, been toying with that idea for a while but never got around to trying it out. Works pretty well in this case; the one wrinkle is that the signature for AutoCloseable::close includes a checked exception. I've added a new (internal) LoaderSwap class that implements AutoCloseable and removes that checked exception to address that.

If this looks good, we can retrofit other parts of the code base to leverage it in a follow-up.

[tombentley]

what happens if it's not a source connector?

[C0urante]

The callback is invoked with an error (added to Javadocs)

[tombentley]

I wonder if runOnTickThread might be a better name, since it more explicitly describes what it's doing?

[C0urante]

That works, yeah 👍

[tombentley]

Surely we should always invoke the callback, even on success, since that's the contract for Callback?

[C0urante]

This follows the same pattern as AbstractHerder::maybeAddConfigErrors, which accepts a Callback but only invokes it on errors. This is useful if you'd like to establish some reusable logic that terminates control flow for a method and reports an error to a callback if something goes wrong, but otherwise allows control flow to continue and possibly fail later.

I'll take a page out of AbstractHerder::maybeAddConfigErrors's book and add Javadocs making note of this fact.

[tombentley]

Thanks for the explanation and the Javadoc.

[tombentley]

So we don't consider 1 minute 'very long'?

[C0urante]

It seemed reasonable considering how vital being able to reach the config topic is to the health of a Connect worker, and that the penalty for failure here is that a task will fail to start. But given the existing workerSyncTimeoutMs field and its use, it seems better to just follow that precedent and use that value to dictate how long we're willing to wait to reach the end of the config topic in most cases.

[tombentley]

There's a lot of repetition of this if (!writeToConfigTopicAsLeader()){ throw new ConnectException} pattern. In fact it look like all invocations of writeToConfigTopicAsLeader are of this form. So what not just put the if/throw within writeToConfigTopicAsLeader?

[C0urante]

I pushed a change to #11778 that basically does this; will rebase and update the new config topic writes introduced in this PR accordingly. One noteworthy difference now is that the exception message is always the same regardless of which operation failed; I tried to make it generic and user-friendly enough to work with that, but if that doesn't work well enough, we can add a message parameter to this method and use it as part of the message for the exception that gets thrown on failure. BTW, it might be more helpful to leave comments about this topic on that PR, but I'll do my best to handle them either way.

[tombentley]

Is it OK to not invoke the callback in the case where we weren't leader?

[C0urante]

Yes, although the internal API for this is a little convoluted:

• addRequest accepts an action (a Callable<Void>) and a callback (a Callback<Void>)
• When requests submitted to addRequest are run, the callback is always invoked after they complete; if they throw an exception, it's invoked with that exception, and if they don't, it's invoked with null for both parameters
• The callback we pass to addRequest here is the result of wrapping the callback given to the deleteConnectorConfig method in the forwardErrorCallback method, which causes it to be invoked if and only if an exception is thrown when the request is run
• As a result, if we throw any exceptions from the action that we pass to addRequest, they're guaranteed to be passed to the callback supplied to deleteConnectorConfig

Although I think it's cleaner to throw exceptions instead of invoking Callback::onCompletion with an exception and then doing a return null, for consistency's sake, it's probably better to do the former, since that's the existing pattern. I'll address this first in #11778 and then add it here in the subsequent rebase.

[C0urante]

On second thought, I think it's probably fine to leave things as they are without adding a manual invocation of Callback::onCompletion and a return null. Yes, writeToConfigTopicAsLeader may throw an exception, but so could writes to the config topic before changes for this KIP were made (such as here, here, and here).

If we were throwing an exception from within the body of the herder request instead of a method that the request invokes, it'd make sense to change that to instead be a manual invocation of the callback with the exception. But just calling a method that might throw an exception is different, and follows existing precedent in the code base without having to jump through special callback-related hoops.

[tombentley]

This could block indefinitely, since KafkaConfigBackingStore calls configLog.readToEnd().get(), which seems at odds with the it should not block for very long requirement.

[C0urante]

Good point, replaced configLog.readToEnd().get() with configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS), which is used everywhere else in the KafkaConfigBackingStore where we read to the end of the log to ensure that writes that we just performed have landed. It comes with the downside that it makes zombie fencing rounds more frail, but that's better than squatting indefinitely on the herder thread.

I also fixed another potential blocking issue around this area by shifting the call to onZombieFencingSuccess (or rather, the registration of it as a follow-up to the future returned by Worker::fenceZombies) into a separate method that can then be invoked after the ZombieFencing object has been constructed and the lock on the DistributedHerder instance has been relinquished.

[tombentley]

I find the method name a bit confusing, because it sends in either case. Perhaps something like sendPossiblyFencibly would be better, wdyt?

[C0urante]

sendPossiblyFencibly (fencably?) does work but it's a bit verbose. Do you think sendPrivileged works? It refers to the concept inherited by the ConfigBackingStore interface and its claimWritePrivileges method, and the write itself is technically privileged in that it should only ever be performed by the leader, even if those privileges are only enforced when the backing store is configured to use a fencable producer.

[C0urante]

Changed to sendPrivileged, can change to something else if desired

[C0urante]

Thanks Tom, some great catches. Going to rebase tomorrow or Thursday which should address the one or two outstanding comments; everything else should be addressed now and ready for another round.

[mimaison]

Thanks @C0urante for the PR. I've not looked at all the tests yet but it looks pretty good overall!

[mimaison]

What about fenceZombieSourceTasks()? I find fenceZombies() a bit too generic

[C0urante]

Fine by me 👍

[mimaison]

Do you already have the PR that clears this?

[C0urante]

No, but I will update #11782 to remove it as soon as this is merged.

[mimaison]

Do we need this? Also does this method need to return Response?

[C0urante]

This is to force a 200 OK response instead of a 204 no content response, which would be returned otherwise. I'd just use a 204 except the KIP specifies that this endpoint should "serve an empty-bodied 200 response" and I wanted to stick to that.

Given that this endpoint is internal and it's a tiny detail, I'd be fine with switch to a 204 response if it's alright with you.

[mimaison]

As far as I can tell the other internal endpoint returns 204 so I'd be in favor of doing the same here

[C0urante]

👍 done

[mimaison]

Nit, let's keep the new line

[C0urante]

🤦 sorry, done.

[mimaison]

This type of small cleanups are really appreciated, thanks!

[mimaison]

I was confused for a moment as I remember seeing these methods in another PR. I see this PR has conflicts so this must be the reason and they'll disappear from here once this is rebased on trunk

[C0urante]

Yep, exactly 👍
Going to try to do the rebase today, but may not be able to finish by EOD as it's going to be fairly involved.

[C0urante]

Ah whoops, those changes were made on #11780, which hasn't been merged yet, so a rebase isn't going to automatically draw them in. I'll do the change manually here but there may be other small changes in not-yet-merged PRs that don't get pulled in here. It should be fine as those changes are included in whichever PR gets merged last.

[C0urante]

Rebase complete; should be resolved now.

[mimaison]

Is this going to be called from other places in the remaining PRs? If not we could get rid of it

[C0urante]

It's used in integration tests later on: https://github.com/C0urante/kafka/blob/3d65e799925096d519b4adf906be05cba70addeb/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java#L828

[mimaison]

Would task_count or even count (like state) be clearer?

[C0urante]

I think given the key format ("tasks-count-<connector>") this is probably fine, and the name of the field is also specified in the KIP. But similar to the 200 vs. 204 HTTP response for the fencing endpoint, this is internal and a small detail, so I can change it if we agree that this kind of detail doesn't need to precisely match what's in the KIP.

[tombentley]

I wonder whether the key format should assume that a count is involved, or whether it should be named for the purpose to which it's being put (zombie fencing). e.g. maybe tasks-fencing-<connector> is a better key, with task_count at the field name for this V0 schema which happens to use just the count as the implementation?

[C0urante]

🤷 gave that a try. @mimaison LMKWYT

[mimaison]

newer?

[C0urante]

Haha yep, caught and fixed this in an upstream PR that's since been merged. Will pick up in the rebase.

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

Line 323 in a6c5a74

"support for source connectors, or use a newer Kafka broker version.",

[C0urante]

Rebase complete; should be resolved now.

[mimaison]

Isn't ordering still guaranteed with retries when idempotency is enabled?

[C0urante]

Yep, this got fixed in #11778, which just got merged. A rebase should take care of this.

[C0urante]

Rebase complete; should be resolved now.

[mimaison]

Thanks for the quick updates. I'll try to make another pass tomorrow

[tombentley]

Thanks for the fixes @C0urante! I've left a bunch more comments, but these are nits, and assuming you agree with them this now LGTM.

[tombentley]

Trivial point, but swapping the order of these parameters would match the order that they're used in test(), and, at the constructor call site, the method, path ordering matches how these things appear in an actual HTTP request.

[C0urante]

👍 done

[tombentley]

I guess we could also swap there parameter order here too?

[C0urante]

👍 done

[tombentley]

Thanks for the explanation and the Javadoc.

[tombentley]

Can we document that access is protected by this object's monitor.

[C0urante]

👍 done

[tombentley]

This can be final too, I think

[C0urante]

It's initialized in start(), not in the constructor.

[tombentley]

I wonder whether the key format should assume that a count is involved, or whether it should be named for the purpose to which it's being put (zombie fencing). e.g. maybe tasks-fencing-<connector> is a better key, with task_count at the field name for this V0 schema which happens to use just the count as the implementation?

[tombentley]

We should include the connector name, I think.

[C0urante]

It's implicitly included in the record key but that info is redundant. Updated to just use the connector name and make the message clearer

[tombentley]

We're adding this else if clause to a method that's now ~250 lines long. I think we can factor the block of each if and else if into its own method.

[C0urante]

Yeah, we've definitely reached that point 👍

While doing this decomposition I kept the bodies for each new method identical to the if/else if blocks that they were extracted from, with these exceptions:

• Log messages that include the record key are adjusted to use the connector name in its place (this doesn't drop any information)
• Calls to Object::getClass for logging messages are all converted to calls to the newly-introduced and null-safe className method
• The @SuppressWarnings("unchecked") annotation is removed from method signatures and is instead added only to assignments within the method bodies that require it

[mimaison]

LGTM

[C0urante]

Thanks Mickael 👍