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 broker-to-raft-controller communication. 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,59 @@ 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.")
+
+        # using version.vstring (distutils.version.LooseVersion) is a tricky way of ensuring
+        # that this check works with DEV_BRANCH
+        # When running VerifiableProducer 0.8.X, both the current branch version and 0.8.X should show up because of the
+        # way verifiable producer pulls in some development directories into its classpath
+        #
+        # If the test fails here because 'ps .. | grep' couldn't find the process it means
+        # the login and grep that is_version() performs is slower than
+        # the time it takes the producer to produce its messages.
+        # Easy fix is to decrease throughput= above, the good fix is to make the producer
+        # not terminate until explicitly killed in this case.
+        if node.version <= LATEST_0_8_2:
+            assert is_version(node, [node.version.vstring, DEV_BRANCH.vstring], logger=self.logger)
+        else:
+            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)
+
 

tests/kafkatest/services/kafka/kafka.py

@@ -445,12 +445,22 @@ def security_config(self):
         if not self._security_config:
             client_sasl_mechanism_to_use = self.client_sasl_mechanism if self.quorum_info.using_zk or self.quorum_info.has_brokers else self.controller_sasl_mechanism
             interbroker_sasl_mechanism_to_use = self.interbroker_sasl_mechanism if self.quorum_info.using_zk or self.quorum_info.has_brokers else self.intercontroller_sasl_mechanism
+            if not self.quorum_info.using_raft:
+                raft_sasl_mechanisms = []
+                raft_tls = False
+            else:
+                if self.controller_quorum.controller_security_protocol in [SecurityConfig.SASL_PLAINTEXT, SecurityConfig.SASL_SSL]:
+                    raft_sasl_mechanisms = [self.controller_quorum.controller_sasl_mechanism]
+                else:
+                    raft_sasl_mechanisms = []
+                raft_tls = self.controller_quorum.controller_security_protocol in [SecurityConfig.SSL, SecurityConfig.SASL_SSL]
             self._security_config = SecurityConfig(self.context, self.security_protocol, self.interbroker_security_protocol,
                                                    zk_sasl=self.zk.zk_sasl if self.quorum_info.using_zk else False, zk_tls=self.zk_client_secure,
                                                    client_sasl_mechanism=client_sasl_mechanism_to_use,
                                                    interbroker_sasl_mechanism=interbroker_sasl_mechanism_to_use,
                                                    listener_security_config=self.listener_security_config,
-                                                   tls_version=self.tls_version)
+                                                   tls_version=self.tls_version,
+                                                   raft_sasl_mechanisms=raft_sasl_mechanisms, raft_tls=raft_tls)
         for port in self.port_mappings.values():
             if port.open:
                 self._security_config.enable_security_protocol(port.security_protocol)
@@ -498,6 +508,9 @@ def start(self, add_principals=""):
         if self.quorum_info.has_brokers_and_controllers and (
                 self.controller_security_protocol != self.intercontroller_security_protocol or
                 self.controller_sasl_mechanism != self.intercontroller_sasl_mechanism):
+            # This is not supported because both the broker and the controller take the first entry from
+            # controller.listener.names and the value from sasl.mechanism.controller.protocol;
+            # they share a single config, so they must both see/use identical values.
             raise Exception("Co-located Raft-based Brokers (%s/%s) and Controllers (%s/%s) cannot talk to Controllers via different security protocols" %
                             (self.controller_security_protocol, self.controller_sasl_mechanism,
                              self.intercontroller_security_protocol, self.intercontroller_sasl_mechanism))
@@ -666,6 +679,9 @@ def start_node(self, node, timeout_sec=60):
                                                       for node in self.controller_quorum.nodes[:self.controller_quorum.num_nodes_controller_role]])
             # define controller.listener.names
             self.controller_listener_names = ','.join(self.controller_listener_name_list())
+            # define sasl.mechanism.controller.protocol to match remote quorum if one exists
+            if self.remote_controller_quorum:
+                self.controller_sasl_mechanism = self.remote_controller_quorum.controller_sasl_mechanism
 
         prop_file = self.prop_file(node)
         self.logger.info("kafka.properties:")

tests/kafkatest/services/kafka/templates/kafka.properties

@@ -95,6 +95,9 @@ zookeeper.ssl.truststore.password=test-ts-passwd
 {% if quorum_info.using_zk or quorum_info.has_brokers %}
 sasl.mechanism.inter.broker.protocol={{ security_config.interbroker_sasl_mechanism }}
 {% endif %}
+{% if quorum_info.using_raft %}
+sasl.mechanism.controller.protocol={{ controller_sasl_mechanism }}
+{% endif %}
 sasl.enabled.mechanisms={{ ",".join(security_config.enabled_sasl_mechanisms) }}
 sasl.kerberos.service.name=kafka
 {% if authorizer_class_name is not none %}

tests/kafkatest/services/security/security_config.py

@@ -145,7 +145,8 @@ class SecurityConfig(TemplateRenderer):
     def __init__(self, context, security_protocol=None, interbroker_security_protocol=None,
                  client_sasl_mechanism=SASL_MECHANISM_GSSAPI, interbroker_sasl_mechanism=SASL_MECHANISM_GSSAPI,
                  zk_sasl=False, zk_tls=False, template_props="", static_jaas_conf=True, jaas_override_variables=None,
-                 listener_security_config=ListenerSecurityConfig(), tls_version=None):
+                 listener_security_config=ListenerSecurityConfig(), tls_version=None,
+                 raft_sasl_mechanisms=[], raft_tls=False):
         """
         Initialize the security properties for the node and copy
         keystore and truststore to the remote node if the transport protocol 
@@ -173,8 +174,9 @@ def __init__(self, context, security_protocol=None, interbroker_security_protoco
         if interbroker_security_protocol is None:
             interbroker_security_protocol = security_protocol
         self.interbroker_security_protocol = interbroker_security_protocol
-        self.has_sasl = self.is_sasl(security_protocol) or self.is_sasl(interbroker_security_protocol) or zk_sasl
-        self.has_ssl = self.is_ssl(security_protocol) or self.is_ssl(interbroker_security_protocol) or zk_tls
+        self.raft_sasl_mechanisms = raft_sasl_mechanisms
+        self.has_sasl = self.is_sasl(security_protocol) or self.is_sasl(interbroker_security_protocol) or zk_sasl or raft_sasl_mechanisms
+        self.has_ssl = self.is_ssl(security_protocol) or self.is_ssl(interbroker_security_protocol) or zk_tls or raft_tls
         self.zk_sasl = zk_sasl
         self.zk_tls = zk_tls
         self.static_jaas_conf = static_jaas_conf
@@ -250,14 +252,16 @@ def setup_sasl(self, node):
                     'is_ibm_jdk': any('IBM' in line for line in java_version),
                     'SecurityConfig': SecurityConfig,
                     'client_sasl_mechanism': self.client_sasl_mechanism,
-                    'enabled_sasl_mechanisms': self.enabled_sasl_mechanisms
+                    'enabled_sasl_mechanisms': self.enabled_sasl_mechanisms,
+                    'usable_sasl_mechanisms': self.usable_sasl_mechanisms
                 }
             )
         else:
             jaas_conf = self.properties['sasl.jaas.config']
 
         if self.static_jaas_conf:
             node.account.create_file(SecurityConfig.JAAS_CONF_PATH, jaas_conf)
+            print(jaas_conf)
             node.account.create_file(SecurityConfig.ADMIN_CLIENT_AS_BROKER_JAAS_CONF_PATH,
                                      self.render_jaas_config(
                                          "admin_client_as_broker_jaas.conf",
@@ -343,6 +347,17 @@ def interbroker_sasl_mechanism(self):
     def enabled_sasl_mechanisms(self):
         return set([self.client_sasl_mechanism, self.interbroker_sasl_mechanism])
 
+    @property
+    def usable_sasl_mechanisms(self):
+        """
+        Differs from enabled_sasl_mechanisms in that it also includes SASL mechanisms that
+        are used for Raft communication
+        :return: enabled_sasl_mechanisms plus any mechanisms that are used for Raft communication
+        """
+        retval = set([self.client_sasl_mechanism, self.interbroker_sasl_mechanism] + self.raft_sasl_mechanisms)
+        print(retval)
+        return retval
+
     @property
     def has_sasl_kerberos(self):
         return self.has_sasl and (SecurityConfig.SASL_MECHANISM_GSSAPI in self.enabled_sasl_mechanisms)