You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/09/20 21:52:21 UTC
[GitHub] [kafka] wcarlson5 opened a new pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
wcarlson5 opened a new pull request #11347:
URL: https://github.com/apache/kafka/pull/11347
If for some reason partition revoked did not get called due to some bug it would be good to log a warning that the previous assignment was not clean. This could affect eos and it might be good to be able to revoke a partition instead of using partitions lost
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r714987990
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
Review comment:
Sure that works
--
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
[GitHub] [kafka] mjsax merged pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #11347:
URL: https://github.com/apache/kafka/pull/11347
--
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
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r714990115
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
+ log.warn("The previous assignment contains a partition more than once. " +
+ "This might result in violation of EOS if enabled. \n" +
Review comment:
that is probably easier to read. I will do that
--
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
[GitHub] [kafka] mjsax commented on pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#issuecomment-927172666
Merged to `trunk` and cherry-picked to `3.0` branch.
--
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
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r712541150
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
Review comment:
@mjsax This should warn about bad assignments. I was not sure what the best msg was so I game as much details as I could. If you have any better ideas I am happy to change it.
--
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
[GitHub] [kafka] mjsax commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r713232644
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
+ log.warn("The previous assignment contains a partition more than once. " +
+ "This might result in violation of EOS if enabled. \n" +
Review comment:
Not sure if we need to call out EOS in particular? It's incorrect in any case.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
+ log.warn("The previous assignment contains a partition more than once. " +
+ "This might result in violation of EOS if enabled. \n" +
Review comment:
Wondering if we should print the full old-assignment a single time if we detect any error instead? Would be helpful to get the `client.id -> old-assignment` mapping?
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r713725768
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
Review comment:
That is, in this way, it makes the check lightweight, and if we want to find out which partition cause the issue, we can "lazily" iterate them after the size check failed.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r713720965
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
Review comment:
I think this check each time is quite expensive if the partition size is large. So, I think we can do "lazy check" for this case. Because `allOwnedPartitions` is a `Set`, we can check if any duplicated partitions existed via the size sum. That is:
```java
// get the partition size we're going to add
int consumerOwnedSize = subscription.ownedPartitions().size();
int prevSize = allOwnedPartitions.size();
allOwnedPartitions.addAll(subscription.ownedPartitions());
if (allOwnedPartitions.size() < prevSize + consumerOwnedSize) {
// duplicated partitions in 2 consumers found
// log warning here
// if we want to find out which partition cause the problem, we can also iterate them here.
}
```
WDYT? Thanks.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11347: KAFKA-13296: warn if previous assignment has duplicate partitions
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11347:
URL: https://github.com/apache/kafka/pull/11347#discussion_r713720965
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -351,6 +351,17 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
// add the consumer and any info in its subscription to the client
clientMetadata.addConsumer(consumerId, subscription.ownedPartitions());
+ if (allOwnedPartitions.stream().anyMatch(t -> subscription.ownedPartitions().contains(t))) {
Review comment:
I think this check each time is quite expensive if the partition size is large. So, I think we can do "lightweight check" for this case. Because `allOwnedPartitions` is a `Set`, we can check if any duplicated partitions existed via the size sum. That is:
```java
// get the partition size we're going to add
int consumerOwnedSize = subscription.ownedPartitions().size();
int prevSize = allOwnedPartitions.size();
allOwnedPartitions.addAll(subscription.ownedPartitions());
if (allOwnedPartitions.size() < prevSize + consumerOwnedSize) {
// duplicated partitions in 2 consumers found
// log warning here
// if we want to find out which partition cause the problem, we can also iterate them here.
}
```
WDYT? Thanks.
--
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