You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2025/05/09 15:07:21 UTC
[PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
apoorvmittal10 opened a new pull request, #19666:
URL: https://github.com/apache/kafka/pull/19666
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19666:
URL: https://github.com/apache/kafka/pull/19666#discussion_r2085658136
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
Review Comment:
Done.
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19666:
URL: https://github.com/apache/kafka/pull/19666#discussion_r2085659033
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
+ * The partition cache map is used to store the SharePartition objects for each share group topic-partition.
+ */
+ private final Map<SharePartitionKey, SharePartition> partitionMap;
+
+ SharePartitionCache() {
+ this.groupMap = new HashMap<>();
+ this.partitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SharePartition get(SharePartitionKey partitionKey) {
+ return partitionMap.get(partitionKey);
+ }
+
+ public SharePartition put(SharePartitionKey partitionKey, SharePartition sharePartition) {
Review Comment:
Actually it's only being used in tests, I wrote the comments on such methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19666:
URL: https://github.com/apache/kafka/pull/19666#discussion_r2084054302
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
+ * The partition cache map is used to store the SharePartition objects for each share group topic-partition.
+ */
+ private final Map<SharePartitionKey, SharePartition> partitionMap;
+
+ SharePartitionCache() {
+ this.groupMap = new HashMap<>();
+ this.partitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SharePartition get(SharePartitionKey partitionKey) {
+ return partitionMap.get(partitionKey);
+ }
+
+ public SharePartition put(SharePartitionKey partitionKey, SharePartition sharePartition) {
Review Comment:
Do we really need a `put` method? I would think that we need to make sure the two maps are in-sync, so having a method to put into just one of them sounds strange.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -651,14 +652,15 @@ void processShareFetch(ShareFetch shareFetch) {
}
private SharePartition getOrCreateSharePartition(SharePartitionKey sharePartitionKey) {
- return partitionCacheMap.computeIfAbsent(sharePartitionKey,
+ return partitionCache.computeIfAbsent(sharePartitionKey,
k -> {
int leaderEpoch = ShareFetchUtils.leaderEpoch(replicaManager, sharePartitionKey.topicIdPartition().topicPartition());
// Attach listener to Partition which shall invoke partition change handlers.
// However, as there could be multiple share partitions (per group name) for a single topic-partition,
// hence create separate listeners per share partition which holds the share partition key
// to identify the respective share partition.
- SharePartitionListener listener = new SharePartitionListener(sharePartitionKey, replicaManager, partitionCacheMap);
+ SharePartitionListener listener = new SharePartitionListener(sharePartitionKey, replicaManager,
Review Comment:
nit: Probably slightly tidier without the broken line.
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
Review Comment:
nit: Add a blank line above this comment.
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
+ * The partition cache map is used to store the SharePartition objects for each share group topic-partition.
+ */
+ private final Map<SharePartitionKey, SharePartition> partitionMap;
+
+ SharePartitionCache() {
+ this.groupMap = new HashMap<>();
+ this.partitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SharePartition get(SharePartitionKey partitionKey) {
+ return partitionMap.get(partitionKey);
+ }
+
+ public SharePartition put(SharePartitionKey partitionKey, SharePartition sharePartition) {
+ return partitionMap.put(partitionKey, sharePartition);
+ }
+
+ public int size() {
+ return partitionMap.size();
+ }
+
+ public boolean containsKey(SharePartitionKey partitionKey) {
+ return partitionMap.containsKey(partitionKey);
+ }
+
+ public boolean isEmpty() {
+ return partitionMap.isEmpty();
+ }
+
+ public synchronized SharePartition remove(SharePartitionKey partitionKey) {
+ groupMap.computeIfPresent(partitionKey.groupId(), (k, v) -> {
+ v.remove(partitionKey.topicIdPartition());
+ return v;
+ });
+ return partitionMap.remove(partitionKey);
+ }
+
+ public synchronized SharePartition computeIfAbsent(SharePartitionKey partitionKey, Function<SharePartitionKey, SharePartition> mappingFunction) {
+ groupMap.putIfAbsent(partitionKey.groupId(), new HashSet<>());
+ groupMap.get(partitionKey.groupId()).add(partitionKey.topicIdPartition());
+ return partitionMap.computeIfAbsent(partitionKey, mappingFunction);
+ }
+
+ public synchronized void removeGroup(String groupId) {
+ Set<TopicIdPartition> topicIdPartitions = groupMap.remove(groupId);
Review Comment:
I'd add a null check here, just in case.
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
Review Comment:
Instead of "share group topic-partition", I suggest "share-partition". For one thing, it matches the class name of the cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19666:
URL: https://github.com/apache/kafka/pull/19666#discussion_r2086868597
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -143,6 +191,7 @@ public synchronized ShareSessionKey maybeCreateSession(
ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH));
sessions.put(session.key(), session);
updateNumPartitions(session);
+ numMembersPerGroup.merge(session.key().groupId(), 1, Integer::sum);
Review Comment:
Again, I'd prefer `HashMap.compute`.
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -111,6 +158,7 @@ public synchronized ShareSession remove(ShareSession session) {
ShareSession removeResult = sessions.remove(session.key());
if (removeResult != null) {
numPartitions = numPartitions - session.cachedSize();
+ numMembersPerGroup.merge(session.key().groupId(), -1, Integer::sum);
Review Comment:
I would rather see this use `HashMap.compute` and remove the key if the group transitions to zero.
##########
server/src/main/java/org/apache/kafka/server/share/ShareGroupListener.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.Uuid;
+
+/**
+ * The ShareGroupListener is used to notify when there is a change in the share group members.
+ */
+public interface ShareGroupListener {
+
+ /**
+ * Called when member leaves the group.
+ *
+ * @param groupId The id of the group.
+ * @param memberId The id of the member.
+ */
+ void onMemberLeave(String groupId, Uuid memberId);
+
+ /**
+ * Called when the group is empty.
+ *
+ * @param groupId The id of the group.
+ */
+ void onEmpty(String groupId);
Review Comment:
`onGroupEmpty` would be preferable I think.
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -51,15 +52,31 @@ public class ShareSessionCache {
* Metric for the rate of eviction of share sessions.
*/
private final Meter evictionsMeter;
-
- private final int maxEntries;
- private long numPartitions = 0;
+ /**
+ * The listener for connection disconnect events for the client.
+ */
private final ConnectionDisconnectListener connectionDisconnectListener;
-
- // A map of session key to ShareSession.
+ /**
+ * Map of session key to ShareSession.
+ */
private final Map<ShareSessionKey, ShareSession> sessions = new HashMap<>();
-
+ /**
+ * Map of groupId to number of members in the group.
+ */
+ private final Map<String, Integer> numMembersPerGroup = new HashMap<>();
Review Comment:
The handling of this map has the characteristic that entries are never deleted, even when a group is empty for an extended period. I'm not convinced this is a good principle.
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groups;
+
+ /**
+ * The map is used to store the SharePartition objects for each share group topic-partition.
+ */
+ private final Map<SharePartitionKey, SharePartition> partitions;
+
+ SharePartitionCache() {
+ this.groups = new HashMap<>();
+ this.partitions = new ConcurrentHashMap<>();
+ }
+
+ public SharePartition get(SharePartitionKey partitionKey) {
+ return partitions.get(partitionKey);
+ }
+
+ public synchronized SharePartition remove(SharePartitionKey partitionKey) {
+ groups.computeIfPresent(partitionKey.groupId(), (k, v) -> {
+ v.remove(partitionKey.topicIdPartition());
+ return v;
+ });
+ return partitions.remove(partitionKey);
+ }
+
+ public synchronized SharePartition computeIfAbsent(SharePartitionKey partitionKey, Function<SharePartitionKey, SharePartition> mappingFunction) {
+ groups.putIfAbsent(partitionKey.groupId(), new HashSet<>());
Review Comment:
How about `groups.computeIfAbsent(partitionKey.groupId(), new HashSet<>()).add(partitionKey.topicIdPartition());` to avoid an unnecessary lookup in the synchronized block.
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -101,6 +118,36 @@ public synchronized ShareSession remove(ShareSessionKey key) {
return null;
}
+ /**
+ * Maybe remove the session and notify listeners. This is called when the connection is disconnected
+ * for the client. The session may have already been removed by the client as part of final epoch,
+ * hence check if the session is still present in the cache.
+ *
+ * @param key The share session key.
+ */
+ public synchronized void maybeRemoveAndNotifyListeners(ShareSessionKey key) {
Review Comment:
I observe that we are calling code synchronized on the share-partition cache when already in a synchronized method of the share-session cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19116, KAFKA-19258: Handling share group member change events [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19666:
URL: https://github.com/apache/kafka/pull/19666#discussion_r2085658297
##########
core/src/main/java/kafka/server/share/SharePartitionCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.server.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.server.share.SharePartitionKey;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/**
+ * The SharePartitionCache is used to cache the SharePartition objects for each share group topic-partition.
+ * The cache is used to avoid creating new SharePartition instances. The cache is thread-safe.
+ */
+public class SharePartitionCache {
+
+ /**
+ * The map to store the share group id and the set of topic-partitions for that group.
+ */
+ private final Map<String, Set<TopicIdPartition>> groupMap;
+ /**
+ * The partition cache map is used to store the SharePartition objects for each share group topic-partition.
+ */
+ private final Map<SharePartitionKey, SharePartition> partitionMap;
+
+ SharePartitionCache() {
+ this.groupMap = new HashMap<>();
+ this.partitionMap = new ConcurrentHashMap<>();
+ }
+
+ public SharePartition get(SharePartitionKey partitionKey) {
+ return partitionMap.get(partitionKey);
+ }
+
+ public SharePartition put(SharePartitionKey partitionKey, SharePartition sharePartition) {
+ return partitionMap.put(partitionKey, sharePartition);
+ }
+
+ public int size() {
+ return partitionMap.size();
+ }
+
+ public boolean containsKey(SharePartitionKey partitionKey) {
+ return partitionMap.containsKey(partitionKey);
+ }
+
+ public boolean isEmpty() {
+ return partitionMap.isEmpty();
+ }
+
+ public synchronized SharePartition remove(SharePartitionKey partitionKey) {
+ groupMap.computeIfPresent(partitionKey.groupId(), (k, v) -> {
+ v.remove(partitionKey.topicIdPartition());
+ return v;
+ });
+ return partitionMap.remove(partitionKey);
+ }
+
+ public synchronized SharePartition computeIfAbsent(SharePartitionKey partitionKey, Function<SharePartitionKey, SharePartition> mappingFunction) {
+ groupMap.putIfAbsent(partitionKey.groupId(), new HashSet<>());
+ groupMap.get(partitionKey.groupId()).add(partitionKey.topicIdPartition());
+ return partitionMap.computeIfAbsent(partitionKey, mappingFunction);
+ }
+
+ public synchronized void removeGroup(String groupId) {
+ Set<TopicIdPartition> topicIdPartitions = groupMap.remove(groupId);
Review Comment:
Done.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -651,14 +652,15 @@ void processShareFetch(ShareFetch shareFetch) {
}
private SharePartition getOrCreateSharePartition(SharePartitionKey sharePartitionKey) {
- return partitionCacheMap.computeIfAbsent(sharePartitionKey,
+ return partitionCache.computeIfAbsent(sharePartitionKey,
k -> {
int leaderEpoch = ShareFetchUtils.leaderEpoch(replicaManager, sharePartitionKey.topicIdPartition().topicPartition());
// Attach listener to Partition which shall invoke partition change handlers.
// However, as there could be multiple share partitions (per group name) for a single topic-partition,
// hence create separate listeners per share partition which holds the share partition key
// to identify the respective share partition.
- SharePartitionListener listener = new SharePartitionListener(sharePartitionKey, replicaManager, partitionCacheMap);
+ SharePartitionListener listener = new SharePartitionListener(sharePartitionKey, replicaManager,
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org