Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

[consumer] Trigger a rejoin on partition racks' change [KIP-881] #4291

Merged
merged 3 commits into from
Jun 5, 2023

Conversation

milindl
Copy link
Contributor

@milindl milindl commented May 25, 2023

This change handles the final part of KIP-881: triggering a rejoin, in case we detect that one of the topics we are subscribed to, has a change in the set of racks of one of its partitions.
More precisely,
trigger a rejoin, if

  1. client.rack is set for consumer
  2. there exists a partition P of a topic T, such that we are subscribed to T and racks_in_cached_metadata(P) != racks_in_fresh_metadata(P).

As for the implementation details:

  1. rd_kafka_metadata_partition_internal_t now contain an racks and their count.
  2. rd_kafka_topic_info_t now contains a list of rd_kafka_metadata_partition_internal_t*
  3. Constructors for (2) have been modified for the changes.

For non-regex case (where we look up the topic information from the topic metadata cache), the topic metadata cache entry is internally, using rd_kafka_metadata_topic_internal_t. We just populate the racks inside the rd_kafka_metadata_partition_internal_t from the broker/rack mapping.
This operation allocates strings, since there's no linkage of the lifetime between the cache entry and the broker/rack mapping.

For the regex case (where we look up the topic information from the full cache), we use the rd_kafka_metadata_partition_internal_t inside the full cache. We just populate the racks inside the rd_kafka_metadata_partition_internal_t from the broker/rack mapping. However, since the full cache also contains the broker/rack mapping, we don't allocate extra space for the string, just point inside the broker/rack mapping.

All the allocation/deduplication/sorting costs are only paid if the client rack is set. And when the replica racks are set. Otherwise, it's avoided.

A test is also added, and some fixes/changes to the mock broker to facilitate that.

@milindl milindl requested a review from emasab May 25, 2023 11:06
@milindl
Copy link
Contributor Author

milindl commented May 25, 2023

There are some issues with 0104 - fetch from follower (mock). They seem maybe related to changes in the mock broker, but I'm not certain. I'm looking into it.

@milindl
Copy link
Contributor Author

milindl commented May 25, 2023

Yes, it seems like the group coordinator is not set correctly somehow. Setting it explicitly makes it work. But I'm not sure why. Investigating further.

src/rdkafka_metadata_cache.c Outdated Show resolved Hide resolved
src/rdkafka_metadata_cache.c Outdated Show resolved Hide resolved
src/rdkafka_mock.c Outdated Show resolved Hide resolved
src/rdlist.c Outdated Show resolved Hide resolved
src/rdkafka_mock.c Outdated Show resolved Hide resolved
src/rdkafka_mock.c Outdated Show resolved Hide resolved
once `min_events` are processed in the positive case
@milindl milindl requested a review from emasab May 30, 2023 09:29
@milindl milindl force-pushed the dev_kip_881_replica_rack_change_refresh_simple branch from 50816e3 to 986c16e Compare May 30, 2023 10:23
Copy link
Contributor

@emasab emasab left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM!

@milindl milindl merged commit f608e34 into master Jun 5, 2023
@milindl milindl deleted the dev_kip_881_replica_rack_change_refresh_simple branch June 5, 2023 09:10
# for free to join this conversation on GitHub. Already have an account? # to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants