[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188283174 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,574 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap()) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testOneConsumerNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap()) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new HashMap<>(); +// Initial Subscriptions are: A -> T1, T3 | B -> T1, T3 + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap()) +); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap()) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +mkAssignment(expectedAssignment, topic1Uuid, Arrays.asList(0, 1)); +mkAssignment(expectedAssignment, topic1Uuid, Collections.singleton(2)); +// Topic 3 Partitions Assignment +mkAssignment(expectedAssignment, topic3Uuid, Collections.singleton(0)); +mkAssignment(expectedAssignment, topic3Uuid, Collections.singleton(1)); + +assertAssignment(expectedAssignment, computedAssignment); +} + +@Test +public void testFirstAssignmentThreeConsumersThreeTopicsDifferentSubscriptions() { +Map topics = new HashMap<>();
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188284004 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,258 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * The Server Side Sticky Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +// Used in the potentiallyUnfilledMembers map and the UnfilledMembers map. +private static class MemberWithRemainingAssignments { +private final String memberId; +/** + * Number of partitions required to meet the assignment quota + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} + +/** + * @return memberId + */ +public String memberId() { +return memberId; +} + /** + * @return Remaining number of partitions + */ +public Integer remaining() { +return remaining; +} +} + +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. Review Comment: plain english for everything? -- 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...@
[GitHub] [kafka] showuon opened a new pull request, #13691: MINOR: remove kraft readme link
showuon opened a new pull request, #13691: URL: https://github.com/apache/kafka/pull/13691 The `config/kraft/README.md` is already removed. We should also remove the link. ### 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188323433 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,258 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * The Server Side Sticky Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +// Used in the potentiallyUnfilledMembers map and the UnfilledMembers map. +private static class MemberWithRemainingAssignments { +private final String memberId; +/** + * Number of partitions required to meet the assignment quota + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} + +/** + * @return memberId + */ +public String memberId() { +return memberId; +} + /** + * @return Remaining number of partitions + */ +public Integer remaining() { +return remaining; +} +} + +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if
[GitHub] [kafka] dengziming closed pull request #12318: MINOR: Support --release in FeatureCommand
dengziming closed pull request #12318: MINOR: Support --release in FeatureCommand URL: https://github.com/apache/kafka/pull/12318 -- 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
[GitHub] [kafka] dengziming commented on pull request #12318: MINOR: Support --release in FeatureCommand
dengziming commented on PR #12318: URL: https://github.com/apache/kafka/pull/12318#issuecomment-1539714488 Closing this since we are using --metadata instead of --release. -- 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
[GitHub] [kafka] dengziming commented on pull request #13691: MINOR: remove kraft readme link
dengziming commented on PR #13691: URL: https://github.com/apache/kafka/pull/13691#issuecomment-1539736425 We are still mentioning it in `upgrade.html` that says "See the config/kraft/README.md file for details." , I think we should also change it to `See the README.md#Running a Kafka broker in KRaft mode for details.` ![image](https://user-images.githubusercontent.com/26023240/237047974-b6d1b94d-d678-43c3-9552-b63b51e85a7c.png) -- 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
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188378752 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + * Fi
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188385263 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188386919 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188387610 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188387610 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[GitHub] [kafka] showuon commented on a diff in pull request #13516: KAFKA-14752: Kafka examples improvements - processor changes
showuon commented on code in PR #13516: URL: https://github.com/apache/kafka/pull/13516#discussion_r1188380635 ## examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java: ## @@ -79,111 +86,114 @@ public ExactlyOnceMessageProcessor(final String inputTopic, @Override public void run() { -// Init transactions call should always happen first in order to clear zombie transactions from previous generation. -producer.initTransactions(); - -final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); - -consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { -@Override -public void onPartitionsRevoked(Collection partitions) { -printWithTxnId("Revoked partition assignment to kick-off rebalancing: " + partitions); -} - -@Override -public void onPartitionsAssigned(Collection partitions) { -printWithTxnId("Received partition assignment after rebalancing: " + partitions); -messageRemaining.set(messagesRemaining(consumer)); -} -}); - -int messageProcessed = 0; -while (messageRemaining.get() > 0) { -try { -ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); -if (records.count() > 0) { -// Begin a new transaction session. -producer.beginTransaction(); -for (ConsumerRecord record : records) { -// Process the record and send to downstream. -ProducerRecord customizedRecord = transform(record); -producer.send(customizedRecord); +int processedRecords = 0; +long remainingRecords = Long.MAX_VALUE; +try { +// it is recommended to have a relatively short txn timeout in order to clear pending offsets faster +int transactionTimeoutMs = 10_000; +KafkaProducer producer = +new Producer(outputTopic, true, transactionalId, true, -1, transactionTimeoutMs, null).get(); + +// consumer must be in read_committed mode, which means it won't be able to read uncommitted data +boolean readCommitted = true; +KafkaConsumer consumer = new Consumer( +"processor-consumer", bootstrapServers, inputTopic, "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null) +.createKafkaConsumer(); + +// called first and once to fence zombies and abort any pending transaction +producer.initTransactions(); + +consumer.subscribe(singleton(inputTopic), this); + +Utils.printOut("Processing new records"); +while (!closed && remainingRecords > 0) { +try { +ConsumerRecords records = consumer.poll(ofMillis(200)); +if (!records.isEmpty()) { +// begin a new transaction session +producer.beginTransaction(); + +for (ConsumerRecord record : records) { +// process the record and send downstream +ProducerRecord newRecord = +new ProducerRecord<>(outputTopic, record.key(), record.value() + "-ok"); +producer.send(newRecord); +} + +// checkpoint the progress by sending offsets to group coordinator broker +// note that this API is only available for broker >= 2.5 + producer.sendOffsetsToTransaction(getOffsetsToCommit(consumer), consumer.groupMetadata()); + +// commit the transaction including offsets +producer.commitTransaction(); +processedRecords += records.count(); } - -Map offsets = consumerOffsets(); - -// Checkpoint the progress by sending offsets to group coordinator broker. -// Note that this API is only available for broker >= 2.5. -producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()); - -// Finish the transaction. All sent records should be visible for consumption now. -producer.commitTransaction(); -messageProcessed += records.count(); +} catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException + | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) { +Utils.printErr(e.getMessage()); +// we can't recover from these exceptions +shutdown(); +
[GitHub] [kafka] showuon commented on pull request #13517: KAFKA-14752: Kafka examples improvements - demo changes
showuon commented on PR #13517: URL: https://github.com/apache/kafka/pull/13517#issuecomment-1539806138 Please resolve the conflict. 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] urbandan commented on pull request #13690: KAFKA-14978 ExactlyOnceWorkerSourceTask should remove parent metrics
urbandan commented on PR #13690: URL: https://github.com/apache/kafka/pull/13690#issuecomment-1539809972 Thanks @viktorsomogyi ! -- 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
[GitHub] [kafka] showuon commented on a diff in pull request #13679: KAFKA-14291: KRaft controller should return right finalized features in ApiVersionResponse
showuon commented on code in PR #13679: URL: https://github.com/apache/kafka/pull/13679#discussion_r1188407891 ## core/src/main/scala/kafka/server/ApiVersionManager.scala: ## @@ -80,7 +81,18 @@ class SimpleApiVersionManager( private val apiVersions = ApiVersionsResponse.collectApis(enabledApis.asJava, enableUnstableLastVersion) override def apiVersionResponse(requestThrottleMs: Int): ApiVersionsResponse = { -ApiVersionsResponse.createApiVersionsResponse(requestThrottleMs, apiVersions, brokerFeatures, zkMigrationEnabled) +throw new UnsupportedOperationException("This method is not supported in SimpleApiVersionManager, use apiVersionResponse(throttleTimeMs, finalizedFeatures, epoch) instead") Review Comment: Actually it's not clear when to use `SimpleApiVersionManager` and when to use `DefaultApiVersionManager`. Could we add some comments above each class? -- 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
[GitHub] [kafka] showuon commented on pull request #13679: KAFKA-14291: KRaft controller should return right finalized features in ApiVersionResponse
showuon commented on PR #13679: URL: https://github.com/apache/kafka/pull/13679#issuecomment-1539844838 Also, checkstyle failed. Please fix them. 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #13691: MINOR: remove kraft readme link
showuon commented on PR #13691: URL: https://github.com/apache/kafka/pull/13691#issuecomment-1539857562 @dengziming , good point! I've updated it to `the KRaft section for details.` to link it to the KRaft section in the doc. 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] Hangleton commented on a diff in pull request #13535: KAFKA-9579 Fetch implementation for records in the remote storage through a specific purgatory.
Hangleton commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1188387671 ## core/src/main/java/kafka/log/remote/RemoteLogManager.java: ## @@ -600,25 +623,210 @@ public String toString() { } } -long findHighestRemoteOffset(TopicIdPartition topicIdPartition) throws RemoteStorageException { -Optional offset = Optional.empty(); -Optional maybeLog = fetchLog.apply(topicIdPartition.topicPartition()); -if (maybeLog.isPresent()) { -UnifiedLog log = maybeLog.get(); -Option maybeLeaderEpochFileCache = log.leaderEpochCache(); -if (maybeLeaderEpochFileCache.isDefined()) { -LeaderEpochFileCache cache = maybeLeaderEpochFileCache.get(); -OptionalInt epoch = cache.latestEpoch(); -while (!offset.isPresent() && epoch.isPresent()) { -offset = remoteLogMetadataManager.highestOffsetForEpoch(topicIdPartition, epoch.getAsInt()); -epoch = cache.previousEpoch(epoch.getAsInt()); +public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws RemoteStorageException, IOException { +int fetchMaxBytes = remoteStorageFetchInfo.fetchMaxBytes; +TopicPartition tp = remoteStorageFetchInfo.topicPartition; +FetchRequest.PartitionData fetchInfo = remoteStorageFetchInfo.fetchInfo; + +boolean includeAbortedTxns = remoteStorageFetchInfo.fetchIsolation == FetchIsolation.TXN_COMMITTED; + +long offset = fetchInfo.fetchOffset; +int maxBytes = Math.min(fetchMaxBytes, fetchInfo.maxBytes); + +Optional logOptional = fetchLog.apply(tp); +OptionalInt epoch = OptionalInt.empty(); + +if (logOptional.isPresent()) { +Option leaderEpochCache = logOptional.get().leaderEpochCache(); +if (leaderEpochCache.isDefined()) { +epoch = leaderEpochCache.get().epochForOffset(offset); +} +} + +Optional rlsMetadataOptional = epoch.isPresent() +? fetchRemoteLogSegmentMetadata(tp, epoch.getAsInt(), offset) +: Optional.empty(); + +if (!rlsMetadataOptional.isPresent()) { +String epochStr = (epoch.isPresent()) ? Integer.toString(epoch.getAsInt()) : "NOT AVAILABLE"; +throw new OffsetOutOfRangeException("Received request for offset " + offset + " for leader epoch " ++ epochStr + " and partition " + tp + " which does not exist in remote tier."); +} + +RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlsMetadataOptional.get(); +int startPos = lookupPositionForOffset(remoteLogSegmentMetadata, offset); +InputStream remoteSegInputStream = null; +try { +// Search forward for the position of the last offset that is greater than or equal to the target offset +remoteSegInputStream = remoteLogStorageManager.fetchLogSegment(remoteLogSegmentMetadata, startPos); +RemoteLogInputStream remoteLogInputStream = new RemoteLogInputStream(remoteSegInputStream); + +RecordBatch firstBatch = findFirstBatch(remoteLogInputStream, offset); + +if (firstBatch == null) +return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, +includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty()); + +int firstBatchSize = firstBatch.sizeInBytes(); +// An empty record is sent instead of an incomplete batch when +// - there is no minimum-one-message constraint and +// - the first batch size is more than maximum bytes that can be sent. +// - for FetchRequest version 3 or above and +if (!remoteStorageFetchInfo.minOneMessage && +!remoteStorageFetchInfo.hardMaxBytesLimit && +firstBatchSize > maxBytes) { +return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY); +} + +int updatedFetchSize = +remoteStorageFetchInfo.minOneMessage && firstBatchSize > maxBytes ? firstBatchSize : maxBytes; + +ByteBuffer buffer = ByteBuffer.allocate(updatedFetchSize); +int remainingBytes = updatedFetchSize; + +firstBatch.writeTo(buffer); +remainingBytes -= firstBatchSize; + +if (remainingBytes > 0) { +// read the input stream until min of (EOF stream or buffer's remaining capacity). +Utils.readFully(remoteSegInputStream, buffer); +} +buffer.flip(); + +FetchDataInfo fetchDataInfo = new FetchDataInfo( +new LogOffsetMetadata(offset, remoteLogSegmentMetadata.startOffset(), startPos), +MemoryRecords.readableRecords(buffer)); +
[GitHub] [kafka] Hangleton commented on a diff in pull request #13535: KAFKA-9579 Fetch implementation for records in the remote storage through a specific purgatory.
Hangleton commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1188387671 ## core/src/main/java/kafka/log/remote/RemoteLogManager.java: ## @@ -600,25 +623,210 @@ public String toString() { } } -long findHighestRemoteOffset(TopicIdPartition topicIdPartition) throws RemoteStorageException { -Optional offset = Optional.empty(); -Optional maybeLog = fetchLog.apply(topicIdPartition.topicPartition()); -if (maybeLog.isPresent()) { -UnifiedLog log = maybeLog.get(); -Option maybeLeaderEpochFileCache = log.leaderEpochCache(); -if (maybeLeaderEpochFileCache.isDefined()) { -LeaderEpochFileCache cache = maybeLeaderEpochFileCache.get(); -OptionalInt epoch = cache.latestEpoch(); -while (!offset.isPresent() && epoch.isPresent()) { -offset = remoteLogMetadataManager.highestOffsetForEpoch(topicIdPartition, epoch.getAsInt()); -epoch = cache.previousEpoch(epoch.getAsInt()); +public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws RemoteStorageException, IOException { +int fetchMaxBytes = remoteStorageFetchInfo.fetchMaxBytes; +TopicPartition tp = remoteStorageFetchInfo.topicPartition; +FetchRequest.PartitionData fetchInfo = remoteStorageFetchInfo.fetchInfo; + +boolean includeAbortedTxns = remoteStorageFetchInfo.fetchIsolation == FetchIsolation.TXN_COMMITTED; + +long offset = fetchInfo.fetchOffset; +int maxBytes = Math.min(fetchMaxBytes, fetchInfo.maxBytes); + +Optional logOptional = fetchLog.apply(tp); +OptionalInt epoch = OptionalInt.empty(); + +if (logOptional.isPresent()) { +Option leaderEpochCache = logOptional.get().leaderEpochCache(); +if (leaderEpochCache.isDefined()) { +epoch = leaderEpochCache.get().epochForOffset(offset); +} +} + +Optional rlsMetadataOptional = epoch.isPresent() +? fetchRemoteLogSegmentMetadata(tp, epoch.getAsInt(), offset) +: Optional.empty(); + +if (!rlsMetadataOptional.isPresent()) { +String epochStr = (epoch.isPresent()) ? Integer.toString(epoch.getAsInt()) : "NOT AVAILABLE"; +throw new OffsetOutOfRangeException("Received request for offset " + offset + " for leader epoch " ++ epochStr + " and partition " + tp + " which does not exist in remote tier."); +} + +RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlsMetadataOptional.get(); +int startPos = lookupPositionForOffset(remoteLogSegmentMetadata, offset); +InputStream remoteSegInputStream = null; +try { +// Search forward for the position of the last offset that is greater than or equal to the target offset +remoteSegInputStream = remoteLogStorageManager.fetchLogSegment(remoteLogSegmentMetadata, startPos); +RemoteLogInputStream remoteLogInputStream = new RemoteLogInputStream(remoteSegInputStream); + +RecordBatch firstBatch = findFirstBatch(remoteLogInputStream, offset); + +if (firstBatch == null) +return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false, +includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty()); + +int firstBatchSize = firstBatch.sizeInBytes(); +// An empty record is sent instead of an incomplete batch when +// - there is no minimum-one-message constraint and +// - the first batch size is more than maximum bytes that can be sent. +// - for FetchRequest version 3 or above and +if (!remoteStorageFetchInfo.minOneMessage && +!remoteStorageFetchInfo.hardMaxBytesLimit && +firstBatchSize > maxBytes) { +return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY); +} + +int updatedFetchSize = +remoteStorageFetchInfo.minOneMessage && firstBatchSize > maxBytes ? firstBatchSize : maxBytes; + +ByteBuffer buffer = ByteBuffer.allocate(updatedFetchSize); +int remainingBytes = updatedFetchSize; + +firstBatch.writeTo(buffer); +remainingBytes -= firstBatchSize; + +if (remainingBytes > 0) { +// read the input stream until min of (EOF stream or buffer's remaining capacity). +Utils.readFully(remoteSegInputStream, buffer); +} +buffer.flip(); + +FetchDataInfo fetchDataInfo = new FetchDataInfo( +new LogOffsetMetadata(offset, remoteLogSegmentMetadata.startOffset(), startPos), +MemoryRecords.readableRecords(buffer)); +
[GitHub] [kafka] lanshiqin commented on a diff in pull request #13584: MINOR: Add log segment unit tests, If the maximum offset beyond index, appen…
lanshiqin commented on code in PR #13584: URL: https://github.com/apache/kafka/pull/13584#discussion_r1188451651 ## core/src/test/scala/unit/kafka/log/LogSegmentTest.scala: ## @@ -65,6 +68,29 @@ class LogSegmentTest { Utils.delete(logDir) } + /** + * If the maximum offset beyond index, appended to the log section, it throws LogSegmentOffsetOverflowException Review Comment: Good idea. I've modified 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] satishd commented on a diff in pull request #13535: KAFKA-9579 Fetch implementation for records in the remote storage through a specific purgatory.
satishd commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1188470147 ## core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java: ## @@ -0,0 +1,83 @@ +/* + * 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 kafka.log.remote; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.OffsetOutOfRangeException; +import org.apache.kafka.common.record.Records; +import org.apache.kafka.server.log.remote.storage.RemoteStorageException; +import org.apache.kafka.storage.internals.log.FetchDataInfo; +import org.apache.kafka.storage.internals.log.LogOffsetMetadata; +import org.apache.kafka.storage.internals.log.RemoteLogReadResult; +import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +import java.util.function.Consumer; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class RemoteLogReaderTest { +RemoteLogManager mockRLM = mock(RemoteLogManager.class); +LogOffsetMetadata logOffsetMetadata = new LogOffsetMetadata(100); +Records records = mock(Records.class); + + +@Test +public void testRemoteLogReaderWithoutError() throws RemoteStorageException, IOException { +FetchDataInfo fetchDataInfo = new FetchDataInfo(logOffsetMetadata, records); + when(mockRLM.read(any(RemoteStorageFetchInfo.class))).thenReturn(fetchDataInfo); + +Consumer callback = mock(Consumer.class); +RemoteStorageFetchInfo remoteStorageFetchInfo = new RemoteStorageFetchInfo(0, false, new TopicPartition("test", 0), null, null, false); +RemoteLogReader remoteLogReader = new RemoteLogReader(remoteStorageFetchInfo, mockRLM, callback); +remoteLogReader.call(); + +// verify the callback did get invoked with the expected remoteLogReadResult +ArgumentCaptor remoteLogReadResultArg = ArgumentCaptor.forClass(RemoteLogReadResult.class); +verify(callback, times(1)).accept(remoteLogReadResultArg.capture()); +RemoteLogReadResult actualRemoteLogReadResult = remoteLogReadResultArg.getValue(); +assertFalse(actualRemoteLogReadResult.error.isPresent()); +assertTrue(actualRemoteLogReadResult.fetchDataInfo.isPresent()); +assertEquals(fetchDataInfo, actualRemoteLogReadResult.fetchDataInfo.get()); +} + +@Test +public void testRemoteLogReaderWithError() throws RemoteStorageException, IOException { Review Comment: We will add more tests in followup PRs. -- 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
[GitHub] [kafka] satishd commented on a diff in pull request #13535: KAFKA-9579 Fetch implementation for records in the remote storage through a specific purgatory.
satishd commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1188479417 ## core/src/main/scala/kafka/server/ReplicaManager.scala: ## @@ -1288,17 +1373,46 @@ class ReplicaManager(val config: KafkaConfig, _: FencedLeaderEpochException | _: ReplicaNotAvailableException | _: KafkaStorageException | - _: OffsetOutOfRangeException | _: InconsistentTopicIdException) => - LogReadResult(info = new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY), -divergingEpoch = None, -highWatermark = UnifiedLog.UnknownOffset, -leaderLogStartOffset = UnifiedLog.UnknownOffset, -leaderLogEndOffset = UnifiedLog.UnknownOffset, -followerLogStartOffset = UnifiedLog.UnknownOffset, -fetchTimeMs = -1L, -lastStableOffset = None, -exception = Some(e)) + createLogReadResult(e) +case e: OffsetOutOfRangeException => + // In case of offset out of range errors, check for remote log manager for non-compacted topics + // to fetch from remote storage. `log` instance should not be null here as that would have been caught earlier + // with NotLeaderForPartitionException or ReplicaNotAvailableException. + // If it is from a follower then send the offset metadata only as the data is already available in remote + // storage. + if (remoteLogManager.isDefined && log != null && log.remoteLogEnabled() && +// Check that the fetch offset is within the offset range within the remote storage layer. +log.logStartOffset <= offset && offset < log.localLogStartOffset()) { Review Comment: That should work fine because it will eventually throw offset out-of-range error if the target offset does not exist. -- 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
[jira] [Updated] (KAFKA-14974) Restore backward compatibility in KafkaBasedLog
[ https://issues.apache.org/jira/browse/KAFKA-14974?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Randall Hauch updated KAFKA-14974: -- Fix Version/s: 3.6.0 > Restore backward compatibility in KafkaBasedLog > --- > > Key: KAFKA-14974 > URL: https://issues.apache.org/jira/browse/KAFKA-14974 > Project: Kafka > Issue Type: Task > Components: KafkaConnect >Affects Versions: 3.5.0, 3.4.1, 3.3.3 >Reporter: Yash Mayya >Assignee: Yash Mayya >Priority: Major > Fix For: 3.6.0 > > > {{KafkaBasedLog}} is a widely used utility class that provides a generic > implementation of a shared, compacted log of records in a Kafka topic. It > isn't in Connect's public API, but has been used outside of Connect and we > try to preserve backward compatibility whenever possible. > https://issues.apache.org/jira/browse/KAFKA-14455 modified the two overloaded > void {{KafkaBasedLog::send}} methods to return a {{{}Future{}}}. While this > change is source compatible, it isn't binary compatible. We can restore > backward compatibility simply by re-instating the older send methods, and > renaming the new Future returning send methods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] rhauch merged pull request #13688: KAFKA-14974: Restore backward compatibility in KafkaBasedLog
rhauch merged PR #13688: URL: https://github.com/apache/kafka/pull/13688 -- 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
[jira] [Created] (KAFKA-14979) Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
hudeqi created KAFKA-14979: -- Summary: Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread Key: KAFKA-14979 URL: https://issues.apache.org/jira/browse/KAFKA-14979 Project: Kafka Issue Type: Improvement Components: core Affects Versions: 3.3.2 Reporter: hudeqi -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-14979) Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
[ https://issues.apache.org/jira/browse/KAFKA-14979?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] hudeqi reassigned KAFKA-14979: -- Assignee: hudeqi > Incorrect lag was calculated when markPartitionsForTruncation in > ReplicaAlterLogDirsThread > -- > > Key: KAFKA-14979 > URL: https://issues.apache.org/jira/browse/KAFKA-14979 > Project: Kafka > Issue Type: Improvement > Components: core >Affects Versions: 3.3.2 >Reporter: hudeqi >Assignee: hudeqi >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] satishd commented on a diff in pull request #13535: KAFKA-9579 Fetch implementation for records in the remote storage through a specific purgatory.
satishd commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1188566365 ## core/src/main/scala/kafka/server/ReplicaManager.scala: ## @@ -1288,17 +1373,46 @@ class ReplicaManager(val config: KafkaConfig, _: FencedLeaderEpochException | _: ReplicaNotAvailableException | _: KafkaStorageException | - _: OffsetOutOfRangeException | _: InconsistentTopicIdException) => - LogReadResult(info = new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY), -divergingEpoch = None, -highWatermark = UnifiedLog.UnknownOffset, -leaderLogStartOffset = UnifiedLog.UnknownOffset, -leaderLogEndOffset = UnifiedLog.UnknownOffset, -followerLogStartOffset = UnifiedLog.UnknownOffset, -fetchTimeMs = -1L, -lastStableOffset = None, -exception = Some(e)) + createLogReadResult(e) +case e: OffsetOutOfRangeException => + // In case of offset out of range errors, check for remote log manager for non-compacted topics + // to fetch from remote storage. `log` instance should not be null here as that would have been caught earlier + // with NotLeaderForPartitionException or ReplicaNotAvailableException. + // If it is from a follower then send the offset metadata only as the data is already available in remote + // storage. + if (remoteLogManager.isDefined && log != null && log.remoteLogEnabled() && +// Check that the fetch offset is within the offset range within the remote storage layer. +log.logStartOffset <= offset && offset < log.localLogStartOffset()) { +// For follower fetch requests, throw an error saying that this offset is moved to tiered storage. +val highWatermark = log.highWatermark +val leaderLogStartOffset = log.logStartOffset Review Comment: It is fine as the offset can always be updated, we will send whatever is the value available at that time. -- 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
[jira] [Assigned] (KAFKA-14966) Extract reusable common logic from OffsetFetcher
[ https://issues.apache.org/jira/browse/KAFKA-14966?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Lianet Magrans reassigned KAFKA-14966: -- Assignee: Lianet Magrans > Extract reusable common logic from OffsetFetcher > > > Key: KAFKA-14966 > URL: https://issues.apache.org/jira/browse/KAFKA-14966 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Major > > The OffsetFetcher is internally used by the KafkaConsumer to fetch offsets, > validate and reset positions. > For the new consumer based on a refactored threading model, similar > functionality will be needed by the ListOffsetsRequestManager component. > This task aims at identifying and extracting the OffsetFetcher functionality > that can be reused by the new consumer implementation. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-14965) Introduce ListOffsetsRequestManager to integrate ListOffsetsRequests into new consumer threading refactor
[ https://issues.apache.org/jira/browse/KAFKA-14965?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Lianet Magrans reassigned KAFKA-14965: -- Assignee: Lianet Magrans > Introduce ListOffsetsRequestManager to integrate ListOffsetsRequests into new > consumer threading refactor > - > > Key: KAFKA-14965 > URL: https://issues.apache.org/jira/browse/KAFKA-14965 > Project: Kafka > Issue Type: Task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Major > > This task introduces new functionality for handling ListOffsetsRequests for > the new consumer implementation, as part for the ongoing work for the > consumer threading model refactor. > This task introduces a new class named {{ListOffsetsRequestManager, > }}responsible of handling ListOffsets requests performed by the consumer to > expose functionality like beginningOffsets, endOffsets and offsetsForTimes. > The Offset{{{}Fetcher{}}} class is used internally by the {{KafkaConsumer}} > to list offsets, so this task will be based on a refactored > Offset{{{}Fetcher{}}}, reusing the fetching logic as much as possible. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] fvaleri commented on a diff in pull request #13516: KAFKA-14752: Kafka examples improvements - processor changes
fvaleri commented on code in PR #13516: URL: https://github.com/apache/kafka/pull/13516#discussion_r1188609481 ## examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java: ## @@ -79,111 +86,114 @@ public ExactlyOnceMessageProcessor(final String inputTopic, @Override public void run() { -// Init transactions call should always happen first in order to clear zombie transactions from previous generation. -producer.initTransactions(); - -final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); - -consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { -@Override -public void onPartitionsRevoked(Collection partitions) { -printWithTxnId("Revoked partition assignment to kick-off rebalancing: " + partitions); -} - -@Override -public void onPartitionsAssigned(Collection partitions) { -printWithTxnId("Received partition assignment after rebalancing: " + partitions); -messageRemaining.set(messagesRemaining(consumer)); -} -}); - -int messageProcessed = 0; -while (messageRemaining.get() > 0) { -try { -ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); -if (records.count() > 0) { -// Begin a new transaction session. -producer.beginTransaction(); -for (ConsumerRecord record : records) { -// Process the record and send to downstream. -ProducerRecord customizedRecord = transform(record); -producer.send(customizedRecord); +int processedRecords = 0; +long remainingRecords = Long.MAX_VALUE; +try { +// it is recommended to have a relatively short txn timeout in order to clear pending offsets faster +int transactionTimeoutMs = 10_000; +KafkaProducer producer = +new Producer(outputTopic, true, transactionalId, true, -1, transactionTimeoutMs, null).get(); + +// consumer must be in read_committed mode, which means it won't be able to read uncommitted data +boolean readCommitted = true; +KafkaConsumer consumer = new Consumer( +"processor-consumer", bootstrapServers, inputTopic, "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null) +.createKafkaConsumer(); + +// called first and once to fence zombies and abort any pending transaction +producer.initTransactions(); + +consumer.subscribe(singleton(inputTopic), this); + +Utils.printOut("Processing new records"); +while (!closed && remainingRecords > 0) { +try { +ConsumerRecords records = consumer.poll(ofMillis(200)); +if (!records.isEmpty()) { +// begin a new transaction session +producer.beginTransaction(); + +for (ConsumerRecord record : records) { +// process the record and send downstream +ProducerRecord newRecord = +new ProducerRecord<>(outputTopic, record.key(), record.value() + "-ok"); +producer.send(newRecord); +} + +// checkpoint the progress by sending offsets to group coordinator broker +// note that this API is only available for broker >= 2.5 + producer.sendOffsetsToTransaction(getOffsetsToCommit(consumer), consumer.groupMetadata()); + +// commit the transaction including offsets +producer.commitTransaction(); +processedRecords += records.count(); } - -Map offsets = consumerOffsets(); - -// Checkpoint the progress by sending offsets to group coordinator broker. -// Note that this API is only available for broker >= 2.5. -producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()); - -// Finish the transaction. All sent records should be visible for consumption now. -producer.commitTransaction(); -messageProcessed += records.count(); +} catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException + | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) { +Utils.printErr(e.getMessage()); +// we can't recover from these exceptions +shutdown(); +
[GitHub] [kafka] fvaleri commented on a diff in pull request #13516: KAFKA-14752: Kafka examples improvements - processor changes
fvaleri commented on code in PR #13516: URL: https://github.com/apache/kafka/pull/13516#discussion_r1188609885 ## examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java: ## @@ -79,111 +86,114 @@ public ExactlyOnceMessageProcessor(final String inputTopic, @Override public void run() { -// Init transactions call should always happen first in order to clear zombie transactions from previous generation. -producer.initTransactions(); - -final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); - -consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { -@Override -public void onPartitionsRevoked(Collection partitions) { -printWithTxnId("Revoked partition assignment to kick-off rebalancing: " + partitions); -} - -@Override -public void onPartitionsAssigned(Collection partitions) { -printWithTxnId("Received partition assignment after rebalancing: " + partitions); -messageRemaining.set(messagesRemaining(consumer)); -} -}); - -int messageProcessed = 0; -while (messageRemaining.get() > 0) { -try { -ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); -if (records.count() > 0) { -// Begin a new transaction session. -producer.beginTransaction(); -for (ConsumerRecord record : records) { -// Process the record and send to downstream. -ProducerRecord customizedRecord = transform(record); -producer.send(customizedRecord); +int processedRecords = 0; +long remainingRecords = Long.MAX_VALUE; +try { +// it is recommended to have a relatively short txn timeout in order to clear pending offsets faster +int transactionTimeoutMs = 10_000; +KafkaProducer producer = +new Producer(outputTopic, true, transactionalId, true, -1, transactionTimeoutMs, null).get(); + +// consumer must be in read_committed mode, which means it won't be able to read uncommitted data +boolean readCommitted = true; +KafkaConsumer consumer = new Consumer( +"processor-consumer", bootstrapServers, inputTopic, "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null) +.createKafkaConsumer(); + +// called first and once to fence zombies and abort any pending transaction +producer.initTransactions(); + +consumer.subscribe(singleton(inputTopic), this); + +Utils.printOut("Processing new records"); +while (!closed && remainingRecords > 0) { +try { +ConsumerRecords records = consumer.poll(ofMillis(200)); +if (!records.isEmpty()) { +// begin a new transaction session +producer.beginTransaction(); + +for (ConsumerRecord record : records) { +// process the record and send downstream +ProducerRecord newRecord = +new ProducerRecord<>(outputTopic, record.key(), record.value() + "-ok"); +producer.send(newRecord); +} + +// checkpoint the progress by sending offsets to group coordinator broker +// note that this API is only available for broker >= 2.5 + producer.sendOffsetsToTransaction(getOffsetsToCommit(consumer), consumer.groupMetadata()); + +// commit the transaction including offsets +producer.commitTransaction(); +processedRecords += records.count(); } - -Map offsets = consumerOffsets(); - -// Checkpoint the progress by sending offsets to group coordinator broker. -// Note that this API is only available for broker >= 2.5. -producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()); - -// Finish the transaction. All sent records should be visible for consumption now. -producer.commitTransaction(); -messageProcessed += records.count(); +} catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException + | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) { Review Comment: Well, this is consistent with the consumer. The exception list here is shorter and much more compact (no repeated instanceof). If you don't mi
[jira] [Commented] (KAFKA-7739) Kafka Tiered Storage
[ https://issues.apache.org/jira/browse/KAFKA-7739?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17720936#comment-17720936 ] Jorge Esteban Quilcate Otoya commented on KAFKA-7739: - [~satish.duggana] could you help clarifying the following: RemoteStorageManager#fetchIndex and RemoteStorageManager#fetchLogSegment state: > Â Â Â * @throws RemoteResourceNotFoundException when there are no resources > associated with the given remoteLogSegmentMetadata. Though when fetching indexes: Transactional Index is optional. So when returning a response should it be null? or throw RemoteResourceNotFoundException? >From the javadoc is unclear whether resources means a specific object/index or >the whole segment log + metadata. > Kafka Tiered Storage > > > Key: KAFKA-7739 > URL: https://issues.apache.org/jira/browse/KAFKA-7739 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Harsha >Assignee: Satish Duggana >Priority: Major > Labels: needs-kip > Fix For: 3.6.0 > > > KIP:Â > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-14979) Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
[ https://issues.apache.org/jira/browse/KAFKA-14979?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] hudeqi updated KAFKA-14979: --- Description: When the partitions of ReplicaFetcherThread finished truncating, the ReplicaAlterLogDirsThread to which these partitions belong needs to be marked truncate. The lag value in the newState (PartitionFetchState) obtained in this process is still the original value (state.lag). If the truncationOffset is smaller than the original state.fetchOffset, then the original lag value is incorrect and needs to be updated. It should be the original lag value plus the difference between the original state.fetchOffset and truncationOffset. If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread may set the wrong lag value for fetcherLagStats when executing processFetchRequest. So it might be more reasonable to recalculate a lag value based on new state.fetchOffset. > Incorrect lag was calculated when markPartitionsForTruncation in > ReplicaAlterLogDirsThread > -- > > Key: KAFKA-14979 > URL: https://issues.apache.org/jira/browse/KAFKA-14979 > Project: Kafka > Issue Type: Improvement > Components: core >Affects Versions: 3.3.2 >Reporter: hudeqi >Assignee: hudeqi >Priority: Major > > When the partitions of ReplicaFetcherThread finished truncating, the > ReplicaAlterLogDirsThread to which these partitions belong needs to be marked > truncate. The lag value in the newState (PartitionFetchState) obtained in > this process is still the original value (state.lag). If the truncationOffset > is smaller than the original state.fetchOffset, then the original lag value > is incorrect and needs to be updated. It should be the original lag value > plus the difference between the original state.fetchOffset and > truncationOffset. > If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread > may set the wrong lag value for fetcherLagStats when executing > processFetchRequest. So it might be more reasonable to recalculate a lag > value based on new state.fetchOffset. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-14979) Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
[ https://issues.apache.org/jira/browse/KAFKA-14979?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] hudeqi updated KAFKA-14979: --- Description: When the partitions of ReplicaFetcherThread finished truncating, the ReplicaAlterLogDirsThread to which these partitions belong needs to be marked truncate. The lag value in the newState (PartitionFetchState) obtained in this process is still the original value (state.lag). If the truncationOffset is smaller than the original state.fetchOffset, then the original lag value is incorrect and needs to be updated. It should be the original lag value plus the difference between the original state.fetchOffset and truncationOffset. If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread may set the wrong lag value for fetcherLagStats when executing "processFetchRequest". So it might be more reasonable to recalculate a lag value based on new state.fetchOffset. was: When the partitions of ReplicaFetcherThread finished truncating, the ReplicaAlterLogDirsThread to which these partitions belong needs to be marked truncate. The lag value in the newState (PartitionFetchState) obtained in this process is still the original value (state.lag). If the truncationOffset is smaller than the original state.fetchOffset, then the original lag value is incorrect and needs to be updated. It should be the original lag value plus the difference between the original state.fetchOffset and truncationOffset. If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread may set the wrong lag value for fetcherLagStats when executing processFetchRequest. So it might be more reasonable to recalculate a lag value based on new state.fetchOffset. > Incorrect lag was calculated when markPartitionsForTruncation in > ReplicaAlterLogDirsThread > -- > > Key: KAFKA-14979 > URL: https://issues.apache.org/jira/browse/KAFKA-14979 > Project: Kafka > Issue Type: Improvement > Components: core >Affects Versions: 3.3.2 >Reporter: hudeqi >Assignee: hudeqi >Priority: Major > > When the partitions of ReplicaFetcherThread finished truncating, the > ReplicaAlterLogDirsThread to which these partitions belong needs to be marked > truncate. The lag value in the newState (PartitionFetchState) obtained in > this process is still the original value (state.lag). If the truncationOffset > is smaller than the original state.fetchOffset, then the original lag value > is incorrect and needs to be updated. It should be the original lag value > plus the difference between the original state.fetchOffset and > truncationOffset. > If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread > may set the wrong lag value for fetcherLagStats when executing > "processFetchRequest". So it might be more reasonable to recalculate a lag > value based on new state.fetchOffset. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] hudeqi opened a new pull request, #13692: KAFKA-14979:Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
hudeqi opened a new pull request, #13692: URL: https://github.com/apache/kafka/pull/13692 When the partitions of ReplicaFetcherThread finished truncating, the ReplicaAlterLogDirsThread to which these partitions belong needs to be marked truncate. The lag value in the newState (PartitionFetchState) obtained in this process is still the original value (state.lag). If the truncationOffset is smaller than the original state.fetchOffset, then the original lag value is incorrect and needs to be updated. It should be the original lag value plus the difference between the original state.fetchOffset and truncationOffset. If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread may set the wrong lag value for fetcherLagStats when executing "processFetchRequest". So it might be more reasonable to recalculate a lag value based on new state.fetchOffset. -- 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
[GitHub] [kafka] akatona84 commented on pull request #11565: KAFKA-13504: Retry connect internal topics' creation in case of InvalidReplicationFactorException
akatona84 commented on PR #11565: URL: https://github.com/apache/kafka/pull/11565#issuecomment-1540178836 rebased -- 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
[GitHub] [kafka] Hangleton commented on a diff in pull request #13584: MINOR: Add log segment unit tests, If the maximum offset beyond index, appen…
Hangleton commented on code in PR #13584: URL: https://github.com/apache/kafka/pull/13584#discussion_r1188635591 ## core/src/test/scala/unit/kafka/log/LogSegmentTest.scala: ## @@ -65,6 +68,32 @@ class LogSegmentTest { Utils.delete(logDir) } + /** + * LogSegmentOffsetOverflowException should be thrown while appending the logs if: + * 1. largestOffset < 0 Review Comment: Nit: we can remove (1) since (1) => (2). ## core/src/test/scala/unit/kafka/log/LogSegmentTest.scala: ## @@ -65,6 +68,32 @@ class LogSegmentTest { Utils.delete(logDir) } + /** + * LogSegmentOffsetOverflowException should be thrown while appending the logs if: + * 1. largestOffset < 0 + * 2. largestOffset - baseOffset < 0 + * 3. largestOffset - baseOffset > Integer.MAX_VALUE + */ + @ParameterizedTest + @CsvSource(Array( +"0, -2147483648", +"0, 2147483648", +"1, 0", +"100, 10", +"2147483648, 0", +"-2147483648, 0", +"2147483648,4294967296" Review Comment: nit: Note that the max int value on the Java platform is `2147483647` so we are exercising max int + 1. ## core/src/test/scala/unit/kafka/log/LogSegmentTest.scala: ## @@ -65,6 +68,32 @@ class LogSegmentTest { Utils.delete(logDir) } + /** + * LogSegmentOffsetOverflowException should be thrown while appending the logs if: + * 1. largestOffset < 0 + * 2. largestOffset - baseOffset < 0 + * 3. largestOffset - baseOffset > Integer.MAX_VALUE + */ + @ParameterizedTest + @CsvSource(Array( +"0, -2147483648", +"0, 2147483648", +"1, 0", +"100, 10", +"2147483648, 0", +"-2147483648, 0", +"2147483648,4294967296" + )) + def testAppendForLogSegmentOffsetOverflowException(baseOffset: Long, largestOffset: Long): Unit = { Review Comment: What is the expected behaviour for `baseOffset == largestOffset`? -- 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
[GitHub] [kafka] hudeqi commented on pull request #13473: KAFKA-14866:Remove controller module metrics when broker is shutting down
hudeqi commented on PR #13473: URL: https://github.com/apache/kafka/pull/13473#issuecomment-1540185668 bump this pr, Is there any other commiters to help review and merge this pr? -- 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
[jira] [Commented] (KAFKA-7739) Kafka Tiered Storage
[ https://issues.apache.org/jira/browse/KAFKA-7739?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17720947#comment-17720947 ] Alexandre Dupriez commented on KAFKA-7739: -- I would think that the absence of the log segment or any mandatory ancillary file results in the stated exception. I think only transaction indexes are optional. > Kafka Tiered Storage > > > Key: KAFKA-7739 > URL: https://issues.apache.org/jira/browse/KAFKA-7739 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Harsha >Assignee: Satish Duggana >Priority: Major > Labels: needs-kip > Fix For: 3.6.0 > > > KIP:Â > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] machi1990 opened a new pull request, #13693: MINOR: fix a small typo in SharedServer.scala
machi1990 opened a new pull request, #13693: URL: https://github.com/apache/kafka/pull/13693 I noticed this typo as I was going through the SharedServer.scala file ### 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] hudeqi commented on pull request #13473: KAFKA-14866:Remove controller module metrics when broker is shutting down
hudeqi commented on PR #13473: URL: https://github.com/apache/kafka/pull/13473#issuecomment-1540199229 > bump this pr, Is there any other commiters to help review and merge this pr? I'm sorry I didn't notice the comment ("We're past code freeze for 3.5 so moving this to the next release") in the corresponding [jira](https://issues.apache.org/jira/browse/KAFKA-14866) -- 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
[jira] [Updated] (KAFKA-14414) Remove unnecessary usage of ObjectSerializationCache
[ https://issues.apache.org/jira/browse/KAFKA-14414?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Divij Vaidya updated KAFKA-14414: - Attachment: producer-only-cpu-33-clean.html > Remove unnecessary usage of ObjectSerializationCache > > > Key: KAFKA-14414 > URL: https://issues.apache.org/jira/browse/KAFKA-14414 > Project: Kafka > Issue Type: Improvement > Components: core >Reporter: Divij Vaidya >Assignee: Divij Vaidya >Priority: Major > Fix For: 3.4.0 > > Attachments: Screenshot 2022-11-21 at 19.23.53.png, > producer-only-cpu-33-clean.html > > > We create an instance of ObjectSerializationCache at > [https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/network/RequestChannel.scala#L113] > which does not get used at all. We always "add" to the cache but never > retrieve from it (as is evident by the fact that we don't store the reference > of the cache anywhere). > Adding information to the cache is expensive because it uses > System.identityHashCode(Object) which is expensive as demonstrated by the > flame graph of producer requests over Apache Kafka 3.3.1 plaintext broker. > {{!Screenshot 2022-11-21 at 19.23.53.png!}} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] C0urante commented on pull request #13690: KAFKA-14978 ExactlyOnceWorkerSourceTask should remove parent metrics
C0urante commented on PR #13690: URL: https://github.com/apache/kafka/pull/13690#issuecomment-1540232079 Tests? -- 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
[GitHub] [kafka] jeffkbkim commented on pull request #13644: KAFKA-14500; [1/N] Rewrite MemberMetadata in Java
jeffkbkim commented on PR #13644: URL: https://github.com/apache/kafka/pull/13644#issuecomment-1540246709 @dajac the test failure is unrelated. can we merge this? -- 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
[GitHub] [kafka] dajac merged pull request #13644: KAFKA-14500; [1/N] Rewrite MemberMetadata in Java
dajac merged PR #13644: URL: https://github.com/apache/kafka/pull/13644 -- 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
[GitHub] [kafka] dajac commented on pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on PR #13443: URL: https://github.com/apache/kafka/pull/13443#issuecomment-1540289311 @rreddy-22 The build failed due to compilation errors. -- 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
[jira] [Commented] (KAFKA-7739) Kafka Tiered Storage
[ https://issues.apache.org/jira/browse/KAFKA-7739?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17720955#comment-17720955 ] Jorge Esteban Quilcate Otoya commented on KAFKA-7739: - Yes, we know transactional index is optional when copying but not when fetching. I agree throwing the exception should be correct, but the javadoc is ambiguous as it refers to "no resources associated with the given remoteLogSegmentMetadata" that may mean _all_ resources, instead of "the requested resource is not found in the remote storage" or similar. > Kafka Tiered Storage > > > Key: KAFKA-7739 > URL: https://issues.apache.org/jira/browse/KAFKA-7739 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Harsha >Assignee: Satish Duggana >Priority: Major > Labels: needs-kip > Fix For: 3.6.0 > > > KIP:Â > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] dajac commented on a diff in pull request #13663: KAFKA-14500; [2/N] Rewrite GroupMetadata in Java
dajac commented on code in PR #13663: URL: https://github.com/apache/kafka/pull/13663#discussion_r1188744749 ## core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala: ## @@ -160,6 +160,7 @@ class GroupMetadataTest { assertThrows(classOf[IllegalStateException], () => group.transitionTo(CompletingRebalance)) } + @Test Review Comment: Nice catch! Do you mind doing a separate PR for this one? This way we can keep changes in this PR scoped in the new module. -- 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
[GitHub] [kafka] jeffkbkim opened a new pull request, #13694: MINOR: add test tag for testDeadToDeadIllegalTransition
jeffkbkim opened a new pull request, #13694: URL: https://github.com/apache/kafka/pull/13694 *More detailed description of your change, if necessary. The PR title and PR message become the squashed commit message, so use a separate comment to ping reviewers.* *Summary of testing strategy (including rationale) for the feature or bug fix. Unit and/or integration tests are expected for any behaviour change and system tests should be considered for larger changes.* ### 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13663: KAFKA-14500; [2/N] Rewrite GroupMetadata in Java
jeffkbkim commented on code in PR #13663: URL: https://github.com/apache/kafka/pull/13663#discussion_r1188772104 ## core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataTest.scala: ## @@ -160,6 +160,7 @@ class GroupMetadataTest { assertThrows(classOf[IllegalStateException], () => group.transitionTo(CompletingRebalance)) } + @Test Review Comment: created https://github.com/apache/kafka/pull/13694 -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188811581 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] urbandan commented on pull request #13690: KAFKA-14978 ExactlyOnceWorkerSourceTask should remove parent metrics
urbandan commented on PR #13690: URL: https://github.com/apache/kafka/pull/13690#issuecomment-1540445965 Added a unit test -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188817725 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188819218 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188821039 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188821932 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188823115 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +
[GitHub] [kafka] kirktrue closed pull request #13495: KAFKA-14274: Introduce FetchRequestManager to integrate fetch into new consumer threading refactor
kirktrue closed pull request #13495: KAFKA-14274: Introduce FetchRequestManager to integrate fetch into new consumer threading refactor URL: https://github.com/apache/kafka/pull/13495 -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188832049 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); Review
[jira] [Updated] (KAFKA-14974) Restore backward compatibility in KafkaBasedLog
[ https://issues.apache.org/jira/browse/KAFKA-14974?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Randall Hauch updated KAFKA-14974: -- Fix Version/s: 3.4.1 3.3.3 > Restore backward compatibility in KafkaBasedLog > --- > > Key: KAFKA-14974 > URL: https://issues.apache.org/jira/browse/KAFKA-14974 > Project: Kafka > Issue Type: Task > Components: KafkaConnect >Affects Versions: 3.5.0, 3.4.1, 3.3.3 >Reporter: Yash Mayya >Assignee: Yash Mayya >Priority: Major > Fix For: 3.4.1, 3.3.3, 3.6.0 > > > {{KafkaBasedLog}} is a widely used utility class that provides a generic > implementation of a shared, compacted log of records in a Kafka topic. It > isn't in Connect's public API, but has been used outside of Connect and we > try to preserve backward compatibility whenever possible. > https://issues.apache.org/jira/browse/KAFKA-14455 modified the two overloaded > void {{KafkaBasedLog::send}} methods to return a {{{}Future{}}}. While this > change is source compatible, it isn't binary compatible. We can restore > backward compatibility simply by re-instating the older send methods, and > renaming the new Future returning send methods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188856580 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of the ex
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188857767 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of the ex
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188858930 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[jira] [Created] (KAFKA-14980) MirrorMaker consumers don't get configs prefixed with source.cluster
Mickael Maison created KAFKA-14980: -- Summary: MirrorMaker consumers don't get configs prefixed with source.cluster Key: KAFKA-14980 URL: https://issues.apache.org/jira/browse/KAFKA-14980 Project: Kafka Issue Type: Bug Components: mirrormaker Affects Versions: 3.5.0 Reporter: Mickael Maison As part of KAFKA-14021, we made a change to MirrorConnectorConfig.sourceConsumerConfig() to grab all configs that start with "source.". Previously it was grabbing configs prefixed with "source.cluster.". This means existing connector configuration stop working, as configurations such as bootstrap.servers are not passed to source consumers. For example, the following connector configuration was valid in 3.4 and now makes the connector tasks fail: {code:json} { "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector", "name": "source", "topics": "test", "tasks.max": "30", "source.cluster.alias": "one", "target.cluster.alias": "two", "source.cluster.bootstrap.servers": "localhost:9092", "target.cluster.bootstrap.servers": "localhost:29092" } {code} The connector attempts to start source consumers with bootstrap.servers = [] and the task crash with {noformat} org.apache.kafka.common.KafkaException: Failed to construct kafka consumer at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:837) at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:671) at org.apache.kafka.connect.mirror.MirrorUtils.newConsumer(MirrorUtils.java:59) at org.apache.kafka.connect.mirror.MirrorSourceTask.start(MirrorSourceTask.java:103) at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.initializeAndStart(AbstractWorkerSourceTask.java:274) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259) at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75) at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers {noformat} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (KAFKA-14980) MirrorMaker consumers don't get configs prefixed with source.cluster
[ https://issues.apache.org/jira/browse/KAFKA-14980?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17720993#comment-17720993 ] Mickael Maison commented on KAFKA-14980: cc [~ChrisEgerton]Â > MirrorMaker consumers don't get configs prefixed with source.cluster > > > Key: KAFKA-14980 > URL: https://issues.apache.org/jira/browse/KAFKA-14980 > Project: Kafka > Issue Type: Bug > Components: mirrormaker >Affects Versions: 3.5.0 >Reporter: Mickael Maison >Priority: Blocker > > As part of KAFKA-14021, we made a change to > MirrorConnectorConfig.sourceConsumerConfig() to grab all configs that start > with "source.". Previously it was grabbing configs prefixed with > "source.cluster.". > This means existing connector configuration stop working, as configurations > such as bootstrap.servers are not passed to source consumers. > For example, the following connector configuration was valid in 3.4 and now > makes the connector tasks fail: > {code:json} > { > "connector.class": > "org.apache.kafka.connect.mirror.MirrorSourceConnector", > "name": "source", > "topics": "test", > "tasks.max": "30", > "source.cluster.alias": "one", > "target.cluster.alias": "two", > "source.cluster.bootstrap.servers": "localhost:9092", >"target.cluster.bootstrap.servers": "localhost:29092" > } > {code} > The connector attempts to start source consumers with bootstrap.servers = [] > and the task crash with > {noformat} > org.apache.kafka.common.KafkaException: Failed to construct kafka consumer > at > org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:837) > at > org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:671) > at > org.apache.kafka.connect.mirror.MirrorUtils.newConsumer(MirrorUtils.java:59) > at > org.apache.kafka.connect.mirror.MirrorSourceTask.start(MirrorSourceTask.java:103) > at > org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.initializeAndStart(AbstractWorkerSourceTask.java:274) > at > org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202) > at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259) > at > org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75) > at > org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181) > at > java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) > at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) > at > java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) > at > java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) > at java.base/java.lang.Thread.run(Thread.java:829) > Caused by: org.apache.kafka.common.config.ConfigException: No resolvable > bootstrap urls given in bootstrap.servers > {noformat} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188873259 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +a
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188891203 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188891494 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + * Fi
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188893478 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { Review Comment: nit: ` : `. -- 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
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188893117 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + * Fi
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1188898373 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + * Fi
[jira] [Commented] (KAFKA-7739) Kafka Tiered Storage
[ https://issues.apache.org/jira/browse/KAFKA-7739?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17721025#comment-17721025 ] Alexandre Dupriez commented on KAFKA-7739: -- Got it. Sorry I misunderstood your suggestion. Thanks for clarifying! (and agree with it). > Kafka Tiered Storage > > > Key: KAFKA-7739 > URL: https://issues.apache.org/jira/browse/KAFKA-7739 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Harsha >Assignee: Satish Duggana >Priority: Major > Labels: needs-kip > Fix For: 3.6.0 > > > KIP:Â > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage] -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-14974) Restore backward compatibility in KafkaBasedLog
[ https://issues.apache.org/jira/browse/KAFKA-14974?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Randall Hauch updated KAFKA-14974: -- Fix Version/s: 3.5.0 > Restore backward compatibility in KafkaBasedLog > --- > > Key: KAFKA-14974 > URL: https://issues.apache.org/jira/browse/KAFKA-14974 > Project: Kafka > Issue Type: Task > Components: KafkaConnect >Affects Versions: 3.5.0, 3.4.1, 3.3.3 >Reporter: Yash Mayya >Assignee: Yash Mayya >Priority: Major > Fix For: 3.5.0, 3.4.1, 3.3.3, 3.6.0 > > > {{KafkaBasedLog}} is a widely used utility class that provides a generic > implementation of a shared, compacted log of records in a Kafka topic. It > isn't in Connect's public API, but has been used outside of Connect and we > try to preserve backward compatibility whenever possible. > https://issues.apache.org/jira/browse/KAFKA-14455 modified the two overloaded > void {{KafkaBasedLog::send}} methods to return a {{{}Future{}}}. While this > change is source compatible, it isn't binary compatible. We can restore > backward compatibility simply by re-instating the older send methods, and > renaming the new Future returning send methods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] ableegoldman opened a new pull request, #13695: HOTFIX: fix the VersionedKeyValueToBytesStoreAdapter#isOpen API
ableegoldman opened a new pull request, #13695: URL: https://github.com/apache/kafka/pull/13695 The VersionedKeyValueToBytesStoreAdapter#isOpen API accidentally returns the value of `inner.persistent()` when it should be returning `inner.isOpen()` -- 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
[GitHub] [kafka] ableegoldman commented on a diff in pull request #13682: MINOR: improved exception/warn logging for stream-stream join store settings
ableegoldman commented on code in PR #13682: URL: https://github.com/apache/kafka/pull/13682#discussion_r1189116797 ## streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java: ## @@ -115,7 +115,7 @@ public boolean persistent() { @Override public boolean isOpen() { -return inner.persistent(); +return inner.isOpen(); Review Comment: Good point. Here: https://github.com/apache/kafka/pull/13695 -- 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
[GitHub] [kafka] ableegoldman commented on pull request #13695: HOTFIX: fix the VersionedKeyValueToBytesStoreAdapter#isOpen API
ableegoldman commented on PR #13695: URL: https://github.com/apache/kafka/pull/13695#issuecomment-1540876914 @vcrfxia @mjsax pulled this fix out for the 3.5 release (cc @mimaison ) -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189210485 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ Review Comment: It looks a bit wonky after formatting it like that, I don't think there's a great way to add this html -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189212267 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ Review Comment: I did it in a way that looks best to me in the next commit -- 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
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189213049 ## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java: ## @@ -0,0 +1,523 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class RangeAssignorTest { +private final RangeAssignor assignor = new RangeAssignor(); +private final Uuid topic1Uuid = Uuid.randomUuid(); +private final Uuid topic2Uuid = Uuid.randomUuid(); +private final Uuid topic3Uuid = Uuid.randomUuid(); +private final String consumerA = "A"; +private final String consumerB = "B"; +private final String consumerC = "C"; + +@Test +public void testOneConsumerNoTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.emptyList(), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertEquals(Collections.emptyMap(), groupAssignment.members()); +} + +@Test +public void testOneConsumerSubscribedToNonExistentTopic() { +Map topics = Collections.singletonMap(topic1Uuid, new AssignmentTopicMetadata(3)); +Map members = Collections.singletonMap( +consumerA, +new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Collections.singletonList(topic2Uuid), +Collections.emptyMap() +) +); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment groupAssignment = assignor.assign(assignmentSpec); + +assertTrue(groupAssignment.members().isEmpty()); +} + +@Test +public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() { +Map topics = new HashMap<>(); +topics.put(topic1Uuid, new AssignmentTopicMetadata(3)); +topics.put(topic3Uuid, new AssignmentTopicMetadata(2)); + +Map members = new TreeMap<>(); + +members.put(consumerA, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +members.put(consumerB, new AssignmentMemberSpec( +Optional.empty(), +Optional.empty(), +Arrays.asList(topic1Uuid, topic3Uuid), +Collections.emptyMap() +)); + +AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics); +GroupAssignment computedAssignment = assignor.assign(assignmentSpec); + +Map>> expectedAssignment = new HashMap<>(); +// Topic 1 Partitions Assignment +expectedAssignment.put(consumerA, mkAssignment(mkTopicAssignment(topic1Uuid, 0, 1))); +expectedAssignment.put(consumerB, mkAssignment(mkTopicAssignment(topic1Uuid, 2))); +// Topic 3 Partitions Assignment + expectedAssignment.get(consumerA).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 0))); + expectedAssignment.get(consumerB).putAll(mkAssignment(mkTopicAssignment(topic3Uuid, 1))); + +
[jira] [Created] (KAFKA-14981) Set `group.instance.id` in streams consumer so that rebalance will not happen if a instance is restarted
Hao Li created KAFKA-14981: -- Summary: Set `group.instance.id` in streams consumer so that rebalance will not happen if a instance is restarted Key: KAFKA-14981 URL: https://issues.apache.org/jira/browse/KAFKA-14981 Project: Kafka Issue Type: Improvement Components: streams Reporter: Hao Li `group.instance.id` enables static membership so that if a consumer is restarted within `session.timeout.ms`, rebalance will not be triggered and originally assignment can be returned directly from broker. We can set this id in Kafka streams using `threadId` so that no rebalance is trigger within `session.timeout.ms` -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189214614 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189216148 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189216494 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[jira] [Commented] (KAFKA-14981) Set `group.instance.id` in streams consumer so that rebalance will not happen if a instance is restarted
[ https://issues.apache.org/jira/browse/KAFKA-14981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17721114#comment-17721114 ] Matthias J. Sax commented on KAFKA-14981: - Very interesting idea – given that we persist the thread-id (aka process-id) in the state directory on local disk, it could help. And even if we don't persist it (because there is no local storage), it seems no harm would be done if the id changes every single time. Wondering if we would need a KIP for this. By gut feeling is no, but not sure. > Set `group.instance.id` in streams consumer so that rebalance will not happen > if a instance is restarted > > > Key: KAFKA-14981 > URL: https://issues.apache.org/jira/browse/KAFKA-14981 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Hao Li >Priority: Minor > > `group.instance.id` enables static membership so that if a consumer is > restarted within `session.timeout.ms`, rebalance will not be triggered and > originally assignment can be returned directly from broker. We can set this > id in Kafka streams using `threadId` so that no rebalance is trigger within > `session.timeout.ms` -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189220058 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189229522 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189229707 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189229928 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189231412 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189231170 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189231570 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189220058 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189235728 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,228 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of members per topic using the given member subscriptions. + * Generate a list of members called potentially unfilled members, which consists of members that have not met the minimum required quota of partitions for the assignment AND + * get a list called assigned sticky partitions per topic, which has the partitions that will be retained in the new assignment. + * Generate a list of unassigned partitions by calculating the difference between the total partitions for the topic and the assigned (sticky) partitions. + *
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189237663 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of th
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189295560 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of the ex
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189296233 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of the ex
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189296779 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ +public class RangeAssignor implements PartitionAssignor { +private static final Logger log = LoggerFactory.getLogger(RangeAssignor.class); + +public static final String RANGE_ASSIGNOR_NAME = "range"; + +@Override +public String name() { +return RANGE_ASSIGNOR_NAME; +} + +/** + * Pair of memberId and remaining partitions to meet the quota. + */ +private static class MemberWithRemainingAssignments { +/** + * Member Id. + */ +private final String memberId; +/** + * Number of partitions required to meet the assignment quota. + */ +private final Integer remaining; + +public MemberWithRemainingAssignments(String memberId, Integer remaining) { +this.memberId = memberId; +this.remaining = remaining; +} +} + +/** + * @return Map of topicIds to a list of members subscribed to them. + */ +private Map> membersPerTopic(final AssignmentSpec assignmentSpec) { +Map> membersPerTopic = new HashMap<>(); +Map membersData = assignmentSpec.members(); + +membersData.forEach((memberId, memberMetadata) -> { +Collection topics = memberMetadata.subscribedTopicIds(); +for (Uuid topicId: topics) { +// Only topics that are present in both the subscribed topics list and the topic metadata should be considered for assignment. +if (assignmentSpec.topics().containsKey(topicId)) { +membersPerTopic +.computeIfAbsent(topicId, k -> new ArrayList<>()) +.add(memberId); +} else { +log.warn("Member " + memberId + " subscribed to topic " + topicId + " which doesn't exist in the topic metadata"); +} +} +}); + +return membersPerTopic; +} + +/** + * The algorithm includes the following steps: + * + * Generate a map of membersPerTopic using the given member subscriptions. + * Generate a list of members (potentiallyUnfilledMembers) that have not met the minimum required quota of partitions for the assignment AND + * get a list (assignedStickyPartitionsPerTopic) of partitions that will be retained in the new assignment. + * Add members from the potentiallyUnfilledMembers list to the unfilledMembersPerTopic map if they haven't met the total required quota + * i.e. minRequiredQuota + 1, if the member is designated to receive one of the ex
[GitHub] [kafka] dajac commented on a diff in pull request #13443: KAFKA-14514: Add Range Assignor on the Server (KIP-848)
dajac commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1189297499 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,238 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * This Range Assignor inherits properties of both the range assignor and the sticky assignor. + * The properties are as follows: + * + * Each member must get at least one partition from every topic that it is subscribed to. The only exception is when + * the number of subscribed members is greater than the number of partitions for that topic. (Range) + * Partitions should be assigned to members in a way that facilitates the join operation when required. (Range) + * This can only be done if every member is subscribed to the same topics and the topics are co-partitioned. + * Two streams are co-partitioned if the following conditions are met: + * + * The keys must have the same schemas. + * The topics involved must have the same number of partitions. + * + * + * Members should retain as much of their previous assignment as possible to reduce the number of partition movements during reassignment. (Sticky) + * + */ Review Comment: Looks good, 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] hudeqi opened a new pull request, #13696: KAFKA-14979:Incorrect lag was calculated when markPartitionsForTruncation in ReplicaAlterLogDirsThread
hudeqi opened a new pull request, #13696: URL: https://github.com/apache/kafka/pull/13696 When the partitions of ReplicaFetcherThread finished truncating, the ReplicaAlterLogDirsThread to which these partitions belong needs to be marked truncate. The lag value in the newState (PartitionFetchState) obtained in this process is still the original value (state.lag). If the truncationOffset is smaller than the original state.fetchOffset, then the original lag value is incorrect and needs to be updated. It should be the original lag value plus the difference between the original state.fetchOffset and truncationOffset. If the original lag value is used incorrectly, then ReplicaAlterLogDirsThread may set the wrong lag value for fetcherLagStats when executing "processFetchRequest". So it might be more reasonable to recalculate a lag value based on new state.fetchOffset. -- 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