[KAFKA-13538] Unexpected TopicExistsException related to Admin#createTopics after broker crash

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 2.8.0
• Component/s: build
• Assignee: Unassigned
• Reporter: Haoze Wu
• Created: 12/Dec/21 21:31
• Updated: 20/Apr/22 16:41

Description

We were using the official Kafka Java API to create a topic in a Kafka broker cluster (3 brokers):

CreateTopicsResult result = admin.createTopics(...);
... = result.all().get(); 

The topic we create always has replication factor = 2, and partition = 2. If one of the brokers crashes for some reason and the client tries to create a topic exactly in this crashed broker, we usually observe that the client may suffer from a delay of a few seconds due to the disconnection issue, and then the client automatically connects to another broker and creates the topic in this broker. Everything is done automatically in the client, under the code of `admin.createTopics(...)` and `result.all().get()`.

However, we found that sometimes we got `TopicExistsException` from `result.all().get()`, but we had never created this topic beforehand.

After some investigation on the source code of client, we found that this issue happens in this way:

1. The client connects to a broker (say, broker X) and then sends the topic creation request.
2. This topic has replication factor = 2 and partition = 2, so broker X may inform another broker of this information.
3. Broker X suddenly crashes for some reason, and the response for the topic creation request has not been sent back to the client.
4. The client eventually learns that broker X crashes, but never gets the response for the topic creation request. Thus the client thinks the topic creation request fails, and thus connects to another broker (say, broker Y) and then sends the topic creation request again.
5. This topic creation request (with replication factor = 2 and partition = 2) had been partially executed before broker X crashes, so broker Y may have done something required by broker X. For example, broker Y has some metadata about this topic. Therefore, when Broker Y does some sanity check with the metadata, it will find this topic exists, so broker Y directly returns `TopicExistsException` as the response.
6. The client receives `TopicExistsException`, and directly believes that this topic has been created, so it is thrown back to the user with the API `result.all().get()`.

There are 2 diagrams illustrating these 6 steps:

Now the core question is whether this workflow violates the semantic & design of the Kafka Client API. We read the “Create Topics Response” section in KIP-4 (https://cwiki.apache.org/confluence/display/kafka/kip-4+-+command+line+and+centralized+administrative+operations). We found that the description in KIP-4 focuses on the batch request of topic creations and how they work independently. It does not talk about how the client should deal with the aforementioned buggy scenario.

According to “common sense”, we think the client should be able to know that the metadata existing in broker Y is actually created by the client via the crashed broker X. Also, the client should not throw `TopicExistsException` to the user.