Skip to content

KAFKA-9181; Maintain clean separation between local and group subscriptions in consumer's SubscriptionState #7941

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
rajinisivaram merged 6 commits into from
Jan 24, 2020
Merged

KAFKA-9181; Maintain clean separation between local and group subscriptions in consumer's SubscriptionState #7941

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
63aa4e3
KAFKA-9181; Ensure SubscriptionState.groupSubscription is updated eve…
rajinisivaram Jan 12, 2020
4384fbc
Address review comment
rajinisivaram Jan 14, 2020
797aa7c
Address review comment
rajinisivaram Jan 22, 2020
6f574ea
Don't include subscription topics in `groupSubscription`
rajinisivaram Jan 22, 2020
57765e1
Address review comment
rajinisivaram Jan 22, 2020
18a4917
Address review comment
rajinisivaram Jan 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1357,7 +1357,7 @@ private static class MetadataSnapshot {

private MetadataSnapshot(SubscriptionState subscription, Cluster cluster, int version) {
Map<String, Integer> partitionsPerTopic = new HashMap<>();
for (String topic : subscription.groupSubscription()) {
for (String topic : subscription.metadataTopics()) {
Integer numPartitions = cluster.partitionCountForTopic(topic);
if (numPartitions != null)
partitionsPerTopic.put(topic, numPartitions);
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ public synchronized MetadataRequest.Builder newMetadataRequestBuilder() {
if (subscription.hasPatternSubscription())
return MetadataRequest.Builder.allTopics();
List<String> topics = new ArrayList<>();
topics.addAll(subscription.groupSubscription());
topics.addAll(subscription.metadataTopics());
topics.addAll(transientTopics);
return new MetadataRequest.Builder(topics, allowAutoTopicCreation);
}
Expand All @@ -72,7 +72,7 @@ synchronized void clearTransientTopics() {

@Override
protected synchronized boolean retainTopic(String topic, boolean isInternal, long nowMs) {
if (transientTopics.contains(topic) || subscription.isGroupSubscribed(topic))
if (transientTopics.contains(topic) || subscription.needsMetadata(topic))
return true;

if (isInternal && !includeInternalTopics)
Expand Down
24 changes: 10 additions & 14 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,12 +183,6 @@ private boolean changeSubscription(Set<String> topicsToSubscribe) {
return false;

subscription = topicsToSubscribe;
if (subscriptionType != SubscriptionType.USER_ASSIGNED) {
groupSubscription = new HashSet<>(groupSubscription);
groupSubscription.addAll(topicsToSubscribe);
} else {
groupSubscription = new HashSet<>(topicsToSubscribe);
}
return true;
}

Expand All @@ -208,7 +202,7 @@ synchronized boolean groupSubscribe(Collection<String> topics) {
* Reset the group's subscription to only contain topics subscribed by this consumer.
*/
synchronized void resetGroupSubscription() {
groupSubscription = subscription;
groupSubscription = Collections.emptySet();
}

/**
Expand Down Expand Up @@ -335,22 +329,24 @@ public synchronized Set<TopicPartition> pausedPartitions() {
}

/**
* Get the subscription for the group. For the leader, this will include the union of the
* subscriptions of all group members. For followers, it is just that member's subscription.
* This is used when querying topic metadata to detect the metadata changes which would
* Get the subcription topics for which metadata is required . For the leader, this will include
* the union of the subscriptions of all group members. For followers, it is just that member's
* subscription. This is used when querying topic metadata to detect the metadata changes which would
* require rebalancing. The leader fetches metadata for all topics in the group so that it
* can do the partition assignment (which requires at least partition counts for all topics
* to be assigned).
*
* @return The union of all subscribed topics in the group if this member is the leader
* of the current generation; otherwise it returns the same set as {@link #subscription()}
*/
synchronized Set<String> groupSubscription() {
return this.groupSubscription;
synchronized Set<String> metadataTopics() {
Set<String> topics = new HashSet<>(groupSubscription);

@hachikuji hachikuji Jan 22, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

It might borderline overkill, but I'm considering if we could avoid the copy here with logic like the following:

if (!groupSubscription.isEmpty())
  return groupSubscription;
else
  return subscription;

Basically relying on the the group subscription being a superset of the local subscription when it is defined.

@rajinisivaram rajinisivaram Jan 22, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@hachikuji Thanks for the review, updated. I had to update one unit test since it wasn't using a superset for group subscription, but think that is ok.

topics.addAll(subscription);
return topics;
}

synchronized boolean isGroupSubscribed(String topic) {
return groupSubscription.contains(topic);
synchronized boolean needsMetadata(String topic) {
return subscription.contains(topic) || groupSubscription.contains(topic);
}

private TopicPartitionState assignedState(TopicPartition tp) {
Expand Down
13 changes: 8 additions & 5 deletions ...ts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -580,7 +580,7 @@ public void testNormalJoinGroupLeader() {

assertFalse(coordinator.rejoinNeededOrPending());
assertEquals(toSet(assigned), subscriptions.assignedPartitions());
assertEquals(subscription, subscriptions.groupSubscription());
assertEquals(subscription, subscriptions.metadataTopics());
assertEquals(0, rebalanceListener.revokedCount);
assertNull(rebalanceListener.revoked);
assertEquals(1, rebalanceListener.assignedCount);
Expand Down Expand Up @@ -639,7 +639,7 @@ public void testOutdatedCoordinatorAssignment() {

assertFalse(coordinator.rejoinNeededOrPending());
assertEquals(toSet(newAssignment), subscriptions.assignedPartitions());
assertEquals(toSet(newSubscription), subscriptions.groupSubscription());
assertEquals(toSet(newSubscription), subscriptions.metadataTopics());
assertEquals(protocol == EAGER ? 1 : 0, rebalanceListener.revokedCount);
assertEquals(1, rebalanceListener.assignedCount);
assertEquals(assigned, rebalanceListener.assigned);
Expand Down Expand Up @@ -678,7 +678,7 @@ public void testPatternJoinGroupLeader() {

assertFalse(coordinator.rejoinNeededOrPending());
assertEquals(2, subscriptions.numAssignedPartitions());
assertEquals(2, subscriptions.groupSubscription().size());
assertEquals(2, subscriptions.metadataTopics().size());
assertEquals(2, subscriptions.subscription().size());
// callback not triggered at all since there's nothing to be revoked
assertEquals(0, rebalanceListener.revokedCount);
Expand Down Expand Up @@ -909,7 +909,7 @@ public void testNormalJoinGroupFollower() {

assertFalse(coordinator.rejoinNeededOrPending());
assertEquals(toSet(assigned), subscriptions.assignedPartitions());
assertEquals(subscription, subscriptions.groupSubscription());
assertEquals(subscription, subscriptions.metadataTopics());
assertEquals(0, rebalanceListener.revokedCount);
assertNull(rebalanceListener.revoked);
assertEquals(1, rebalanceListener.assignedCount);
Expand Down Expand Up @@ -1256,6 +1256,9 @@ public void testSubscriptionChangeWithAuthorizationFailure() {
// Change subscription to include only the authorized topic. Complete rebalance and check that
// references to topic2 have been removed from SubscriptionState.
subscriptions.subscribe(Utils.mkSet(topic1), rebalanceListener);
assertEquals(Collections.singleton(topic1), subscriptions.metadataTopics());
client.prepareMetadataUpdate(TestUtils.metadataUpdateWith("kafka-cluster", 1,
Collections.emptyMap(), singletonMap(topic1, 1)));

Map<String, List<String>> memberSubscriptions = singletonMap(consumerId, singletonList(topic1));
partitionAssignor.prepare(singletonMap(consumerId, Arrays.asList(t1p)));
Expand All @@ -1264,7 +1267,7 @@ public void testSubscriptionChangeWithAuthorizationFailure() {
coordinator.poll(time.timer(Long.MAX_VALUE));

assertEquals(singleton(topic1), subscriptions.subscription());
assertEquals(singleton(topic1), subscriptions.groupSubscription());
assertEquals(singleton(topic1), subscriptions.metadataTopics());
}

@Test
Expand Down