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 2020/05/18 23:35:34 UTC
[GitHub] [kafka] cmccabe opened a new pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
cmccabe opened a new pull request #8690:
URL: https://github.com/apache/kafka/pull/8690
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#issuecomment-661469237
It seem we should add a test that verifies that the producer does the right thing when configures with the RoundRobinPartitioner? Only unit testing the `RoundRoubinPartitioner` might fall short -- and in fact, if we would have had a proper test for this case, this bug would have been discovered upfront.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r457759030
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
Might it be simpler to just do:
```
topicCounterMap.put(topic, prevPartition -1);
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jonmcewen commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
jonmcewen commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r741175055
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
I think I saw race conditions under load that caused partitions to be skipped with solutions like this. Please consider this patch: https://github.com/apache/kafka/pull/11326/files
--
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] jonmcewen commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
jonmcewen commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r741175055
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
I think I saw race conditions under load that caused partitions to be skipped with solutions like this. Please consider this patch: https://github.com/apache/kafka/pull/11326/files
--
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] mikebin commented on pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
mikebin commented on pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#issuecomment-721907104
Looks like this is still pending completion? Just wanted to check on status of getting this merged.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r426950830
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
Is this an issue for third party partitioners as well?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r428998676
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
It's possible. There's no easy fix, though, other than rethinking the `Partitioner` API. That would be an incompatible change.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jonmcewen commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner
Posted by GitBox <gi...@apache.org>.
jonmcewen commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r741175055
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
}
private int nextValue(String topic) {
- AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
- return new AtomicInteger(0);
- });
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
+ @Override
+ public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+ // After onNewBatch is called, we will call partition() again.
+ // So 'rewind' the counter for this topic.
+ AtomicInteger counter = topicCounterMap.
+ computeIfAbsent(topic, k -> new AtomicInteger(0));
+ counter.getAndDecrement();
Review comment:
I think I saw race conditions under load that caused partitions to be skipped with solutions like this. Please consider this patch: https://github.com/apache/kafka/pull/11326/files
--
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