lianetm commented on code in PR #15511:
URL: https://github.com/apache/kafka/pull/15511#discussion_r1523734711
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1497,4 +1502,93 @@ public PollResult poll(final long currentTimeMs) {
List<MemberStateListener> stateListeners() {
return unmodifiableList(stateUpdatesListeners);
}
+
+ private final static class LocalAssignmentImpl implements LocalAssignment {
+
+ private static final long NONE_EPOCH = -1;
+
+ private static final LocalAssignmentImpl NONE = new
LocalAssignmentImpl(NONE_EPOCH, Collections.emptyMap());
+
+ private final long localEpoch;
+
+ private final Map<Uuid, SortedSet<Integer>> partitions;
+
+ public LocalAssignmentImpl(long localEpoch, Map<Uuid,
SortedSet<Integer>> partitions) {
+ this.localEpoch = localEpoch;
+ this.partitions = partitions;
+ if (localEpoch == NONE_EPOCH && !partitions.isEmpty()) {
+ throw new IllegalArgumentException("Local epoch must be set if
there are partitions");
+ }
+ }
+
+ public LocalAssignmentImpl(long localEpoch,
SortedSet<TopicIdPartition> topicIdPartitions) {
+ this.localEpoch = localEpoch;
+ this.partitions = new HashMap<>();
+ if (localEpoch == NONE_EPOCH && !topicIdPartitions.isEmpty()) {
+ throw new IllegalArgumentException("Local epoch must be set if
there are partitions");
+ }
+ topicIdPartitions.forEach(topicIdPartition -> {
+ Uuid topicId = topicIdPartition.topicId();
+ partitions.computeIfAbsent(topicId, k -> new
TreeSet<>()).add(topicIdPartition.partition());
+ });
+ }
+
+ @Override
+ public String toString() {
+ return "{" +
+ "localEpoch=" + localEpoch +
+ ", partitions=" + partitions +
+ '}';
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final LocalAssignmentImpl that = (LocalAssignmentImpl) o;
+ return localEpoch == that.localEpoch && Objects.equals(partitions,
that.partitions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(localEpoch, partitions);
+ }
+
+ @Override
+ public Map<Uuid, SortedSet<Integer>> getPartitions() {
+ return partitions;
+ }
+
+ @Override
+ public boolean isNone() {
+ return localEpoch == NONE_EPOCH;
+ }
+
+ Optional<LocalAssignmentImpl>
updateWith(ConsumerGroupHeartbeatResponseData.Assignment assignment) {
+
+ // Return if we have an assignment, and it is the same as current
assignment; comparison without creating a new collection
+ if (localEpoch != NONE_EPOCH) {
+ // check if the new assignment is different from the current
target assignment
Review Comment:
nit: here we are checking if the new assignment is "the same as" the current
(not diff)....even maybe just remove this comment, as the one above seems
enough?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
