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 4 commits
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 checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@
files="(ConsumerCoordinator|Fetcher|Sender|KafkaProducer|BufferPool|ConfigDef|RecordAccumulator|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator).java"/>

<suppress checks="JavaNCSS"
files="AbstractRequest.java|KerberosLogin.java|WorkerSinkTaskTest.java|TransactionManagerTest.java|SenderTest.java|KafkaAdminClient.java"/>
files="(AbstractRequest|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest).java"/>

<suppress checks="NPathComplexity"
files="(BufferPool|Fetcher|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|Values|PluginUtils|MiniTrogdorCluster|TasksRequest|KafkaProducer).java"/>
Expand Down
4 changes: 2 additions & 2 deletions 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 @@ -211,6 +211,7 @@ public String protocolType() {
protected JoinGroupRequestData.JoinGroupRequestProtocolCollection metadata() {
log.debug("Joining group with current subscription: {}", subscriptions.subscription());
this.joinedSubscription = subscriptions.subscription();
subscriptions.resetGroupSubscription();

@hachikuji hachikuji Jan 14, 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.

I'm confused about a couple things. First, if a subscription changes while we are in the middle of a rebalance, do we need to call onJoinPrepare again? With the old semantics, we probably didn't need to because we revoked everything, but with "cooperative" rebalancing, perhaps we should? cc @guozhangwang

The second thing is that it seems surprising that we raise auth errors from the "group subscription." If consumer 1 has subscribed to "A" and consumer 2 has subscribed to "B," I probably wouldn't expect to get any auth errors for topic "B" from consumer 1. So maybe we should be ignoring auth errors not part of subscription()? Not sure if that is straightforward or if there are any additional implications.

@rajinisivaram rajinisivaram Jan 14, 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. If we decide that we always need to do onJoinPrepare, then we won't need this fix.

It is odd to propagate authorization failures from the group subscription since that feels more like an implementation detail. But at the same time, if we don't get metadata for some topics, is there any point in adding topics from group subscription to the ConsumerMetadata? We probably need to update javadoc for poll() if we don't change the current behaviour.

@guozhangwang guozhangwang Jan 15, 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.

Regarding the first question, if a subscription changes while we are in the middle of a rebalance we would call onPartitionsRevoked immediately on those still-assigned partitions that are not part of the new subscription any more, and for the returned assignment we check if they match our join-subscription and if not we trigger another rebalance (this is pre-429). So I think we still do not need to re-trigger onJoinPrepare when subscription changes in the middle.

@hachikuji hachikuji Jan 17, 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.

@rajinisivaram I think I understand the problem a little better now and I'm coming around to the approach in this PR. It does make sense to me to reset the group subscription at the time that we set joinedSubscription in ConsumerCoordinator. Then I guess my only question is whether we still need to reset it inside onJoinPrepare as well?

Another (possibly worse) idea I had is to change the implementation of resetGroupSubscription. Perhaps instead of setting it equal to the current local subscription, we should set it to null or empty. We could try to maintain a cleaner separation between the local and group subscriptions. We would then need to update ConsumerMetadata to work with both the local and group subscription.

@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 reviews. I pushed two commits. The first one just removes reset from onJoinPrepare since we don't need it any more. I also wanted to try out your suggestion of maintaining clean separation between local and group subscriptions. The last commit does this. Do you think it is worth making that change?

JoinGroupRequestData.JoinGroupRequestProtocolCollection protocolSet = new JoinGroupRequestData.JoinGroupRequestProtocolCollection();

List<String> topics = new ArrayList<>(joinedSubscription);
Expand Down Expand Up @@ -689,7 +690,6 @@ protected void onJoinPrepare(int generation, String memberId) {
}

isLeader = false;
subscriptions.resetGroupSubscription();

@hachikuji hachikuji Jan 23, 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.

I guess we could restore this location with the change to the group subscription maintenance? I don't feel too strongly since the new location works also.

@rajinisivaram rajinisivaram Jan 23, 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. Restored this so that the only change from this PR is the clean separation of group and local subscriptions


if (exception != null) {
throw new KafkaException("User rebalance callback throws an error", exception);
Expand Down 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
47 changes: 43 additions & 4 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 @@ -1231,6 +1231,45 @@ public void testUpdateMetadataDuringRebalance() {
assertEquals(new HashSet<>(Arrays.asList(tp1, tp2)), subscriptions.assignedPartitions());
}

/**
* Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
* that don't re-invoke onJoinPrepare.
*/
@Test
public void testSubscriptionChangeWithAuthorizationFailure() {
final String consumerId = "consumer";

// Subscribe to two topics of which only one is authorized and verify that metadata failure is propagated.
subscriptions.subscribe(Utils.mkSet(topic1, topic2), rebalanceListener);
client.prepareMetadataUpdate(TestUtils.metadataUpdateWith("kafka-cluster", 1,
Collections.singletonMap(topic2, Errors.TOPIC_AUTHORIZATION_FAILED), singletonMap(topic1, 1)));
assertThrows(TopicAuthorizationException.class, () -> coordinator.poll(time.timer(Long.MAX_VALUE)));

client.respond(groupCoordinatorResponse(node, Errors.NONE));
coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

// Fail the first JoinGroup request
client.prepareResponse(joinGroupLeaderResponse(0, consumerId, Collections.emptyMap(),
Errors.GROUP_AUTHORIZATION_FAILED));
assertThrows(GroupAuthorizationException.class, () -> coordinator.poll(time.timer(Long.MAX_VALUE)));

// 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)));
client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE));
client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));
coordinator.poll(time.timer(Long.MAX_VALUE));

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

@Test
public void testWakeupFromAssignmentCallback() {
final String topic = "topic1";
Expand Down