KAFKA-8058: Fix ConnectClusterStateImpl.connectors() method

[C0urante]

This makes the ConnectClusterStateImpl.connectors() method synchronous, whereas before it was implicitly asynchronous with no way to tell whether it had completed or not.

More detail can be found in the Jira.

Tested manually. Seems like a light enough change that even unit tests would be overkill, but if reviewers feel differently tests can be added.
A unit test has also been added for the ConnectClusterStateImpl class.

Committer Checklist (excluded from commit message)

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

[C0urante]

@mageshn @wicknicks @rayokota could you guys take a look when you have a chance? 😃

[wicknicks]

good catch, @C0urante. one comment.

[mageshn]

Nice catch @C0urante. LGTM.

[rayokota]

LGTM

[wicknicks]

Thanks, @C0urante. LGTM.

[C0urante]

@rhauch @ewencp Hi guys! Got a minor bug fix for you; been reviewed and approved by three contributors and seems ready to be looked at by a committer or two. Would appreciate your review when you get the chance.

[kkonstantine]

Nice find @C0urante !
I think using FutureCallback already allows you to do what you implement with the countdown latch.
Also I have a question on whether we need to forward this request to the leader as we do elsewhere.

[kkonstantine]

Forgot to mention that a unit test would be nice (for the whole class even?), since this seems it never worked because it returned immediately.

[C0urante]

@kkonstantine I think I've addressed your two inline comments, and I've added a unit test for the ConnectClusterStateImpl.connectors() method. The other method (connectorHealth(...)) is synchronous and doesn't really look like it warrants an additional test.

As far as the ten-second timeout goes, I'm open to suggestions but it doesn't appear that rebalances or forwarding to master should be a factor.

[kkonstantine]

@C0urante this looks even better now IMO and on par with the rest of the code.
One nit and one more suggested improvement on the tests. Thanks for including those btw. Definitely on the right track.

[C0urante]

@kkonstantine thanks for the review! I've taken your comments into account and pushed the next iteration, which consists of changes to how the ConnectClusterStateImpl is tested. Hope this looks good and addresses your review remarks, but will be happy to keep iterating if not or if there's anything else that needs work.

[kkonstantine]

LGTM!

[rhauch]

Nice find, @C0urante! One really minor suggestion for slight improvement in readability, but not a bit deal.

This has never worked, right? This PR should be relatively easy to backport.

[C0urante]

@rhauch yep, this bug has been present since REST extensions were added to Connect. I've added your suggested change to the PR, let me know if there are any other alterations you'd like.

[rhauch]

Thanks, @C0urante! LGTM.

[ewencp]

Mostly looks good, one comment. This is the sort of thing that should come with list of backports -- the bug appears to already be tagged with at least some info about previous versions that are also affected so we probably don't want to close without backporting.

Regarding timeout, it should really be as conservative as possible without being longer than the minimum of (rebalance timeout, default REST API request timeout).

[C0urante]

@ewencp can you shed some light on why we'd want to take rebalances into account for the timeout? I could be wrong, but I don't think a rebalance would affect the latency of a call to Herder.connectors(...).

[ewencp]

Calling connectors() for DistributedHerder adds a request to the request queue, which are processed serially in the thread that manages cluster membership so they get a consistent view of the state of the Herder. Pretty much all the requests follow that pattern, though it is true some could potentially be optimized to not require it if some other locking were put in place for certain parts of the state.

The thread that processes the requests is also the one managing the member and calling poll() on it, which can block on ensuring membership in the group, which means request processing latency can be affected by rebalances.

[C0urante]

@ewencp thanks for the clarification; I did not realize that membership and request handling were taken care of with the same thread. I've taken your suggestion literally and just made the timeout the minimum of the timeout for the /connectors resource and the rebalance timeout (if running in distributed mode). LMK how this looks!

[C0urante]

BTW, I think we want to backport this to 2.0, which according to the KIP is when Connect REST extensions were introduced and AFAICT the code for the ConnectClusterStateImpl class hasn't been modified since.

[ewencp]

one comment that we could follow up on, but i'm fine merging as is

[ewencp]

nit: i don't love instanceof solutions. might be better to generalize thee rebalance timeout so support a small/zero value for StandaloneConfig.

I wouldn't block merging on this, but this is an anti-pattern, so if there is a cleaner solution, it might be worth exploring.

[rhauch]

@C0urante, what do you think about doing something like the following instead of the instanceof block?

        Integer rebalanceTimeoutMs = config.getInt(DistributedConfig.REBALANCE_TIMEOUT_MS_CONFIG);
        if (rebalanceTimeoutMs != null) {
            herderRequestTimeoutMs = Math.min(herderRequestTimeoutMs, rebalanceTimeoutMs.longValue());
        }

This still uses the DistributedConfig.REBALANCE_TIMEOUT_MS_CONFIG, but otherwise is simply checking for the existence of that property to determine if herderRequestTimeoutMs should be set to a new value.

[C0urante]

@rhauch sgtm! I've added your code snippet verbatim to the PR