KAFKA-16894: Exploit share feature [3/N] (#19542)

This PR uses the v1 of the ShareVersion feature to enable share groups
for KIP-932.

Previously, there were two potential configs which could be used -
`group.share.enable=true` and including "share" in
`group.coordinator.rebalance.protocols`. After this PR, the first of
these is retained, but the second is not. Instead, the preferred switch
is the ShareVersion feature.

The `group.share.enable` config is temporarily retained for testing and
situations in which it is inconvenient to set the feature, but it should
really not be necessary, especially when we get to AK 4.2. The aim is to
remove this internal config at that point.

No tests should be setting `group.share.enable` any more, because they
can use the feature (which is enabled in test environments by default
because that's how features work). For tests which need to disable share
groups, they now set the share feature to v0. The majority of the code
changes were related to correct initialisation of the metadata cache in
tests now that a feature is used.

Reviewers: Apoorv Mittal <[email protected]>

Author: AndrewJSchofield

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java

@@ -55,7 +55,6 @@
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.ClusterTestDefaults;
-import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.coordinator.group.GroupConfig;
 import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig;
@@ -113,8 +112,6 @@
 @ClusterTestDefaults(
     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"),
         @ClusterConfigProperty(key = "group.share.partition.max.record.locks", value = "10000"),
         @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
         @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
@@ -123,8 +120,7 @@
         @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "1"),
         @ClusterConfigProperty(key = "transaction.state.log.min.isr", value = "1"),
         @ClusterConfigProperty(key = "transaction.state.log.replication.factor", value = "1")
-    },
-    types = {Type.KRAFT}
+    }
 )
 public class ShareConsumerTest {
     private final ClusterInstance cluster;
@@ -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"),
             @ClusterConfigProperty(key = "group.share.partition.max.record.locks", value = "10000"),
             @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
             @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
@@ -2002,8 +1996,6 @@ public void testShareConsumerAfterCoordinatorMovement() 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"),
             @ClusterConfigProperty(key = "group.share.partition.max.record.locks", value = "10000"),
             @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
             @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),

core/src/main/java/kafka/server/builders/KafkaApisBuilder.java

@@ -69,7 +69,7 @@ public class KafkaApisBuilder {
     private DelegationTokenManager tokenManager = null;
     private ApiVersionManager apiVersionManager = null;
     private ClientMetricsManager clientMetricsManager = null;
-    private Optional<ShareCoordinator> shareCoordinator = Optional.empty();
+    private ShareCoordinator shareCoordinator = null;
     private GroupConfigManager groupConfigManager = null;
 
     public KafkaApisBuilder setRequestChannel(RequestChannel requestChannel) {
@@ -97,7 +97,7 @@ public KafkaApisBuilder setTxnCoordinator(TransactionCoordinator txnCoordinator)
         return this;
     }
 
-    public KafkaApisBuilder setShareCoordinator(Optional<ShareCoordinator> shareCoordinator) {
+    public KafkaApisBuilder setShareCoordinator(ShareCoordinator shareCoordinator) {
         this.shareCoordinator = shareCoordinator;
         return this;
     }
@@ -194,8 +194,8 @@ public KafkaApis build() {
         if (replicaManager == null) throw new RuntimeException("You must set replicaManager");
         if (groupCoordinator == null) throw new RuntimeException("You must set groupCoordinator");
         if (txnCoordinator == null) throw new RuntimeException("You must set txnCoordinator");
-        if (autoTopicCreationManager == null)
-            throw new RuntimeException("You must set autoTopicCreationManager");
+        if (shareCoordinator == null) throw new RuntimeException("You must set shareCoordinator");
+        if (autoTopicCreationManager == null) throw new RuntimeException("You must set autoTopicCreationManager");
         if (config == null) config = new KafkaConfig(Map.of());
         if (configRepository == null) throw new RuntimeException("You must set configRepository");
         if (metadataCache == null) throw new RuntimeException("You must set metadataCache");
@@ -213,7 +213,7 @@ public KafkaApis build() {
                              replicaManager,
                              groupCoordinator,
                              txnCoordinator,
-                             OptionConverters.toScala(shareCoordinator),
+                             shareCoordinator,
                              autoTopicCreationManager,
                              brokerId,
                              config,

core/src/main/scala/kafka/server/AutoTopicCreationManager.scala

@@ -59,7 +59,7 @@ class DefaultAutoTopicCreationManager(
   channelManager: NodeToControllerChannelManager,
   groupCoordinator: GroupCoordinator,
   txnCoordinator: TransactionCoordinator,
-  shareCoordinator: Option[ShareCoordinator]
+  shareCoordinator: ShareCoordinator
 ) extends AutoTopicCreationManager with Logging {
 
   private val inflightTopics = Collections.newSetFromMap(new ConcurrentHashMap[String, java.lang.Boolean]())
@@ -198,15 +198,11 @@ class DefaultAutoTopicCreationManager(
           .setConfigs(convertToTopicConfigCollections(
             txnCoordinator.transactionTopicConfigs))
       case SHARE_GROUP_STATE_TOPIC_NAME =>
-        val props = shareCoordinator match {
-          case Some(coordinator) => coordinator.shareGroupStateTopicConfigs()
-          case None => new Properties()
-        }
         new CreatableTopic()
           .setName(topic)
           .setNumPartitions(config.shareCoordinatorConfig.shareCoordinatorStateTopicNumPartitions())
           .setReplicationFactor(config.shareCoordinatorConfig.shareCoordinatorStateTopicReplicationFactor())
-          .setConfigs(convertToTopicConfigCollections(props))
+          .setConfigs(convertToTopicConfigCollections(shareCoordinator.shareGroupStateTopicConfigs()))
       case topicName =>
         new CreatableTopic()
           .setName(topicName)

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -126,7 +126,7 @@ class BrokerServer(
 
   var transactionCoordinator: TransactionCoordinator = _
 
-  var shareCoordinator: Option[ShareCoordinator] = None
+  var shareCoordinator: ShareCoordinator = _
 
   var clientToControllerChannelManager: NodeToControllerChannelManager = _
 
@@ -629,49 +629,43 @@ class BrokerServer(
       .build()
   }
 
-  private def createShareCoordinator(): Option[ShareCoordinator] = {
-    if (config.shareGroupConfig.isShareGroupEnabled &&
-      config.shareGroupConfig.shareGroupPersisterClassName().nonEmpty) {
-      val time = Time.SYSTEM
-      val timer = new SystemTimerReaper(
-        "share-coordinator-reaper",
-        new SystemTimer("share-coordinator")
-      )
+  private def createShareCoordinator(): ShareCoordinator = {
+    val time = Time.SYSTEM
+    val timer = new SystemTimerReaper(
+      "share-coordinator-reaper",
+      new SystemTimer("share-coordinator")
+    )
 
-      val serde = new ShareCoordinatorRecordSerde
-      val loader = new CoordinatorLoaderImpl[CoordinatorRecord](
-        time,
-        replicaManager,
-        serde,
-        config.shareCoordinatorConfig.shareCoordinatorLoadBufferSize()
-      )
-      val writer = new CoordinatorPartitionWriter(
-        replicaManager
-      )
-      Some(new ShareCoordinatorService.Builder(config.brokerId, config.shareCoordinatorConfig)
-        .withTimer(timer)
-        .withTime(time)
-        .withLoader(loader)
-        .withWriter(writer)
-        .withCoordinatorRuntimeMetrics(new ShareCoordinatorRuntimeMetrics(metrics))
-        .withCoordinatorMetrics(new ShareCoordinatorMetrics(metrics))
-        .build())
-    } else {
-      None
-    }
+    val serde = new ShareCoordinatorRecordSerde
+    val loader = new CoordinatorLoaderImpl[CoordinatorRecord](
+      time,
+      replicaManager,
+      serde,
+      config.shareCoordinatorConfig.shareCoordinatorLoadBufferSize()
+    )
+    val writer = new CoordinatorPartitionWriter(
+      replicaManager
+    )
+    new ShareCoordinatorService.Builder(config.brokerId, config.shareCoordinatorConfig)
+      .withTimer(timer)
+      .withTime(time)
+      .withLoader(loader)
+      .withWriter(writer)
+      .withCoordinatorRuntimeMetrics(new ShareCoordinatorRuntimeMetrics(metrics))
+      .withCoordinatorMetrics(new ShareCoordinatorMetrics(metrics))
+      .build()
   }
 
   private def createShareStatePersister(): Persister = {
-    if (config.shareGroupConfig.isShareGroupEnabled &&
-      config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
+    if (config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
       val klass = Utils.loadClass(config.shareGroupConfig.shareGroupPersisterClassName, classOf[Object]).asInstanceOf[Class[Persister]]
 
       if (klass.getName.equals(classOf[DefaultStatePersister].getName)) {
         klass.getConstructor(classOf[PersisterStateManager])
           .newInstance(
             new PersisterStateManager(
               NetworkUtils.buildNetworkClient("Persister", config, metrics, Time.SYSTEM, new LogContext(s"[Persister broker=${config.brokerId}]")),
-              new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.get.partitionFor(key), config.interBrokerListenerName),
+              new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName),
               Time.SYSTEM,
               new SystemTimerReaper(
                 "persister-state-manager-reaper",
@@ -782,8 +776,8 @@ class BrokerServer(
         CoreUtils.swallow(groupConfigManager.close(), this)
       if (groupCoordinator != null)
         CoreUtils.swallow(groupCoordinator.shutdown(), this)
-      if (shareCoordinator.isDefined)
-        CoreUtils.swallow(shareCoordinator.get.shutdown(), this)
+      if (shareCoordinator != null)
+        CoreUtils.swallow(shareCoordinator.shutdown(), this)
 
       if (assignmentsManager != null)
         CoreUtils.swallow(assignmentsManager.close(), this)