adixitconfluent commented on code in PR #19659:
URL: https://github.com/apache/kafka/pull/19659#discussion_r2086477440
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -455,9 +468,12 @@ public ShareFetchContext newContext(
ImplicitLinkedHashCollection<>(shareFetchData.size());
shareFetchData.forEach(topicIdPartition ->
cachedSharePartitions.mustAdd(new
CachedSharePartition(topicIdPartition, false)));
- ShareSessionKey responseShareSessionKey =
cache.maybeCreateSession(groupId, reqMetadata.memberId(),
- cachedSharePartitions, clientConnectionId);
- if (responseShareSessionKey == null) {
+ ShareSessionKey responseShareSessionKey = null;
+ if (isShareGroupsSupported.get()) {
+ responseShareSessionKey =
cache.maybeCreateSession(groupId, reqMetadata.memberId(),
+ cachedSharePartitions, clientConnectionId);
+ }
+ if (responseShareSessionKey == null &&
isShareGroupsSupported.get()) {
Review Comment:
I have removed the condition `isShareGroupsSupported.get()`
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -144,6 +146,11 @@ public class SharePartitionManager implements
AutoCloseable {
*/
private final BrokerTopicStats brokerTopicStats;
+ /**
+ * Flag indicating if share groups have been turned on.
+ */
+ private final AtomicBoolean isShareGroupsSupported;
Review Comment:
agreed, I have moved it to `ShareSessionCache`
--
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
