You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2025/06/17 16:53:15 UTC
[PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
lianetm opened a new pull request, #19983:
URL: https://github.com/apache/kafka/pull/19983
Fix to ensure assigned partitions whose topics are not in the consumer explicit subscription are considered not fetchable (so that no records are returned on poll for them)
This scenario could happen in the new async consumer (using the Consumer rebalance protocol) when the subscription changes, because the consumer will keep its assignment until the coordinator sends a new one (broker drives assignments).
This does not happen in the classic consumer because the assignment logic lives on the client-side, so the consumer pro-actively updates assignment as needed.
This PR validates assignment vs subscription on fetch for subscription to explicit topic names only Regular expressions, shared subscription remain unchange (regex case still under discussion, will be handled separately if needed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2154676443
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
Review Comment:
Nit: Set.of()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #19983:
URL: https://github.com/apache/kafka/pull/19983
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2155243860
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
+ state.seek(t1p0, 1);
+
+ assertEquals(Set.of(t1p0), state.assignedPartitions());
+ assertTrue(state.isFetchable(t1p0));
+
+ // Change subscription. Assigned partitions should remain unchanged but not fetchable.
+ state.subscribe(singleton(topic), Optional.of(rebalanceListener));
Review Comment:
Just moving to the new way of doing it (we've had several PRs doing so already btw, so just using it whenever I add new code).
I find better readability with the Set.of/List.of..., and a diff is also that Set.of won't allow nulls
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2155243860
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
+ state.seek(t1p0, 1);
+
+ assertEquals(Set.of(t1p0), state.assignedPartitions());
+ assertTrue(state.isFetchable(t1p0));
+
+ // Change subscription. Assigned partitions should remain unchanged but not fetchable.
+ state.subscribe(singleton(topic), Optional.of(rebalanceListener));
Review Comment:
Yeap, just moving to the new way of doing it (we've had several PRs doing so already btw, so just using it whenever I add new code).
I find better readability with the Set.of/List.of..., and a diff is also that Set.of won't allow nulls
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:
##########
@@ -155,7 +155,10 @@ private Fetch<K, V> fetchRecords(final CompletedFetch nextInLineFetch, int maxRe
log.debug("Not returning fetched records for partition {} since it is no longer assigned", tp);
} else if (!subscriptions.isFetchable(tp)) {
// this can happen when a partition is paused before fetched records are returned to the consumer's
- // poll call or if the offset is being reset
+ // poll call or if the offset is being reset.
+ // It can also happen under the Consumer rebalance protocol, when the consumer changes its subscription.
+ // Until the consumer receives an updated assignment from the coordinator, it can hold assigned partitions
+ // that are not in the subscription anymore, so we make them not fetchable.
Review Comment:
not really at the moment (with the explicit subscription). You're right about that point but it came out when brainstorming about async assignment (ex. broker-side regex, intentionally left out of scope here).
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -487,14 +487,27 @@ public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPart
List<TopicPartition> result = new ArrayList<>();
assignment.forEach((topicPartition, topicPartitionState) -> {
// Cheap check is first to avoid evaluating the predicate if possible
- if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || topicPartitionState.isFetchable())
+ if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
&& isAvailable.test(topicPartition)) {
result.add(topicPartition);
}
});
return result;
}
+ /**
+ * Check if the partition is fetchable.
+ * If the consumer has explicitly subscribed to a list of topic names,
+ * this will also check that the partition is contained in the subscription.
+ */
+ private synchronized boolean isFetchableAndSubscribed(TopicPartition topicPartition, TopicPartitionState topicPartitionState) {
+ if (subscriptionType.equals(SubscriptionType.AUTO_TOPICS) && !subscription.contains(topicPartition.topic())) {
+ log.debug("Assigned partition {} is not in the subscription {} so will be considered not fetchable.", topicPartition, subscription);
+ return false;
+ }
Review Comment:
> Did we explore the option of removing or flagging the errant partition?
Yes, but seemed all more complext.
- "removing the partitions" would require the client pro-actively doing the full thing (triggering callbacks, updating assignments), so it would mean having the client driving assignments along with the coordinator. This is what we intentionally got rid of with the Consumer protocol, where clients let the coordinator drive assignments (so we ant to avoid bringing those complexities back into the client).
- "flagging the partitions" is kind of the goal with this approach really, but without an explicit flag (ex. not using "pause"), just because it would be much more complicated to maintain. Checking assignment vs subscriptions on fetch seems the simplest approach to dynamically "flag" the partition as non-fetchable. Makes sense?
> There is the possibility of logging this on every fetch loop for a topic which has been unsubscribed and not yet removed from the assignment
yeap, totally. Changed to trace better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2154483769
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -487,14 +487,27 @@ public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPart
List<TopicPartition> result = new ArrayList<>();
assignment.forEach((topicPartition, topicPartitionState) -> {
// Cheap check is first to avoid evaluating the predicate if possible
- if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || topicPartitionState.isFetchable())
+ if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
Review Comment:
We'll revisit this in https://issues.apache.org/jira/browse/KAFKA-19410.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -880,7 +893,7 @@ public synchronized boolean isPaused(TopicPartition tp) {
synchronized boolean isFetchable(TopicPartition tp) {
TopicPartitionState assignedOrNull = assignedStateOrNull(tp);
- return assignedOrNull != null && assignedOrNull.isFetchable();
+ return assignedOrNull != null && isFetchableAndSubscribed(tp, assignedOrNull);
Review Comment:
nit: This is a bit odd. If we call `isFetchableAndSubscribed`, we know that `assignedOrNull` is specifically not null. I would just rename that variable to `tps` or such, so there's no implication of nullness.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -1071,4 +1122,33 @@ public void testRequestOffsetResetIfPartitionAssigned() {
assertThrows(IllegalStateException.class, () -> state.isOffsetResetNeeded(unassignedPartition));
}
+
+ // This test ensures the "fetchablePartitions" does not run the custom predicate if the partition is not fetchable
+ // This func is used in the hot path for fetching, to find fetchable partitions that are not in the buffer,
+ // so it should avoid evaluating the predicate if not needed.
+ @Test
+ public void testFetchablePartitionsPerformsCheapChecksFirst() {
+ // Setup fetchable partition and pause it
+ state.assignFromUser(singleton(tp0));
Review Comment:
Another candidate for `Set.of(tp0)`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
+ state.seek(t1p0, 1);
+
+ assertEquals(Set.of(t1p0), state.assignedPartitions());
+ assertTrue(state.isFetchable(t1p0));
+
+ // Change subscription. Assigned partitions should remain unchanged but not fetchable.
+ state.subscribe(singleton(topic), Optional.of(rebalanceListener));
Review Comment:
Not that I can see. Probably ought to be `Set.of` these days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2154676443
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
Review Comment:
Nit: Set.of()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2153469078
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
+ state.seek(t1p0, 1);
+
+ assertEquals(Set.of(t1p0), state.assignedPartitions());
+ assertTrue(state.isFetchable(t1p0));
+
+ // Change subscription. Assigned partitions should remain unchanged but not fetchable.
+ state.subscribe(singleton(topic), Optional.of(rebalanceListener));
Review Comment:
Thanks for the PR, just out of curiosity, is there any difference between singleton() and Set.of() for this??
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2154806961
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription() {
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
Review Comment:
Nit: Set.of
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2155122157
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -487,14 +487,27 @@ public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPart
List<TopicPartition> result = new ArrayList<>();
assignment.forEach((topicPartition, topicPartitionState) -> {
// Cheap check is first to avoid evaluating the predicate if possible
- if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || topicPartitionState.isFetchable())
+ if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
&& isAvailable.test(topicPartition)) {
result.add(topicPartition);
}
});
return result;
}
+ /**
+ * Check if the partition is fetchable.
+ * If the consumer has explicitly subscribed to a list of topic names,
+ * this will also check that the partition is contained in the subscription.
+ */
+ private synchronized boolean isFetchableAndSubscribed(TopicPartition topicPartition, TopicPartitionState topicPartitionState) {
+ if (subscriptionType.equals(SubscriptionType.AUTO_TOPICS) && !subscription.contains(topicPartition.topic())) {
+ log.debug("Assigned partition {} is not in the subscription {} so will be considered not fetchable.", topicPartition, subscription);
+ return false;
+ }
Review Comment:
There is the possibility of logging this on every fetch loop for a topic which has been unsubscribed and not yet removed from the assignment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19356: Prevent new consumer fetch assigned partitions not in explicit subscription [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2153244230
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:
##########
@@ -155,7 +155,10 @@ private Fetch<K, V> fetchRecords(final CompletedFetch nextInLineFetch, int maxRe
log.debug("Not returning fetched records for partition {} since it is no longer assigned", tp);
} else if (!subscriptions.isFetchable(tp)) {
// this can happen when a partition is paused before fetched records are returned to the consumer's
- // poll call or if the offset is being reset
+ // poll call or if the offset is being reset.
+ // It can also happen under the Consumer rebalance protocol, when the consumer changes its subscription.
+ // Until the consumer receives an updated assignment from the coordinator, it can hold assigned partitions
+ // that are not in the subscription anymore, so we make them not fetchable.
Review Comment:
Wasn't one of the cases that was discovered where the coordinator (accidentally) assigned partitions for topics that the consumer wasn't subscribed to?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -487,14 +487,27 @@ public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPart
List<TopicPartition> result = new ArrayList<>();
assignment.forEach((topicPartition, topicPartitionState) -> {
// Cheap check is first to avoid evaluating the predicate if possible
- if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || topicPartitionState.isFetchable())
+ if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
&& isAvailable.test(topicPartition)) {
result.add(topicPartition);
}
});
return result;
}
+ /**
+ * Check if the partition is fetchable.
+ * If the consumer has explicitly subscribed to a list of topic names,
+ * this will also check that the partition is contained in the subscription.
+ */
+ private synchronized boolean isFetchableAndSubscribed(TopicPartition topicPartition, TopicPartitionState topicPartitionState) {
+ if (subscriptionType.equals(SubscriptionType.AUTO_TOPICS) && !subscription.contains(topicPartition.topic())) {
+ log.debug("Assigned partition {} is not in the subscription {} so will be considered not fetchable.", topicPartition, subscription);
+ return false;
+ }
Review Comment:
Did we explore the option of removing or flagging the errant partition? This debug line has the possibility of being output on every fetch loop, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org