Incorrect metadata refresh when the seed broker used for refresh is stale and moved to another Kafka cluster

[HaoSunUber]

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.

[dnwe]

@HaoSunUber 👋🏻 thanks for getting in touch. I'm certainly happy to review a PR that updates Sarama to prefer the least loaded existing broker connection when updating metadata. As you mention that brings us more in-line with the Java client, and is also more efficient if we re-use existing connections. Although the Java client doesn't currently fallback to the bootstrap addresses, I think it makes sense for Sarama to continue to do so

Regarding your specific issue of accidentally fetching from a difference cluster though, I think we probably want to take heed of the advice given in KIP-899 (which is also exploring this space for the Java client) and "check the cluster ID [in any metadata response]" and "the client will fail if the returned cluster ID doesn't match the previously known one" — i.e., (as long as we're Version: 0_10_1_0 or newer) we should cache the cluster ID in the first metadata responses, and reject any that don't match that for the lifetime of the client

[HaoSunUber]

Yes, we also patched the clusterID comparison logic internally when fixing this problem. If you think it is reasonable, I can do another PR for the clusterID validation change. I think the cluster ID is always static in the client lifecycle and should not have any degradation after introducing this validation

[HaoSunUber]

thanks. Code merged. Closed