KAFKA-7670: Admin client testUnreachableBootstrapServer() flaky test

[stanislavkozlovski]

This test easily fails locally around once every 20-30 runs.

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:262)
at org.apache.kafka.clients.admin.KafkaAdminClientTest.testUnreachableBootstrapServer(KafkaAdminClientTest.java:277)
at sun.reflect.GeneratedMethodAccessor1.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298)
at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292)

After spending considerate time debugging, I found out that the weakly-consistent iterator of futureResponses does not iterate through the two responses at all in the failing runs.

No elements are ever removed from futureResponses except in the MockClient#reset() method which I verified is never called. I think the cause is the weakly-consistent iterator of Java which states:

they are guaranteed to traverse elements as they existed upon construction exactly once, and may (but are not guaranteed to) reflect any modifications subsequent to construction.

I'm still unsure how to solve this in the cleanest way. I'm opening this PR as a chance to discuss it with other people

[ijuma]

You mean that we add a method to the collection while it's being iterated and clear the whole collection even though we never processed some of the elements?

[rajinisivaram]

@stanislavkozlovski There is a timing window in the way metadata response is prepared because send of the metadata request can occur on the admin client network thread before the response is added on the main test thread. But metadata requests should be retried and we would expect this to work on retry. So perhaps the issue in the failing case is that retry is not occurring, so may be worth checking if it is retrying and if not, why it is not retrying.

[rajinisivaram]

MockClient is not thread-safe. So we can't safely call MockClient#setUnreachable after the admin client is started and it starts sending metadata requests. I think in the failing scenario, setUnreachable is disconnecting after a metadata request is sent by the admin client, but we are not retrying because the disconnect response is lost (due to concurrent disconnect/poll on a class that is not thread-safe). I think we need to either rewrite MockClient#setUnreachable to make it thread-safe or rewrite the test so that it creates the admin client only setUnreachable is invoked on MockClient. Rewriting the test is probably the simpler option.

[stanislavkozlovski]

@ijuma No, we don't ever clear the collection. I only found out that while iterating through it in some runs we never find any elements inside.

@rajinisivaram I didn't know that, thanks. I can't reproduce this failure when I add more code for debugging purposes - it just doesn't fail. It's unclear to me why that is. I wonder if it is worth it to change the code such that we instantiate the AdminClient on demand

Update: @rajinisivaram just responded as I was writing this. I'll go with the approach of instantiating AdminClient

[lbradstreet]

This looks good to me. Would it be less error prone to make the AdminClientUnitTestEnv take a MockClient in the constructor (pre-prepared), instead of relying on the user to create the env, then prepare the MockClient, then initiate?

[stanislavkozlovski]

@lbradstreet I guess you meant take in a AdminClient? That would result in way more boilerplate I'm afraid.

You're right that this is a bit more error prone. My thinking is that it should be caught in the first test run and as such maybe not have as big as an impact. I don't have strong feelings about this though

[lbradstreet]

@stanislavkozlovski, you're probably right that it would be significantly more boilerplate. Do all prepare statements have to take place prior to the initiate call? Perhaps those could go through the admin client env, to ensure proper ordering. I don't feel too strongly about it though - what you've done seems fine overall.

[ijuma]

@rajinisivaram can you please review these changes?

[rajinisivaram]

@stanislavkozlovski Couldn't we add a new constructor with a boolean parameter to decide whether adminClient should be created here? So all tests except testUnreachableBootstrapServer don't need to remember to invoke initiate(). And testUnreachableBootstrapServer will be explicitly not creating the admin client, so it will be more obvious that it needs to create it separately. Perhaps that would be less error-prone for the future?

[stanislavkozlovski]

Sounds good. I updated the javadoc to describe that as well

[rajinisivaram]

@stanislavkozlovski Thanks for the updates, looks good. Left just a couple of minor comments.

[rajinisivaram]

nit: This could just call initiate()?

[rajinisivaram]

nit: Perhaps call this method createAdminClient() to be consistent with the flag in the constructor?

[hachikuji]

@stanislavkozlovski I found a simpler explanation for the transient failures. Basically MockClient.poll() is not safe for concurrent sends. Previously we did this inside poll():

        List<ClientResponse> copy = new ArrayList<>(responses);
        ClientResponse response;
        while ((response = this.responses.poll()) != null) {
            response.onComplete();
        }
        return copy;

The responses queue can be modified between the copy and the subsequent poll calls. I changed it to this:

        List<ClientResponse> copy = new ArrayList<>();
        ClientResponse response;
        while ((response = this.responses.poll()) != null) {
            response.onComplete();
            copy.add(response);
        }
        return copy;

After this change, I can no longer reproduce the failure. Of course it's debatable whether we should allow concurrent sends, but these tests currently depend on it.

[stanislavkozlovski]

@hachikuji agreed this seems like a cleaner solution. I also tried and failed to reproduce after using your suggested change. Thanks a lot!

[hachikuji]

LGTM. Thanks for the patch!