KAFKA-12374: Add missing config sasl.mechanism.controller.protocol

core/src/main/scala/kafka/raft/RaftManager.scala

@@ -30,10 +30,11 @@ import kafka.utils.{KafkaScheduler, Logging, ShutdownableThread}
 import org.apache.kafka.clients.{ApiVersions, ClientDnsLookup, ManualMetadataUpdater, NetworkClient}
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.metrics.Metrics
-import org.apache.kafka.common.network.{ChannelBuilders, NetworkReceive, Selectable, Selector}
+import org.apache.kafka.common.network.{ChannelBuilders, ListenerName, NetworkReceive, Selectable, Selector}
 import org.apache.kafka.common.protocol.ApiMessage
 import org.apache.kafka.common.requests.RequestHeader
 import org.apache.kafka.common.security.JaasContext
+import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.{LogContext, Time}
 import org.apache.kafka.raft.RaftConfig.{AddressSpec, InetAddressSpec, NON_ROUTABLE_ADDRESS, UnknownAddressSpec}
 import org.apache.kafka.raft.{FileBasedStateStore, KafkaRaftClient, RaftClient, RaftConfig, RaftRequest, RecordSerde}
@@ -241,12 +242,14 @@ class KafkaRaftManager[T](
   }
 
   private def buildNetworkClient(): NetworkClient = {
+    val controllerListenerName = new ListenerName(config.controllerListenerNames.head)
+    val controllerSecurityProtocol = config.listenerSecurityProtocolMap.getOrElse(controllerListenerName, SecurityProtocol.forName(controllerListenerName.value()))
     val channelBuilder = ChannelBuilders.clientChannelBuilder(
-      config.interBrokerSecurityProtocol,
+      controllerSecurityProtocol,
       JaasContext.Type.SERVER,
       config,
-      config.interBrokerListenerName,
-      config.saslMechanismInterBrokerProtocol,
+      controllerListenerName,
+      config.saslMechanismControllerProtocol,
       time,
       config.saslInterBrokerHandshakeRequestEnable,
       logContext

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

@@ -40,6 +40,7 @@ trait ControllerNodeProvider {
   def get(): Option[Node]
   def listenerName: ListenerName
   def securityProtocol: SecurityProtocol
+  def saslMechanism: String
 }
 
 object MetadataCacheControllerNodeProvider {
@@ -56,15 +57,17 @@ object MetadataCacheControllerNodeProvider {
     new MetadataCacheControllerNodeProvider(
       metadataCache,
       listenerName,
-      securityProtocol
+      securityProtocol,
+      config.saslMechanismInterBrokerProtocol
     )
   }
 }
 
 class MetadataCacheControllerNodeProvider(
   val metadataCache: kafka.server.MetadataCache,
   val listenerName: ListenerName,
-  val securityProtocol: SecurityProtocol
+  val securityProtocol: SecurityProtocol,
+  val saslMechanism: String
 ) extends ControllerNodeProvider {
   override def get(): Option[Node] = {
     metadataCache.getControllerId
@@ -78,9 +81,16 @@ object RaftControllerNodeProvider {
             config: KafkaConfig,
             controllerQuorumVoterNodes: Seq[Node]): RaftControllerNodeProvider = {
 
-    val listenerName = new ListenerName(config.controllerListenerNames.head)
-    val securityProtocol = config.listenerSecurityProtocolMap.getOrElse(listenerName, SecurityProtocol.forName(listenerName.value()))
-    new RaftControllerNodeProvider(metaLogManager, controllerQuorumVoterNodes, listenerName, securityProtocol)
+    val controllerListenerName = new ListenerName(config.controllerListenerNames.head)
+    val controllerSecurityProtocol = config.listenerSecurityProtocolMap.getOrElse(controllerListenerName, SecurityProtocol.forName(controllerListenerName.value()))
+    val controllerSaslMechanism = config.saslMechanismControllerProtocol
+    new RaftControllerNodeProvider(
+      metaLogManager,
+      controllerQuorumVoterNodes,
+      controllerListenerName,
+      controllerSecurityProtocol,
+      controllerSaslMechanism
+    )
   }
 }
 
@@ -91,7 +101,8 @@ object RaftControllerNodeProvider {
 class RaftControllerNodeProvider(val metaLogManager: MetaLogManager,
                                  controllerQuorumVoterNodes: Seq[Node],
                                  val listenerName: ListenerName,
-                                 val securityProtocol: SecurityProtocol
+                                 val securityProtocol: SecurityProtocol,
+                                 val saslMechanism: String
                                 ) extends ControllerNodeProvider with Logging {
   val idToNode = controllerQuorumVoterNodes.map(node => node.id() -> node).toMap
 
@@ -179,7 +190,7 @@ class BrokerToControllerChannelManagerImpl(
         JaasContext.Type.SERVER,
         config,
         controllerNodeProvider.listenerName,
-        config.saslMechanismInterBrokerProtocol,
+        controllerNodeProvider.saslMechanism,
         time,
         config.saslInterBrokerHandshakeRequestEnable,
         logContext

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

@@ -378,6 +378,7 @@ object KafkaConfig {
   val NodeIdProp = "node.id"
   val MetadataLogDirProp = "metadata.log.dir"
   val ControllerListenerNamesProp = "controller.listener.names"
+  val SaslMechanismControllerProtocolProp = "sasl.mechanism.controller.protocol"
 
   /************* Authorizer Configuration ***********/
   val AuthorizerClassNameProp = "authorizer.class.name"
@@ -675,6 +676,7 @@ object KafkaConfig {
     "KIP-500. If it is not set, the metadata log is placed in the first log directory from log.dirs."
   val ControllerListenerNamesDoc = "A comma-separated list of the names of the listeners used by the KIP-500 controller. This is required " +
     "if this process is a KIP-500 controller. The ZK-based controller will not use this configuration."
+  val SaslMechanismControllerProtocolDoc = "SASL mechanism used for communication with controllers. Default is GSSAPI."
 
   /************* Authorizer Configuration ***********/
   val AuthorizerClassNameDoc = s"The fully qualified name of a class that implements s${classOf[Authorizer].getName}" +
@@ -1081,6 +1083,7 @@ object KafkaConfig {
       .defineInternal(BrokerSessionTimeoutMsProp, INT, Defaults.BrokerSessionTimeoutMs, null, MEDIUM, BrokerSessionTimeoutMsDoc)
       .defineInternal(MetadataLogDirProp, STRING, null, null, HIGH, MetadataLogDirDoc)
       .defineInternal(ControllerListenerNamesProp, STRING, null, null, HIGH, ControllerListenerNamesDoc)
+      .defineInternal(SaslMechanismControllerProtocolProp, STRING, SaslConfigs.DEFAULT_SASL_MECHANISM, null, HIGH, SaslMechanismControllerProtocolDoc)
 
       /************* Authorizer Configuration ***********/
       .define(AuthorizerClassNameProp, STRING, Defaults.AuthorizerClassName, LOW, AuthorizerClassNameDoc)
@@ -1814,6 +1817,8 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
   def controllerListeners: Seq[EndPoint] =
     listeners.filter(l => controllerListenerNames.contains(l.listenerName.value()))
 
+  def saslMechanismControllerProtocol = getString(KafkaConfig.SaslMechanismControllerProtocolProp)
+
   def controlPlaneListener: Option[EndPoint] = {
     controlPlaneListenerName.map { listenerName =>
       listeners.filter(endpoint => endpoint.listenerName.value() == listenerName.value()).head

core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala

@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
 
 import kafka.utils.{MockTime, TestUtils}
 import org.apache.kafka.clients.{Metadata, MockClient, NodeApiVersions}
+import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.{Node, Uuid}
 import org.apache.kafka.common.internals.ClusterResourceListeners
 import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion
@@ -58,6 +59,8 @@ class BrokerLifecycleManagerTest {
     override def listenerName: ListenerName = new ListenerName("PLAINTEXT")
 
     override def securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT;
+
+    override def saslMechanism: String = SaslConfigs.DEFAULT_SASL_MECHANISM
   }
 
   class BrokerLifecycleManagerTestContext(properties: Properties) {

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -779,6 +779,7 @@ class KafkaConfigTest {
         case KafkaConfig.SslPrincipalMappingRulesProp => // ignore string
 
         //Sasl Configs
+        case KafkaConfig.SaslMechanismControllerProtocolProp => // ignore
         case KafkaConfig.SaslMechanismInterBrokerProtocolProp => // ignore
         case KafkaConfig.SaslEnabledMechanismsProp =>
         case KafkaConfig.SaslClientCallbackHandlerClassProp =>

tests/kafkatest/sanity_checks/test_verifiable_producer.py

@@ -96,4 +96,77 @@ def test_simple_run(self, producer_version, security_protocol = 'PLAINTEXT', sas
         num_produced = self.producer.num_acked
         assert num_produced == self.num_messages, "num_produced: %d, num_messages: %d" % (num_produced, self.num_messages)
 
+    @cluster(num_nodes=4)
+    @matrix(inter_broker_security_protocol=['PLAINTEXT', 'SSL'], metadata_quorum=[quorum.remote_raft])
+    @matrix(inter_broker_security_protocol=['SASL_SSL'], inter_broker_sasl_mechanism=['PLAIN', 'GSSAPI'],
+            metadata_quorum=[quorum.remote_raft])
+    def test_multiple_raft_security_protocols(
+            self, inter_broker_security_protocol, inter_broker_sasl_mechanism='GSSAPI', metadata_quorum=quorum.remote_raft):
+        """
+        Test for remote Raft cases that we can start VerifiableProducer on the current branch snapshot version, and
+        verify that we can produce a small number of messages.  The inter-controller and broker-to-controller
+        security protocols are defined to be different (which differs from the above test, where they were the same).
+        """
+        self.kafka.security_protocol = self.kafka.interbroker_security_protocol = inter_broker_security_protocol
+        self.kafka.client_sasl_mechanism = self.kafka.interbroker_sasl_mechanism = inter_broker_sasl_mechanism
+        controller_quorum = self.kafka.controller_quorum
+        sasl_mechanism = 'PLAIN' if inter_broker_sasl_mechanism == 'GSSAPI' else 'GSSAPI'
+        if inter_broker_security_protocol == 'PLAINTEXT':
+            controller_security_protocol = 'SSL'
+            intercontroller_security_protocol = 'SASL_SSL'
+        elif inter_broker_security_protocol == 'SSL':
+            controller_security_protocol = 'SASL_SSL'
+            intercontroller_security_protocol = 'PLAINTEXT'
+        else: # inter_broker_security_protocol == 'SASL_SSL'
+            controller_security_protocol = 'PLAINTEXT'
+            intercontroller_security_protocol = 'SSL'
+        controller_quorum.controller_security_protocol = controller_security_protocol
+        controller_quorum.controller_sasl_mechanism = sasl_mechanism
+        controller_quorum.intercontroller_security_protocol = intercontroller_security_protocol
+        controller_quorum.intercontroller_sasl_mechanism = sasl_mechanism
+        self.kafka.start()
+
+        node = self.producer.nodes[0]
+        node.version = KafkaVersion(str(DEV_BRANCH))
+        self.producer.start()
+        wait_until(lambda: self.producer.num_acked > 5, timeout_sec=15,
+             err_msg="Producer failed to start in a reasonable amount of time.")
+
+        # See above comment above regarding use of version.vstring (distutils.version.LooseVersion)
+        assert is_version(node, [node.version.vstring], logger=self.logger)
+
+        self.producer.wait()
+        num_produced = self.producer.num_acked
+        assert num_produced == self.num_messages, "num_produced: %d, num_messages: %d" % (num_produced, self.num_messages)
+
+    @cluster(num_nodes=4)
+    @parametrize(metadata_quorum=quorum.remote_raft)
+    def test_multiple_raft_sasl_mechanisms(self, metadata_quorum):
+        """
+        Test for remote Raft cases that we can start VerifiableProducer on the current branch snapshot version, and
+        verify that we can produce a small number of messages.  The inter-controller and broker-to-controller
+        security protocols are both SASL_PLAINTEXT but the SASL mechanisms are different (we set
+        GSSAPI for the inter-controller mechanism and PLAIN for the broker-to-controller mechanism).
+        This test differs from the above tests -- he ones above used the same SASL mechanism for both paths.
+        """
+        self.kafka.security_protocol = self.kafka.interbroker_security_protocol = 'PLAINTEXT'
+        controller_quorum = self.kafka.controller_quorum
+        controller_quorum.controller_security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.controller_sasl_mechanism = 'PLAIN'
+        controller_quorum.intercontroller_security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.intercontroller_sasl_mechanism = 'GSSAPI'
+        self.kafka.start()
+
+        node = self.producer.nodes[0]
+        node.version = KafkaVersion(str(DEV_BRANCH))
+        self.producer.start()
+        wait_until(lambda: self.producer.num_acked > 5, timeout_sec=15,
+             err_msg="Producer failed to start in a reasonable amount of time.")
+
+        # See above comment above regarding use of version.vstring (distutils.version.LooseVersion)
+        assert is_version(node, [node.version.vstring], logger=self.logger)
+
+        self.producer.wait()
+        num_produced = self.producer.num_acked
+        assert num_produced == self.num_messages, "num_produced: %d, num_messages: %d" % (num_produced, self.num_messages)