AMBARI-13803. Reconfiguring Kafka service via ambari-web generates an error in stack advisor. (jaimin)

ambari-server/src/main/resources/stacks/HDP/2.3/services/stack_advisor.py

@@ -323,7 +323,11 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
       putHdfsSitePropertyAttribute('dfs.namenode.inode.attributes.provider.class', 'delete', 'true')
 
   def recommendKAFKAConfigurations(self, configurations, clusterData, services, hosts):
-    core_site = services["configurations"]["core-site"]["properties"]
+    kafka_broker = getServicesSiteProperties(services, "kafka-broker")
+
+    # kerberos security for kafka is decided from `security.inter.broker.protocol` property value
+    security_enabled = (kafka_broker is not None and 'security.inter.broker.protocol' in  kafka_broker
+                        and 'SASL' in kafka_broker['security.inter.broker.protocol'])
     putKafkaBrokerProperty = self.putProperty(configurations, "kafka-broker", services)
     putKafkaLog4jProperty = self.putProperty(configurations, "kafka-log4j", services)
     putKafkaBrokerAttributes = self.putPropertyAttribute(configurations, "kafka-broker")
@@ -385,20 +389,16 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
 
 
       else:
-      # Cluster is kerberized
-        if 'hadoop.security.authentication' in core_site and core_site['hadoop.security.authentication'] == 'kerberos' and \
+        # Kerberized Cluster with Ranger plugin disabled
+        if security_enabled and \
           services['configurations']['kafka-broker']['properties']['authorizer.class.name'] == 'org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer':
           putKafkaBrokerProperty("authorizer.class.name", 'kafka.security.auth.SimpleAclAuthorizer')
+        # Non-kerberos Cluster with Ranger plugin disabled
         else:
           putKafkaBrokerAttributes('authorizer.class.name', 'delete', 'true')
-      # Cluster with Ranger is not kerberized
-    elif ('hadoop.security.authentication' not in core_site or core_site['hadoop.security.authentication'] != 'kerberos'):
-      putKafkaBrokerAttributes('authorizer.class.name', 'delete', 'true')
-
-
 
-    # Cluster without Ranger is not kerberized
-    elif ('hadoop.security.authentication' not in core_site or core_site['hadoop.security.authentication'] != 'kerberos'):
+    # Non-Kerberos Cluster without Ranger
+    elif not security_enabled:
       putKafkaBrokerAttributes('authorizer.class.name', 'delete', 'true')
 
 

ambari-server/src/test/python/stacks/2.3/common/test_stack_advisor.py

@@ -430,7 +430,7 @@ class TestHDP23StackAdvisor(TestCase):
     # Test authorizer.class.name with Ranger Kafka plugin disabled in kerberos environment
     configurations['kafka-broker']['properties'] = {}
     configurations['kafka-broker']['property_attributes'] = {}
-    services['configurations']['core-site']['properties']['hadoop.security.authentication'] = 'kerberos'
+    services['configurations']['kafka-broker']['properties']['security.inter.broker.protocol'] = 'PLAINTEXTSASL'
     services['configurations']['kafka-broker']['properties']['authorizer.class.name'] = 'org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer'
     self.stackAdvisor.recommendKAFKAConfigurations(configurations, clusterData, services, None)
     self.assertEquals(configurations['kafka-broker']['properties']['authorizer.class.name'], 'kafka.security.auth.SimpleAclAuthorizer' , "Test authorizer.class.name with Ranger Kafka plugin disabled in kerberos environment")
@@ -438,7 +438,7 @@ class TestHDP23StackAdvisor(TestCase):
     # Test authorizer.class.name with Ranger Kafka plugin enabled in non-kerberos environment
     configurations['kafka-broker']['properties'] = {}
     configurations['kafka-broker']['property_attributes'] = {}
-    del services['configurations']['core-site']['properties']['hadoop.security.authentication']
+    del services['configurations']['kafka-broker']['properties']['security.inter.broker.protocol']
     services['configurations']['kafka-broker']['properties']['authorizer.class.name'] = 'kafka.security.auth.SimpleAclAuthorizer'
     services['configurations']['ranger-kafka-plugin-properties']['properties']['ranger-kafka-plugin-enabled'] = 'Yes'
     self.stackAdvisor.recommendKAFKAConfigurations(configurations, clusterData, services, None)
@@ -447,7 +447,7 @@ class TestHDP23StackAdvisor(TestCase):
     # Test authorizer.class.name with Ranger Kafka plugin enabled in kerberos environment
     configurations['kafka-broker']['properties'] = {}
     configurations['kafka-broker']['property_attributes'] = {}
-    services['configurations']['core-site']['properties']['hadoop.security.authentication'] = 'kerberos'
+    services['configurations']['kafka-broker']['properties']['security.inter.broker.protocol'] = 'PLAINTEXTSASL'
     services['configurations']['kafka-broker']['properties']['authorizer.class.name'] = 'kafka.security.auth.SimpleAclAuthorizer'
     services['configurations']['ranger-kafka-plugin-properties']['properties']['ranger-kafka-plugin-enabled'] = 'Yes'
     self.stackAdvisor.recommendKAFKAConfigurations(configurations, clusterData, services, None)