KAFKA-15563: Provide informative error messages when Connect REST requests time out

[C0urante]

Jira

There are three common scenarios when Kafka Connect REST requests time out:

1. A connector operation (such as a call to Connector::validate) has blocked
2. A distributed worker's herder tick thread is blocked (which can be for a variety of reasons, including failure to reach the group coordinator, failure to complete startup, and failure to reach the config topic)
3. A distributed worker has to issue its own request to another worker, and that request blocks (note that this does not include transparently-forwarded requests, and currently only includes requests to submit task configurations and perform zombie fencing)

This PR adds error messages to REST requests that help capture which operations are causing the worker to be unable to respond in time. The error messages include a description of the action the worker is blocked on, and a human-readable timestamp of when that action began.

Some rejected alternatives:

• One or more new JMX metrics (harder for users to access, would require a KIP, and would likely be just as difficult to implement, especially when trying to accommodate blocks in both connector interactions, which can take place on an isolated per-request basis, and in a distributed worker's herder tick thread, which can affect multiple REST requests)
• Exposing the stack trace of the distributed worker's herder tick thread as part of the error message for timeouts (does not help with requests that are not blocked because of the herder tick thread, and may present a security risk by including source code for user-provided plugins such as ConfigProvider instances)

During review, it may be helpful to focus on these files first (as they contain the internal API for recording and reporting callback stages):

• HerderRequestHandler and ConnectorPluginsResource (same change applied across both files)
• Callback
• ConvertingFutureCallback
• Stage
• StagedTimeoutException
• TemporaryStage

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[C0urante]

@gharris1727 @yashmayya I'd be interested in your thoughts on this one if/when you have time. I've heard several complaints from Kafka Connect users about REST requests timing out and want to make that easier on everyone involved, but I'm also conscious that if we're not careful about this kind of change, it can present a serious maintenance burden on the project.

I've tried to implement this in a way to reduce impact on the signal-to-noise ratio in critical parts of the codebase (i.e., DistributedHerder) as much as possible, but even with that, I'm not certain that this is worth the ROI. Would welcome some outside perspectives on whether this is worth adding, and if not, alternatives we can explore.

[C0urante]

I've run into some issues with integration tests on Jenkins. I'm experimenting now with some potential fixes, and the rest of the PR should still be ready for review in the meantime.

We should make sure that there's at least one (if not several) Jenkins run with no Connect integration test failures that appear caused by these changes before merging.

[gharris1727]

Hey @C0urante thanks for taking this on!

This is certainly an interesting first pass at the problem, and I share your concerns about the maintenance burden and SNR.

One of the things that I think may be causing some friction for the implementation is that the granularity of the Stages doesn't match the granularity of the actual tick thread & herder requests. What I mean to say is that one function has multiple different stages associated with it, instead of there being a 1:1 relationship.

Perhaps the design could be simpler if each code-block was given a description of what it did statically, like a new String argument to addRequest. This would be significantly less granular than the current implementation, but we could refactor the control flow of current requests into multiple smaller requests to improve the granularity. This could also be beneficial for the structure of the herder tick thread depending on how we refactor it.

We could either wait to implement the improved error messages after such a refactor, or land the low granularity error messages with a plan to improve them. Refactoring the herder requests into smaller sub-units may also be too involved and have even worse ROI, i'm not sure.

Also just some smells that I picked up on:

• I think having the stage be both a property of the callback and the herder was a bit odd; perhaps the HTTP request thread can just query the herder directly when the timeout occurs.
• Related, the linear "set the Stage for all queued requests" in seems bad to me. The queue is supposed to be small of course, but if timeouts are happening, it may be happening due to high tick thread contention/request load. In that situation, having every bit of progress on the tick thread need to update the queued requests is ultimately a quadratic cost, potentially making tick thread contention worse.
• Setting the stage to null after completing some operation completes leaves the opportunity for race conditions to degrade the error messages. Between slow operations (presumably when the thread is Running) the error message wouldn't have the context for what slow operation just finished, even though if the timeout happened a fraction of a second earlier, it would be displayed. This is acceptable if the wait times are large, the time spent Running is small, and we never forget to add a Stage for a slow operation. But if any of those aren't true, the diagnostic power of this feature decreases.
• Statistically, the operation which is running when the timeout occurs is the most likely to have caused the timeout, but all of the actions running since the request started will contribute. To use an analogy: If you're experiencing slowness on your computer, you can sample the CPU and figure out that Program A happens to be running at that moment. But wouldn't it be more informative to know that over the past 30 seconds, 99% of the CPU was spent on program B, and you just got "unlucky" and observed program A was running? If you can only sample the CPU, you'd have to take multiple samples (issue multiple REST requests) to learn what the real problem was.

[C0urante]

Thanks Greg. To be clear, this isn't really a first pass--it's the first published one :)

With regards to giving each request a single scope--this would fail to capture blocking operations outside of the request (such as other requests, ensuring group membership, and reading to the end of the config topic). I suppose we could do something similar where we decompose tick() into a series of either 1) method invocations that have a single scope or 2) herder requests that have their own scopes (at least one, possibly more).

As far as breaking down herder requests into several smaller requests goes, I don't think it's wise to do this when not absolutely necessary. The only reason we added that kind of decomposition in #8069 was to avoid blocking on interactions with connector plugins that could otherwise disable the herder's work thread. If there are functional benefits to decomposing operations further, then we can and should explore those, but the risk of regression alone outweighs the benefits of doing that solely for cosmetic benefits, not to mention the additional implementation complexity and hit in readability. I don't think this rules out the possibility of trying to add some more scope-based structure to herder requests so that, for example, every request is a series of single-scope method invocations, but I'm also not sure that that approach is realistic since it may increase the odds of inaccurate error messages.

I'm also trying to optimize for user benefit with our error messages here. Telling someone that the worker is blocked reading to the end of the config topic is useful; telling someone that the worker is blocked while deleting a connector is less useful, especially if that error message is delivered in response to the request that triggered connector deletion. If we do opt for a different approach, I'd like it if we could add fine-grained error messages with the first PR, and not as a follow-up item.

With regards to the so-called "smells":

I think having the stage be both a property of the callback and the herder was a bit odd; perhaps the HTTP request thread can just query the herder directly when the timeout occurs.

I explored this approach initially, but it comes with a pretty severe drawback: there's no guarantee that a request that took too long timed out because it was blocked on the herder thread. Connector validation is the biggest example of this, though there's also connector restarts, which currently block on the connector actually completing startup. In cases like these, it could be misleading to users to tell them what the herder thread is doing when that information has no bearing on why their request timed out.

Related, the linear "set the Stage for all queued requests" in seems bad to me. The queue is supposed to be small of course, but if timeouts are happening, it may be happening due to high tick thread contention/request load. In that situation, having every bit of progress on the tick thread need to update the queued requests is ultimately a quadratic cost, potentially making tick thread contention worse.

I'm happy to do some benchmarking if we're worried about the performance impact of this approach. I'm personally a little skeptical that setting a new value for the ConvertingFutureCallback.currentStage field is likely to be significant, even if it is marked volatile. It's also a little strange to refer to this as a quadratic cost, since I can't envision a reasonable version of this approach that records any more than a few dozen herder tick thread stages, even with maximum granularity.

Setting the stage to null after completing some operation completes leaves the opportunity for race conditions to degrade the error messages. Between slow operations (presumably when the thread is Running) the error message wouldn't have the context for what slow operation just finished, even though if the timeout happened a fraction of a second earlier, it would be displayed. This is acceptable if the wait times are large, the time spent Running is small, and we never forget to add a Stage for a slow operation. But if any of those aren't true, the diagnostic power of this feature decreases.

I think the key point here is that we need to be careful to record stages for potentially-slow operations, which I've tried to do here. But it's true that in the future, if we move things around and forget to wrap a blocking operation with, e.g., a TickThreadStage, then users will lose insight into operations. I haven't found a good alternative yet that doesn't run the risk of reporting incorrect operations. Perhaps we could augment the error message not just with when the operation started, but also, if applicable, when it completed? That way, in a racy scenario like one where we block for 89 seconds reading from the config topic and then time out a second after, users could still get reasonable insight into the problem ("the last thing the worker was working on was <some operation>, which began at <time> and ended at <time>"). Regardless, I do acknowledge this as a drawback of the current approach.

Statistically, the operation which is running when the timeout occurs is the most likely to have caused the timeout, but all of the actions running since the request started will contribute. To use an analogy: If you're experiencing slowness on your computer, you can sample the CPU and figure out that Program A happens to be running at that moment. But wouldn't it be more informative to know that over the past 30 seconds, 99% of the CPU was spent on program B, and you just got "unlucky" and observed program A was running? If you can only sample the CPU, you'd have to take multiple samples (issue multiple REST requests) to learn what the real problem was.

Aha! I think the last sentence is actually a great, succinct way of justifying this approach. The first reaction most people have to a timeout message like this is to assume (or hope, or pray) that it's a transient problem, and issue one or more requests to retry the operation. The idea here is to provide more and more benefit as the number of retries increase. If you see a single timeout error and then the next request succeeds, it doesn't matter too much why the first one timed out. On the other hand, if your worker is borked and every request is hanging, then this approach (especially with the granularity it provides and the emphasis it makes on capturing potentially-blocking operations) gives a nice head start to diagnosing the problem without having to pore through worker logs first.

Still, if we want to be even more informative, we could augment the currently-proposed error message not just with the current stage, but a history of stages and their start (and possibly end) times. IMO this is a bit much for the first patch and it'd be nice to wait for user feedback to see if it's actually necessary, but I wouldn't be opposed to it if others (including you) feel differently.

Overall, I think it might be helpful to provide a few principles I've tried to satisfy with this approach and see if they're agreeable:

• It's acceptable to sometimes provide no error message
• It's much less acceptable to provide an inaccurate error message
• We should optimize for cases when the worker has been blocked for a long time over cases when the first request times out

[gharris1727]

it's the first published one :)

This is a much better way to put it, and captures what I was going for :)

I suppose we could do something similar where we decompose tick()

👍

If there are functional benefits to decomposing operations further, then we can and should explore those...
but the risk of regression alone outweighs the benefits of doing that solely for cosmetic benefits, not to mention the additional implementation complexity and hit in readability.

I don't think there would be a functional change, but I think that readability could be improved. I do understand your concern about regressions, and I worry if this code has become too difficult and risky to change to attempt such refactors. It will only get worse as time goes on.

there's no guarantee that a request that took too long timed out because it was blocked on the herder thread.

Thanks for this clarification. A request can be blocked on various threads during it's lifetime, not just the herder tick thread.

I'm happy to do some benchmarking if we're worried about the performance impact of this approach. I'm personally a little skeptical that setting a new value for the ConvertingFutureCallback.currentStage field is likely to be significant, even if it is marked volatile.

I completely agree, this single operation should be lightweight such that we can perform thousands of them and not meaningfully change the runtime of the overall request. You could do benchmarking to prove that, but I don't think you have to spend the time doing that if you change the asymptotic behavior.

It's also a little strange to refer to this as a quadratic cost, since I can't envision a reasonable version of this approach that records any more than a few dozen herder tick thread stages, even with maximum granularity.

I should have clarified: it's quadratic in the number of requests, not the stages per request. Here's the situation:
Lets say every request is identical and goes through K Stages, and that we are processing N requests in total, but they come in two at a time, such that there is only Q=1 element in the queue at a time.
For each request (N):
For each stage (K):
For each element in the queue (1):
set the queued stage
This ultimately does N*K work.

If the requests all come in at once, and the Q=N initially, then this happens:
For each request (N):
For each stage (K):
For each element in the queue (N, N-1, .... 1):
set the queued stage
This ultimately does 1/2 * N^2 * K work (1/2 because the queue gets shorter after each request).

So the longer the queue is, the more work the herder has to do. It's a very weak positive feedback loop.

I think you can eliminate it with a layer of indirection. If instead of storing a Stage in the callback, you could store a Supplier that retrieves the tickThreadStage. You would set the supplier once when the request is added to the queue, and evaluate the supplier when a timeout occurs. setTickThreadStage doesn't need to update each of the individual requests, because if a timeout occurs they will retrieve the latest one via the getter. Once a request leaves the queue, it can change the Supplier to no longer blame the tick thread for timeouts.

"the last thing the worker was working on was , which began at and ended at "

I think this is reasonable, and I'll leave it up to you if you want to implement it. Modeling a "Stage" as an interval of time rather than a single instant seems natural, even if the REST error will mostly only see in-progress intervals.

Still, if we want to be even more informative, we could augment the currently-proposed error message not just with the current stage, but a history of stages and their start (and possibly end) times. IMO this is a bit much for the first patch and it'd be nice to wait for user feedback to see if it's actually necessary, but I wouldn't be opposed to it if others (including you) feel differently.

Sure, we can push full traces to an optional follow-up. Full traces would be more resource intensive and possibly too verbose for the REST API.

[C0urante]

Okay, I've pushed a couple new commits that:

• Introduce the notion of a completion time for a callback stage
• Add granularity to the callback stages for the distributed herder's tick thread

I know that this doesn't cover everything we discussed, but I did give the rest a try.

I explored an approach where we defined broader tick thread stages (declaring them in DistributedHerder::tick and not in methods it invokes, and following a similar approach for herder requests). This turned out to be infeasible because of the control flow during a rebalance, where the herder invokes WorkerGroupMember::poll or WorkerGroupMember::ensureActive, which in turn can end up invoking DistributedHerder.RebalanceListener::onRevoked, which in turn can perform operations that warrant a distinct tick stage from, e.g., "ensuring membership in the cluster".

Instead, I've tried for an approach where the tick thread stages are defined as narrowly as possible, and only around operations that we can reasonably anticipate will block. This does slightly increase the odds of a stage being completed when a request times out, but since the information about that stage isn't lost anymore, the fallout from that scenario is limited.

I also experimented with the Supplier<Stage> approach to reduce the runtime complexity of stage tracking for herder requests, but found that this was more difficult to unit test. Instead of being able to track the set of all recorded stages for a callback, we would have to manually query the Supplier after each anticipated herder stage update, which is more work and can fail to collect some stages if not queried at the correct time (especially if it's too difficult to query at a specific point in time during a call to DistributedHerder::tick). Since we both agree that performance shouldn't be a concern here, I hope this is acceptable.

I've also verified with three consecutive Jenkins runs that the new integration test should finally be flake-free.

Let me know what you think; hoping that this is closer to acceptable, though I suspect there's still work left to be done (including adding more unit testing coverage).

[C0urante]

@gharris1727 @yashmayya would either of you have a moment to take a look at this one? Hoping to merge in time for 3.7.0 if possible. Thanks!

[gharris1727]

Thanks Chris!

I like the augmentation of the BlockingConnectorTest, and the additional ConnectWorkerIntegrationTest.

The new error messages in HerderRequestHandler and ConnectorPluginsResource are also very clear and to the point.

[gharris1727]

Is this the only error message possible when shutting down kafka, or the most common?
I would expect that an error about ensuring membership in the cluster could appear.

[C0urante]

Surprisingly, this is actually the first operation that will block indefinitely if the Kafka cluster goes down.

We proactively revoke all connectors and tasks on the worker if we're unable to reach the Kafka group coordinator for long enough, which in turn causes the worker to flush the status store, which blocks until Kafka becomes available again.

I'd be worried about this (the herder thread getting blocked, which is usually pretty awful), but for a system designed to integrate with Kafka, it doesn't seem so bad or so unreasonable for our availability to be tied to the targeted Kafka cluster's.

I've augmented the stage description to hopefully shed a little more light on the issue

[gharris1727]

Ah I see. So the polling the group coordinator for up to ... or until interrupted stage is only temporary and flushing updates to the status topic is permanent, so it'll always eventually get stuck on this stage.

Do you think it's possible for the preceeding Thread.sleep() to cause a flake here, if the worker is in the "polling the group coordinator" stage for too long? Perhaps we could replace the sleep with a wait-until-condition that repeatedly makes the request until the flushing status store error appears.

[C0urante]

Ah, good idea! Added assert-with-retry logic to the assertTimeoutException utility method.

I've also tweaked how we track stages when the herder is polling the group coordinator. Previously, when the WorkerGroupMember and WorkerCoordinator methods only accepted a Runnable, those stages would never be closed. Now, those methods accept a Supplier<UncheckedCloseable> that, at the moment, always returns a Distributed.TickThreadStage, which can be used with a try-with-resources block to automatically register and complete tick thread stages.

I know that I've used UncheckedCloseable incorrectly in the past; I believe this time doesn't suffer from any of the issues my previous usages did. But if the result is still undesirable, we can explore alternatives like introducing a new interface.

[gharris1727]

LGTM, thanks for the improvement Chris!

I hope this will lead to faster time-to-remediate for incidents involving unresponsive workers, and facilitate giving feedback to the framework & plugin developers about improving consistently slow operations.

[C0urante]

I've just noticed that this part is no longer necessary since we've merged a fix for KAFKA-15693.

I'll remove this comment and change the value to zero. If the test continues to pass on Jenkins, should be safe to merge.

[C0urante]

Thanks Greg! I've made one final tweak to the integration tests. Hoping this doesn't affect test stability; will merge if things look alright on Jenkins.

[C0urante]

All CI failures appear unrelated, merging.