showuon commented on a change in pull request #10509:
URL: https://github.com/apache/kafka/pull/10509#discussion_r622064430
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
##########
@@ -263,16 +279,59 @@ private boolean allSubscriptionsEqual(Set<String>
allTopics,
if (log.isDebugEnabled()) {
log.debug("final assignment: " + assignment);
}
-
+
return assignment;
}
- private SortedSet<TopicPartition> getTopicPartitions(Map<String, Integer>
partitionsPerTopic) {
- SortedSet<TopicPartition> allPartitions =
- new
TreeSet<>(Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
- for (Entry<String, Integer> entry: partitionsPerTopic.entrySet()) {
- String topic = entry.getKey();
- for (int i = 0; i < entry.getValue(); ++i) {
+ /**
+ * get the unassigned partition list by computing the difference set of
the sortedPartitions(all partitions)
+ * and sortedToBeRemovedPartitions. We use two pointers technique here:
+ *
+ * We loop the sortedPartition, and compare the ith element in sorted
toBeRemovedPartitions(i start from 0):
+ * - if not equal to the ith element, add to unassignedPartitions
+ * - if equal to the the ith element, get next element from
sortedToBeRemovedPartitions
+ *
+ * @param sortedPartitions: sorted all partitions
+ * @param sortedToBeRemovedPartitions: sorted partitions, all are included
in the sortedPartitions
+ * @return the partitions don't assign to any current consumers
+ */
+ private List<TopicPartition> getUnassignedPartitions(List<TopicPartition>
sortedPartitions,
Review comment:
> the removed assigned partitions takes almost the same time, but after
is still faster. (explained below)
I think the reason is:
1. `sort(toBeRemovedPartitions)` is `C * M * log(C * M)` , that's correct in
theory! But in our assignment case, usually, the consumer assignment is
paritition-sorted (due to the consumer leader create the assignment in order).
So, based on the java doc in `List.sort`:
> This implementation is a stable, adaptive, iterative mergesort that
requires far fewer than n lg(n) comparisons when the input array is partially
sorted, while offering the performance of a traditional mergesort when the
input array is randomly ordered. If the input array is nearly sorted, the
implementation requires approximately n comparisons.
That means, the sort should be faster than we thought.
2. The nature of `ArrayList` is cache-friendly. That is, when retrieving
data, we can get their neighbor data at the same time, it improves the data
retrieval from memory. So, this nature makes the ArrayList` iteration and add M
elements faster than TreeSet.
I don't know if there are other potential reason, but the performance is
indeed improved by `ArrayList`.
> an unknown but possibly small performance improvement that uses a lot more
memory, and a somewhat worse algorithm with only the one data structure and
slightly cleaner code.
I'd say, now, with my previous improvement, the memory usage is basically
the same. I'd choose to choose my change to make the assignor faster. After
all, faster assignor, faster rebalance.
That's my 2 cents.
--
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:
[email protected]
