squah-confluent commented on code in PR #23736: URL: https://github.com/apache/kafka/pull/23736#discussion_r4337808479
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; + + /** + * The partitions of the topic being placed that can go to the members below their quota, which + * are those without an owner and those beyond the quota of their owner. + */ + private final IntList free = new IntList(16); + + /** + * The subscribers of the topic being placed below their quota, in member order. + */ + private final IntList receivers = new IntList(16); + + /** + * The members that got partitions of the topic being placed, to reset {@link #placed}. + */ + private final IntList placedMembers = new IntList(16); + + /** + * @param group The members and topics of the group. + * @param currentAssignment The current assignment. + * @param quotas The quotas of the members. + */ + PartitionPlacement(IndexedGroup group, CurrentAssignment currentAssignment, Quotas quotas) { + this.group = group; + this.currentAssignment = currentAssignment; + this.quotas = quotas; + this.placed = new int[group.memberCount()]; + this.hasExtraPartitionOfTopic = new boolean[group.memberCount()]; + } + + /** + * Places the partitions of the topic. + * + * @param topic The topic. + * @param assignment Receives, for every partition of the topic, the member getting it. Its + * length is at least the partition count of the topic. + * @return {@code true} if a partition may move. {@code false} guarantees that every partition + * stays with its owner, and the caller need not read the assignment. + */ + boolean place(int topic, int[] assignment) { + markExtras(topic, true); + int minQuota = quotas.minQuota(topic); + var owners = currentAssignment.owners(topic); + free.clear(); + for (int partition = 0; partition < group.partitionCount(topic); partition++) { + int owner = owners == null ? -1 : owners[partition]; + if (owner >= 0 && placed[owner] < quota(owner, minQuota)) { + assignment[partition] = owner; + addPlaced(owner); + } else { + free.add(partition); + } + } + // The quotas add up to the partition count, so when every partition stays with its owner, + // every member has its quota and nothing moves. + boolean moves = !free.isEmpty(); + if (moves) { + collectReceivers(topic, minQuota); + int next = 0; + for (int i = 0; i < receivers.size(); i++) { + int member = receivers.get(i); + for (int missing = quota(member, minQuota) - placed[member]; missing > 0; missing--) { + assignment[free.get(next++)] = member; + } + } + } + reset(topic); + return moves; + } + + /** + * Sets or clears the marks of the members getting an extra partition of the topic. + */ + private void markExtras(int topic, boolean mark) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + hasExtraPartitionOfTopic[members.get(i)] = mark; + } + } + + /** + * @return The quota of the member for the topic being placed. + */ + private int quota(int member, int minQuota) { + return hasExtraPartitionOfTopic[member] ? minQuota + 1 : minQuota; + } Review Comment: (for discussion, not necessarily looking for a change in this PR) This would fit better as a method on `Quotas` except it doesn't have the `hasExtraPartitionOfTopic` representation to answer in constant time. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; + + /** + * The partitions of the topic being placed that can go to the members below their quota, which + * are those without an owner and those beyond the quota of their owner. + */ + private final IntList free = new IntList(16); + + /** + * The subscribers of the topic being placed below their quota, in member order. + */ + private final IntList receivers = new IntList(16); + + /** + * The members that got partitions of the topic being placed, to reset {@link #placed}. + */ + private final IntList placedMembers = new IntList(16); + + /** + * @param group The members and topics of the group. + * @param currentAssignment The current assignment. + * @param quotas The quotas of the members. + */ + PartitionPlacement(IndexedGroup group, CurrentAssignment currentAssignment, Quotas quotas) { + this.group = group; + this.currentAssignment = currentAssignment; + this.quotas = quotas; + this.placed = new int[group.memberCount()]; + this.hasExtraPartitionOfTopic = new boolean[group.memberCount()]; + } + + /** + * Places the partitions of the topic. + * + * @param topic The topic. + * @param assignment Receives, for every partition of the topic, the member getting it. Its + * length is at least the partition count of the topic. + * @return {@code true} if a partition may move. {@code false} guarantees that every partition + * stays with its owner, and the caller need not read the assignment. + */ + boolean place(int topic, int[] assignment) { + markExtras(topic, true); + int minQuota = quotas.minQuota(topic); + var owners = currentAssignment.owners(topic); + free.clear(); + for (int partition = 0; partition < group.partitionCount(topic); partition++) { + int owner = owners == null ? -1 : owners[partition]; + if (owner >= 0 && placed[owner] < quota(owner, minQuota)) { + assignment[partition] = owner; + addPlaced(owner); + } else { + free.add(partition); + } + } + // The quotas add up to the partition count, so when every partition stays with its owner, + // every member has its quota and nothing moves. + boolean moves = !free.isEmpty(); + if (moves) { + collectReceivers(topic, minQuota); + int next = 0; + for (int i = 0; i < receivers.size(); i++) { + int member = receivers.get(i); + for (int missing = quota(member, minQuota) - placed[member]; missing > 0; missing--) { + assignment[free.get(next++)] = member; + } + } + } + reset(topic); + return moves; + } + + /** + * Sets or clears the marks of the members getting an extra partition of the topic. + */ + private void markExtras(int topic, boolean mark) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + hasExtraPartitionOfTopic[members.get(i)] = mark; + } + } + + /** + * @return The quota of the member for the topic being placed. + */ + private int quota(int member, int minQuota) { + return hasExtraPartitionOfTopic[member] ? minQuota + 1 : minQuota; + } + + /** + * Counts one more partition placed on the member. + */ + private void addPlaced(int member) { + if (placed[member]++ == 0) { + placedMembers.add(member); + } + } + + /** + * Collects in {@link #receivers} the subscribers of the topic below their quota, in member + * order. With a minimum quota of 0, only the members with an extra partition can be below their + * quota. + */ + private void collectReceivers(int topic, int minQuota) { + receivers.clear(); + if (minQuota == 0) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + if (placed[members.get(i)] == 0) { + receivers.add(members.get(i)); + } + } + } else { + for (int cohort : group.cohortsOf(topic)) { + for (int member : group.membersOf(cohort)) { + if (placed[member] < quota(member, minQuota)) { + receivers.add(member); + } + } + } + } + // The members of a single cohort are in member order already, unlike the members with an + // extra partition or the members of several cohorts. + if (minQuota == 0 || group.cohortsOf(topic).length > 1) { + receivers.sort(); + } + } + + /** + * Resets the numbers of partitions placed on the members, after a topic or before placing it + * again. + */ + private void clearPlaced() { + for (int i = 0; i < placedMembers.size(); i++) { + placed[placedMembers.get(i)] = 0; + } + placedMembers.clear(); + } + + /** + * Resets {@link #hasExtraPartitionOfTopic} and {@link #placed} after placing the topic. + */ + private void reset(int topic) { + markExtras(topic, false); + clearPlaced(); + } Review Comment: (for discussion, not necessarily looking for a change in this PR) Does poking at the elements we touched really beat the vectorized clear that `Arrays.fill` will do? ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; + + /** + * The partitions of the topic being placed that can go to the members below their quota, which + * are those without an owner and those beyond the quota of their owner. + */ + private final IntList free = new IntList(16); + + /** + * The subscribers of the topic being placed below their quota, in member order. + */ + private final IntList receivers = new IntList(16); + + /** + * The members that got partitions of the topic being placed, to reset {@link #placed}. + */ + private final IntList placedMembers = new IntList(16); + + /** + * @param group The members and topics of the group. + * @param currentAssignment The current assignment. + * @param quotas The quotas of the members. + */ + PartitionPlacement(IndexedGroup group, CurrentAssignment currentAssignment, Quotas quotas) { + this.group = group; + this.currentAssignment = currentAssignment; + this.quotas = quotas; + this.placed = new int[group.memberCount()]; + this.hasExtraPartitionOfTopic = new boolean[group.memberCount()]; + } + + /** + * Places the partitions of the topic. + * + * @param topic The topic. + * @param assignment Receives, for every partition of the topic, the member getting it. Its + * length is at least the partition count of the topic. + * @return {@code true} if a partition may move. {@code false} guarantees that every partition + * stays with its owner, and the caller need not read the assignment. + */ + boolean place(int topic, int[] assignment) { + markExtras(topic, true); + int minQuota = quotas.minQuota(topic); + var owners = currentAssignment.owners(topic); + free.clear(); + for (int partition = 0; partition < group.partitionCount(topic); partition++) { + int owner = owners == null ? -1 : owners[partition]; + if (owner >= 0 && placed[owner] < quota(owner, minQuota)) { + assignment[partition] = owner; + addPlaced(owner); + } else { + free.add(partition); + } + } + // The quotas add up to the partition count, so when every partition stays with its owner, + // every member has its quota and nothing moves. + boolean moves = !free.isEmpty(); + if (moves) { + collectReceivers(topic, minQuota); + int next = 0; + for (int i = 0; i < receivers.size(); i++) { + int member = receivers.get(i); + for (int missing = quota(member, minQuota) - placed[member]; missing > 0; missing--) { + assignment[free.get(next++)] = member; + } + } + } + reset(topic); + return moves; + } + + /** + * Sets or clears the marks of the members getting an extra partition of the topic. + */ + private void markExtras(int topic, boolean mark) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + hasExtraPartitionOfTopic[members.get(i)] = mark; + } + } + + /** + * @return The quota of the member for the topic being placed. + */ + private int quota(int member, int minQuota) { + return hasExtraPartitionOfTopic[member] ? minQuota + 1 : minQuota; + } + + /** + * Counts one more partition placed on the member. + */ + private void addPlaced(int member) { + if (placed[member]++ == 0) { + placedMembers.add(member); + } + } + + /** + * Collects in {@link #receivers} the subscribers of the topic below their quota, in member + * order. With a minimum quota of 0, only the members with an extra partition can be below their + * quota. + */ + private void collectReceivers(int topic, int minQuota) { + receivers.clear(); + if (minQuota == 0) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + if (placed[members.get(i)] == 0) { + receivers.add(members.get(i)); + } + } + } else { + for (int cohort : group.cohortsOf(topic)) { + for (int member : group.membersOf(cohort)) { + if (placed[member] < quota(member, minQuota)) { + receivers.add(member); + } + } + } + } + // The members of a single cohort are in member order already, unlike the members with an + // extra partition or the members of several cohorts. + if (minQuota == 0 || group.cohortsOf(topic).length > 1) { + receivers.sort(); + } Review Comment: I wonder if the determinism requirement is really worth it. It definitely incurs some overhead. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacementTest.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests of {@link PartitionPlacement}, on quotas set by hand. A topic with {@code P} partitions and + * {@code N} subscribers has a minimum quota of {@code P / N}, and each test gives its {@code P % N} + * extra partitions to chosen subscribers. + * + * <p>The member ids sort as A < B < C, so that A is member 0, B is member 1 and C is member + * 2, and the topic ids as T1 < T2, so that T1 is topic 0 and T2 is topic 1. + */ +public class PartitionPlacementTest { + private static final Uuid T1 = new Uuid(1L, 1L); + private static final Uuid T2 = new Uuid(2L, 1L); + + private static final int A = 0; + private static final int B = 1; + private static final int C = 2; + + /** + * Fills the assignment before placing, to tell whether {@code place} wrote it. + */ + private static final int UNTOUCHED = -7; Review Comment: nit: Since this is only used once and is the uninitialized elements pattern, we could inline it. nit: `-7` is an odd choice. Does `-1` work? ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacementTest.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests of {@link PartitionPlacement}, on quotas set by hand. A topic with {@code P} partitions and + * {@code N} subscribers has a minimum quota of {@code P / N}, and each test gives its {@code P % N} + * extra partitions to chosen subscribers. + * + * <p>The member ids sort as A < B < C, so that A is member 0, B is member 1 and C is member + * 2, and the topic ids as T1 < T2, so that T1 is topic 0 and T2 is topic 1. + */ +public class PartitionPlacementTest { + private static final Uuid T1 = new Uuid(1L, 1L); + private static final Uuid T2 = new Uuid(2L, 1L); + + private static final int A = 0; + private static final int B = 1; + private static final int C = 2; + + /** + * Fills the assignment before placing, to tell whether {@code place} wrote it. + */ + private static final int UNTOUCHED = -7; + + /** + * T1 has 7 partitions and 3 subscribers, so a minimum quota of 2 and 1 extra partition, A's. + * The quota of A is 3, and the quotas of B and C are 2. A keeps its 3 lowest partitions 1, 3 + * and 5, B keeps 0 and 2, and C keeps 4. Partition 6 of A goes to C, the only member below its + * quota. + */ + @Test + public void testOwnersKeepTheirLowestPartitionsUpToTheirQuota() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 1, 3, 5, 6))) + .withMember("B", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 2))) + .withMember("C", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 4))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 7).build()); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {B, A, B, A, C, A, C}, place(fixture, placement, T1)); + } + + /** + * T1 has 6 partitions and 3 subscribers, so a minimum quota of 2 and no extra partition. C owns + * partitions 1 to 4, and keeps 1 and 2. The other partitions, 0 and 5 without an owner and 3 + * and 4 beyond the quota of C, go to A and then B, the lowest ids first. + */ + @Test + public void testFreePartitionsGoToTheMembersBelowTheirQuotaInMemberOrder() { + var spec = new GroupSpecFixture() + .withMember("C", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 1, 2, 3, 4))) + .withMember("B", Set.of(T1)) + .withMember("A", Set.of(T1)) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 6).build()); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, C, C, A, B, B}, place(fixture, placement, T1)); + } + + /** + * A and C subscribe to T1, and B to T1 and T2, so that A and C form cohort 0 and B cohort 1. T1 + * has 6 partitions over its 3 subscribers, so a minimum quota of 2. Nobody owns them, and they + * go to A, B and C in member order, not cohort by cohort. T2 has a single partition and + * subscriber, B, which gets it with the same placement. + */ + @Test + public void testFreePartitionsGoInMemberOrderAcrossCohorts() { + var spec = new GroupSpecFixture() + .withMember("C", Set.of(T1)) + .withMember("B", Set.of(T1, T2)) + .withMember("A", Set.of(T1)) + .build(); + var describer = new TopicsFixture() + .withTopic(T1, 6) + .withTopic(T2, 1) + .build(); + var fixture = new QuotasFixture(spec, describer); + assertArrayEquals(new int[] {0, 1}, fixture.group.cohortsOf(fixture.topic(T1))); + + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, A, B, B, C, C}, place(fixture, placement, T1)); + assertArrayEquals(new int[] {B}, place(fixture, placement, T2)); + } + + /** + * T1 has 2 partitions and 3 subscribers, so a minimum quota of 0 and 2 extra partitions, C's + * and A's. C owns both partitions and keeps partition 0, and A gets partition 1, while B, whose + * quota is 0, gets nothing. + */ + @Test + public void testWithAMinQuotaOfZeroOnlyTheMembersWithAnExtraPartitionGetOne() { Review Comment: naming nit: Is this a free partitions test like the above two? It tests `collectReceivers` like the above two methods. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; Review Comment: (for discussion, not necessarily looking for a change in this PR) Giving this the side-eye since it's the second time we've converted `Quotas` into this representation. The first was in `AssignRemainingExtraPartitions.hasExtraPartitionOfTopic`. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; + + /** + * The partitions of the topic being placed that can go to the members below their quota, which + * are those without an owner and those beyond the quota of their owner. + */ + private final IntList free = new IntList(16); + + /** + * The subscribers of the topic being placed below their quota, in member order. + */ + private final IntList receivers = new IntList(16); + + /** + * The members that got partitions of the topic being placed, to reset {@link #placed}. + */ + private final IntList placedMembers = new IntList(16); + + /** + * @param group The members and topics of the group. + * @param currentAssignment The current assignment. + * @param quotas The quotas of the members. + */ + PartitionPlacement(IndexedGroup group, CurrentAssignment currentAssignment, Quotas quotas) { + this.group = group; + this.currentAssignment = currentAssignment; + this.quotas = quotas; + this.placed = new int[group.memberCount()]; + this.hasExtraPartitionOfTopic = new boolean[group.memberCount()]; + } + + /** + * Places the partitions of the topic. + * + * @param topic The topic. + * @param assignment Receives, for every partition of the topic, the member getting it. Its + * length is at least the partition count of the topic. + * @return {@code true} if a partition may move. {@code false} guarantees that every partition + * stays with its owner, and the caller need not read the assignment. + */ + boolean place(int topic, int[] assignment) { + markExtras(topic, true); + int minQuota = quotas.minQuota(topic); + var owners = currentAssignment.owners(topic); + free.clear(); + for (int partition = 0; partition < group.partitionCount(topic); partition++) { + int owner = owners == null ? -1 : owners[partition]; + if (owner >= 0 && placed[owner] < quota(owner, minQuota)) { + assignment[partition] = owner; + addPlaced(owner); + } else { + free.add(partition); + } + } + // The quotas add up to the partition count, so when every partition stays with its owner, + // every member has its quota and nothing moves. + boolean moves = !free.isEmpty(); + if (moves) { + collectReceivers(topic, minQuota); Review Comment: (for discussion, not necessarily looking for a change in this PR) An alternative to this would be to iterate over all members directly. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacement.java: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.coordinator.group.assignor.uniform2.util.IntList; + +/** + * Places the partitions of a topic on its subscribers, given their quotas, moving as few + * partitions as possible. + * + * <p>Every owner keeps its partitions up to its quota, the lowest partition ids first. The other + * partitions, those without owner and those beyond the quota of their owner, go to the subscribers + * below their quota, in member order, the lowest partition ids first. A partition thus moves only + * when its owner's quota is below what it owns. + */ +final class PartitionPlacement { + /** + * The members and topics of the group. + */ + private final IndexedGroup group; + + /** + * The current assignment, which tells the owners of every partition. + */ + private final CurrentAssignment currentAssignment; + + /** + * The quotas of the members. + */ + private final Quotas quotas; + + /** + * Per member, the number of partitions of the topic being placed that it got so far. Reset + * after every topic. + */ + private final int[] placed; + + /** + * Per member, whether it gets an extra partition of the topic being placed. Reset after every + * topic. + */ + private final boolean[] hasExtraPartitionOfTopic; + + /** + * The partitions of the topic being placed that can go to the members below their quota, which + * are those without an owner and those beyond the quota of their owner. + */ + private final IntList free = new IntList(16); + + /** + * The subscribers of the topic being placed below their quota, in member order. + */ + private final IntList receivers = new IntList(16); + + /** + * The members that got partitions of the topic being placed, to reset {@link #placed}. + */ + private final IntList placedMembers = new IntList(16); + + /** + * @param group The members and topics of the group. + * @param currentAssignment The current assignment. + * @param quotas The quotas of the members. + */ + PartitionPlacement(IndexedGroup group, CurrentAssignment currentAssignment, Quotas quotas) { + this.group = group; + this.currentAssignment = currentAssignment; + this.quotas = quotas; + this.placed = new int[group.memberCount()]; + this.hasExtraPartitionOfTopic = new boolean[group.memberCount()]; + } + + /** + * Places the partitions of the topic. + * + * @param topic The topic. + * @param assignment Receives, for every partition of the topic, the member getting it. Its + * length is at least the partition count of the topic. + * @return {@code true} if a partition may move. {@code false} guarantees that every partition + * stays with its owner, and the caller need not read the assignment. + */ + boolean place(int topic, int[] assignment) { + markExtras(topic, true); + int minQuota = quotas.minQuota(topic); + var owners = currentAssignment.owners(topic); + free.clear(); + for (int partition = 0; partition < group.partitionCount(topic); partition++) { + int owner = owners == null ? -1 : owners[partition]; + if (owner >= 0 && placed[owner] < quota(owner, minQuota)) { + assignment[partition] = owner; + addPlaced(owner); + } else { + free.add(partition); + } + } + // The quotas add up to the partition count, so when every partition stays with its owner, + // every member has its quota and nothing moves. + boolean moves = !free.isEmpty(); + if (moves) { + collectReceivers(topic, minQuota); + int next = 0; + for (int i = 0; i < receivers.size(); i++) { + int member = receivers.get(i); + for (int missing = quota(member, minQuota) - placed[member]; missing > 0; missing--) { + assignment[free.get(next++)] = member; + } + } + } + reset(topic); + return moves; + } + + /** + * Sets or clears the marks of the members getting an extra partition of the topic. + */ + private void markExtras(int topic, boolean mark) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + hasExtraPartitionOfTopic[members.get(i)] = mark; + } + } + + /** + * @return The quota of the member for the topic being placed. + */ + private int quota(int member, int minQuota) { + return hasExtraPartitionOfTopic[member] ? minQuota + 1 : minQuota; + } + + /** + * Counts one more partition placed on the member. + */ + private void addPlaced(int member) { + if (placed[member]++ == 0) { + placedMembers.add(member); + } + } + + /** + * Collects in {@link #receivers} the subscribers of the topic below their quota, in member + * order. With a minimum quota of 0, only the members with an extra partition can be below their + * quota. + */ + private void collectReceivers(int topic, int minQuota) { + receivers.clear(); + if (minQuota == 0) { + var members = quotas.membersWithExtraPartition(topic); + for (int i = 0; i < members.size(); i++) { + if (placed[members.get(i)] == 0) { + receivers.add(members.get(i)); + } + } Review Comment: This happens when a topic has more subscribers than partitions. I wonder if this is common enough to warrant its own branch. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacementTest.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests of {@link PartitionPlacement}, on quotas set by hand. A topic with {@code P} partitions and + * {@code N} subscribers has a minimum quota of {@code P / N}, and each test gives its {@code P % N} + * extra partitions to chosen subscribers. + * + * <p>The member ids sort as A < B < C, so that A is member 0, B is member 1 and C is member + * 2, and the topic ids as T1 < T2, so that T1 is topic 0 and T2 is topic 1. + */ +public class PartitionPlacementTest { + private static final Uuid T1 = new Uuid(1L, 1L); + private static final Uuid T2 = new Uuid(2L, 1L); + + private static final int A = 0; + private static final int B = 1; + private static final int C = 2; + + /** + * Fills the assignment before placing, to tell whether {@code place} wrote it. Review Comment: nit: Does `{@link` work here? ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacementTest.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests of {@link PartitionPlacement}, on quotas set by hand. A topic with {@code P} partitions and + * {@code N} subscribers has a minimum quota of {@code P / N}, and each test gives its {@code P % N} + * extra partitions to chosen subscribers. + * + * <p>The member ids sort as A < B < C, so that A is member 0, B is member 1 and C is member + * 2, and the topic ids as T1 < T2, so that T1 is topic 0 and T2 is topic 1. + */ +public class PartitionPlacementTest { + private static final Uuid T1 = new Uuid(1L, 1L); + private static final Uuid T2 = new Uuid(2L, 1L); + + private static final int A = 0; + private static final int B = 1; + private static final int C = 2; + + /** + * Fills the assignment before placing, to tell whether {@code place} wrote it. + */ + private static final int UNTOUCHED = -7; + + /** + * T1 has 7 partitions and 3 subscribers, so a minimum quota of 2 and 1 extra partition, A's. + * The quota of A is 3, and the quotas of B and C are 2. A keeps its 3 lowest partitions 1, 3 + * and 5, B keeps 0 and 2, and C keeps 4. Partition 6 of A goes to C, the only member below its + * quota. + */ + @Test + public void testOwnersKeepTheirLowestPartitionsUpToTheirQuota() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 1, 3, 5, 6))) + .withMember("B", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 2))) + .withMember("C", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 4))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 7).build()); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {B, A, B, A, C, A, C}, place(fixture, placement, T1)); + } + + /** + * T1 has 6 partitions and 3 subscribers, so a minimum quota of 2 and no extra partition. C owns + * partitions 1 to 4, and keeps 1 and 2. The other partitions, 0 and 5 without an owner and 3 + * and 4 beyond the quota of C, go to A and then B, the lowest ids first. + */ + @Test + public void testFreePartitionsGoToTheMembersBelowTheirQuotaInMemberOrder() { + var spec = new GroupSpecFixture() + .withMember("C", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 1, 2, 3, 4))) + .withMember("B", Set.of(T1)) + .withMember("A", Set.of(T1)) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 6).build()); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, C, C, A, B, B}, place(fixture, placement, T1)); + } + + /** + * A and C subscribe to T1, and B to T1 and T2, so that A and C form cohort 0 and B cohort 1. T1 + * has 6 partitions over its 3 subscribers, so a minimum quota of 2. Nobody owns them, and they + * go to A, B and C in member order, not cohort by cohort. T2 has a single partition and + * subscriber, B, which gets it with the same placement. + */ + @Test + public void testFreePartitionsGoInMemberOrderAcrossCohorts() { + var spec = new GroupSpecFixture() + .withMember("C", Set.of(T1)) + .withMember("B", Set.of(T1, T2)) + .withMember("A", Set.of(T1)) + .build(); + var describer = new TopicsFixture() + .withTopic(T1, 6) + .withTopic(T2, 1) + .build(); + var fixture = new QuotasFixture(spec, describer); + assertArrayEquals(new int[] {0, 1}, fixture.group.cohortsOf(fixture.topic(T1))); + + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, A, B, B, C, C}, place(fixture, placement, T1)); + assertArrayEquals(new int[] {B}, place(fixture, placement, T2)); + } + + /** + * T1 has 2 partitions and 3 subscribers, so a minimum quota of 0 and 2 extra partitions, C's + * and A's. C owns both partitions and keeps partition 0, and A gets partition 1, while B, whose + * quota is 0, gets nothing. + */ + @Test + public void testWithAMinQuotaOfZeroOnlyTheMembersWithAnExtraPartitionGetOne() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1)) + .withMember("B", Set.of(T1)) + .withMember("C", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 1))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 2).build()); + fixture.addExtras("C", T1); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {C, A}, place(fixture, placement, T1)); + } + + /** + * T1 has 5 partitions and 2 subscribers, so a minimum quota of 2 and 1 extra partition, A's. + * The quotas are 3 for A and 2 for B, exactly what they own, and nothing moves. + */ + @Test + public void testNothingMovesWhenEveryOwnerOwnsItsQuota() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 2, 4))) + .withMember("B", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 1, 3))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 5).build()); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertFalse(placement.place(fixture.topic(T1), new int[5])); + } + + /** + * Same quotas as in the previous test, 3 for A and 2 for B, but partition 4 has no owner, and B + * gets it. + */ + @Test + public void testPartitionsMoveWhenOneHasNoOwner() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 1, 2))) + .withMember("B", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 3))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 5).build()); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, A, A, B, B}, place(fixture, placement, T1)); + } + + /** + * Same quotas as in the previous test, 3 for A and 2 for B, and every partition has an owner, + * but A owns 4 partitions, and partition 3, beyond its quota, goes to B. + */ + @Test + public void testPartitionsMoveWhenAnOwnerOwnsMoreThanItsQuota() { + var spec = new GroupSpecFixture() + .withMember("A", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 0, 1, 2, 3))) + .withMember("B", Set.of(T1), mkAssignment(mkTopicAssignment(T1, 4))) + .build(); + var fixture = new QuotasFixture(spec, new TopicsFixture().withTopic(T1, 5).build()); + fixture.addExtras("A", T1); + var placement = new PartitionPlacement(fixture.group, fixture.currentAssignment, fixture.quotas); + + assertArrayEquals(new int[] {A, A, A, B, B}, place(fixture, placement, T1)); + } Review Comment: What's new in these tests that aren't covered by previous tests? `testOwnersKeepTheirLowestPartitionsUpToTheirQuota` covers losing partitions over the quota and `testFreePartitionsGoToTheMembersBelowTheirQuotaInMemberOrder` covers assignment of unowned partitions. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/uniform2/PartitionPlacementTest.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.assignor.uniform2; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests of {@link PartitionPlacement}, on quotas set by hand. A topic with {@code P} partitions and + * {@code N} subscribers has a minimum quota of {@code P / N}, and each test gives its {@code P % N} + * extra partitions to chosen subscribers. + * + * <p>The member ids sort as A < B < C, so that A is member 0, B is member 1 and C is member + * 2, and the topic ids as T1 < T2, so that T1 is topic 0 and T2 is topic 1. Review Comment: nit: The tests are pretty easy to understand without this, thanks to the A, B, C constants. We could probably remove 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
