apoorvmittal10 commented on code in PR #19542:
URL: https://github.com/apache/kafka/pull/19542#discussion_r2064023410
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -1851,8 +1847,6 @@ public void
testShareAutoOffsetResetByDurationInvalidFormat() throws Exception {
brokers = 3,
serverProperties = {
@ClusterConfigProperty(key = "auto.create.topics.enable", value =
"false"),
- @ClusterConfigProperty(key =
"group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
- @ClusterConfigProperty(key = "group.share.enable", value = "true"),
Review Comment:
`testComplexShareConsumer` in this file also has both properties specifed,
shall we remove them as well?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3940,8 +3905,12 @@ class KafkaApis(val requestChannel: RequestChannel,
.setCurrentLeader(partitionData.currentLeader)
}
+ private def shareVersion(): ShareVersion = {
+
ShareVersion.fromFeatureLevel(metadataCache.features.finalizedFeatures.getOrDefault(ShareVersion.FEATURE_NAME,
0.toShort))
+ }
+
private def isShareGroupProtocolEnabled: Boolean = {
- config.shareGroupConfig.isShareGroupEnabled
+ config.shareGroupConfig.isShareGroupEnabled ||
shareVersion().supportsShareGroups
Review Comment:
Query: So we have removed the config `isShareGroupEnabled` usage from
BokerServer and tests but still uses in KafkaApis, why?
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -629,49 +629,43 @@ class BrokerServer(
.build()
}
- private def createShareCoordinator(): Option[ShareCoordinator] = {
- if (config.shareGroupConfig.isShareGroupEnabled &&
- config.shareGroupConfig.shareGroupPersisterClassName().nonEmpty) {
Review Comment:
So irrespective of feature flag or config the share-coordinator thread will
run now. I think this is what you mentioned to be fixed in further PRs to start
using feature listeners, correct?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -10935,7 +11343,20 @@ class KafkaApisTest extends Logging {
@Test
def testShareGroupHeartbeatRequestTopicAuthorizationFailed(): Unit = {
- metadataCache = new KRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_0)
+ metadataCache = {
+ val cache = new KRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_1)
+ val delta = new MetadataDelta(MetadataImage.EMPTY);
+ delta.replay(new FeatureLevelRecord()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setFeatureLevel(MetadataVersion.MINIMUM_VERSION.featureLevel())
+ )
+ delta.replay(new FeatureLevelRecord()
+ .setName(ShareVersion.FEATURE_NAME)
+ .setFeatureLevel(ShareVersion.SV_1.featureLevel())
+ )
+ cache.setImage(delta.apply(MetadataProvenance.EMPTY))
Review Comment:
nit: Seems repeating at a lot of instance, should we have a method to enable
share groups which can be called in all the 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
