Description

Hi, my name is Hao Sun from the Uber Kafka team. We plan to report an issue with the Sarama metadata refresh.

Context about the Uber instance lifecycle management:

At Uber, our Kafka broker instance will be automatically replaced when there is a host hardware issue. Afterward, this replaced instance with the same IP address will sometimes be reused in another Kafka cluster and never belong to the original Kafka cluster.

Stale broker Metadata Refresh Issue

In our current Sarama metadata refresh logic, we will use the first available broker in the seed broker list to fetch the metadata. This seed broker list never gets changed after client initialization.

1. Suppose we pass [broker1, broker2] in the seed broker list. Everything looks good after the Sarama client initialization.
2. At some point, the broker2 got replaced internally due to a hardware issue. The Sarama client still looks good because it uses broker1 to fetch metadata, although broker2 still exists in the seed broker list.
3. Later, broker1 has some issues and becomes unavailable. According to the tryRefreshMetadata method, broker1 will be removed to dead broker list, and the Sarama client will move the second broker(broker2) to fetch metadata by using anyBroker method. Because the broker2 is online but in a different cluster, the client think this broker is healthy and fetch metadata from it.

Unfortunately, the metadata from broker2 isn't from the right cluster. All cached metadata info in the client is messed up. When we use this client to produce/consume, we will encounter the topic partition error. The more dangerous case is if two Kafka cluster has topics with the same name, we are more likely to produce/consume the wrong topics from the wrong cluster.

Proposed fix:

We propose to change the behavior of how to pick up a broker to do metadata refresh. Instead of using the static seed broker list, we will pick up the least loaded broker from the cached broker list to refresh the metadata. If the cached broker list is empty, we will use the first broker in the seed broker list as we are doing now. We already have this idea patch internally and hope to create a PR. This fix idea is the same as the open-source Java Kafka client. The seed broker list is the bootstrap list only used during startup. Each metadata refresh will pick up the least loaded broker from the cached broker list. Java source code.

Let me know what you think. If it looks good, I will create a PR for review.

Thanks.
Hao

Versions

Basic client behavior changes for all Sarama/Kafka/Go versions.