AMBARI-13642. Warn if Xmn size is oversubscribed in AMS HBase master and region server config, Kafka fails to start unless AMS is installed. (Aravindan Vijayan via swagle)

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/params.py

@@ -94,7 +94,7 @@ hbase_log_dir = config['configurations']['ams-hbase-env']['hbase_log_dir']
 master_heapsize = config['configurations']['ams-hbase-env']['hbase_master_heapsize']
 regionserver_heapsize = config['configurations']['ams-hbase-env']['hbase_regionserver_heapsize']
 
-regionserver_xmn_max = default('configurations/ams-hbase-env/hbase_regionserver_xmn_max', None)
+regionserver_xmn_max = default('/configurations/ams-hbase-env/hbase_regionserver_xmn_max', None)
 if regionserver_xmn_max:
   regionserver_xmn_percent = config['configurations']['ams-hbase-env']['hbase_regionserver_xmn_ratio']
   regionserver_xmn_size = calc_xmn_from_xms(regionserver_heapsize, regionserver_xmn_percent, regionserver_xmn_max)

ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml

@@ -277,10 +277,13 @@
   </property>
   <property>
     <name>kafka.metrics.reporters</name>
-    <value>org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter</value>
+    <value></value>
     <description>
       kafka ganglia metrics reporter and kafka timeline metrics reporter
     </description>
+    <value-attributes>
+      <empty-value-valid>true</empty-value-valid>
+    </value-attributes>
   </property>
   <property>
     <name>kafka.ganglia.metrics.reporter.enabled</name>

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

@@ -20,7 +20,7 @@ limitations under the License.
 import re
 import os
 import sys
-from math import ceil
+from math import ceil, floor
 
 from stack_advisor import DefaultStackAdvisor
 
@@ -480,27 +480,24 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
         putAmsHbaseSiteProperty("hbase.regionserver.global.memstore.lowerLimit", 0.25)
         putAmsHbaseSiteProperty("phoenix.query.maxGlobalMemoryPercentage", 20)
         putTimelineServiceProperty("phoenix.query.maxGlobalMemoryPercentage", 30)
-        hbase_xmn_size = '512'
       elif total_sinks_count >= 500:
         putAmsHbaseSiteProperty("hbase.regionserver.handler.count", 60)
         putAmsHbaseSiteProperty("hbase.regionserver.hlog.blocksize", 134217728)
         putAmsHbaseSiteProperty("hbase.regionserver.maxlogs", 64)
         putAmsHbaseSiteProperty("hbase.hregion.memstore.flush.size", 268435456)
-        hbase_xmn_size = '512'
-      elif total_sinks_count >= 250:
-        hbase_xmn_size = '256'
-      else:
-        hbase_xmn_size = '128'
       pass
 
-    # Embedded mode heap size : master + regionserver
+    # Distributed mode heap size
     if rootDir.startswith("hdfs://"):
       putHbaseEnvProperty("hbase_master_heapsize", "512")
+      putHbaseEnvProperty("hbase_master_xmn_size", "102") #20% of 512 heap size
       putHbaseEnvProperty("hbase_regionserver_heapsize", hbase_heapsize)
-      putHbaseEnvProperty("regionserver_xmn_size", hbase_xmn_size)
+      putHbaseEnvProperty("regionserver_xmn_size", round_to_n(0.15*hbase_heapsize,64))
     else:
+      # Embedded mode heap size : master + regionserver
+      hbase_rs_heapsize = 512
       putHbaseEnvProperty("hbase_master_heapsize", hbase_heapsize)
-      putHbaseEnvProperty("hbase_master_xmn_size", hbase_xmn_size)
+      putHbaseEnvProperty("hbase_master_xmn_size", round_to_n(0.15*(hbase_heapsize+hbase_rs_heapsize),64))
 
     # If no local DN in distributed mode
     if rootDir.startswith("hdfs://"):
@@ -876,6 +873,49 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
     amsHbaseSite = getSiteProperties(configurations, "ams-hbase-site")
     logDirItem = self.validatorEqualsPropertyItem(properties, "hbase_log_dir",
                                                   ams_env, "metrics_collector_log_dir")
+
+    # Validate Xmn settings.
+    hbase_master_heapsize = to_number(properties["hbase_master_heapsize"])
+    hbase_master_xmn_size = to_number(properties["hbase_master_xmn_size"])
+    hbase_regionserver_heapsize = to_number(properties["hbase_regionserver_heapsize"])
+    hbase_regionserver_xmn_size = to_number(properties["regionserver_xmn_size"])
+
+    masterXmnItem = None
+    regionServerXmnItem = None
+    hbase_rootdir = amsHbaseSite.get("hbase.rootdir")
+    is_hbase_distributed = hbase_rootdir.startswith('hdfs://')
+
+    if is_hbase_distributed:
+      minMasterXmn = 0.12 *  hbase_master_heapsize
+      maxMasterXmn = 0.2 *  hbase_master_heapsize
+      if hbase_master_xmn_size < minMasterXmn:
+        masterXmnItem = self.getWarnItem("Value is lesser than the recommended minimum Xmn size of {0} "
+                                         "(12% of hbase_master_heapsize)".format(int(math.ceil(minMasterXmn))))
+
+      if hbase_master_xmn_size > maxMasterXmn:
+        masterXmnItem = self.getWarnItem("Value is greater than the recommended maximum Xmn size of {0} "
+                                         "(20% of hbase_master_heapsize)".format(int(math.floor(maxMasterXmn))))
+
+      minRegionServerXmn = 0.12 *  hbase_regionserver_heapsize
+      maxRegionServerXmn = 0.2 *  hbase_regionserver_heapsize
+      if hbase_regionserver_xmn_size < minRegionServerXmn:
+        regionServerXmnItem = self.getWarnItem("Value is lesser than the recommended minimum Xmn size of {0} "
+                              "(12% of hbase_regionserver_heapsize)".format(int(math.ceil(minRegionServerXmn))))
+
+      if hbase_regionserver_xmn_size > maxRegionServerXmn:
+        regionServerXmnItem = self.getWarnItem("Value is greater than the recommended maximum Xmn size of {0} "
+                              "(20% of hbase_regionserver_heapsize)".format(int(math.floor(maxRegionServerXmn))))
+    else:
+      minMasterXmn = 0.12 *  ( hbase_master_heapsize + hbase_regionserver_heapsize )
+      maxMasterXmn = 0.2 *  ( hbase_master_heapsize + hbase_regionserver_heapsize )
+      if hbase_master_xmn_size < minMasterXmn:
+        masterXmnItem = self.getWarnItem("Value is lesser than the recommended minimum Xmn size of {0} "
+                        "(12% of hbase_master_heapsize + hbase_regionserver_heapsize)".format(int(math.ceil(minMasterXmn))))
+
+      if hbase_master_xmn_size > maxMasterXmn:
+        masterXmnItem = self.getWarnItem("Value is greater than the recommended maximum Xmn size of {0} "
+                        "(20% of hbase_master_heapsize + hbase_regionserver_heapsize)".format(int(math.floor(maxMasterXmn))))
+
     validationItems = []
     masterHostItem = None
 
@@ -918,19 +958,13 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
 
           requiredMemory = getMemorySizeRequired(hostComponents, configurations)
           unusedMemory = host["Hosts"]["total_mem"] * 1024 - requiredMemory # in bytes
-          hbase_rootdir = amsHbaseSite.get("hbase.rootdir")
           if unusedMemory > 4294967296:  # warn user, if more than 4GB RAM is unused
             heapPropertyToIncrease = "hbase_regionserver_heapsize" if hbase_rootdir.startswith("hdfs://") else "hbase_master_heapsize"
             xmnPropertyToIncrease = "regionserver_xmn_size" if hbase_rootdir.startswith("hdfs://") else "hbase_master_xmn_size"
             collector_heapsize = int((unusedMemory - 4294967296)/5) + to_number(ams_env.get("metrics_collector_heapsize"))*1048576
             hbase_heapsize = int((unusedMemory - 4294967296)*4/5) + to_number(properties.get(heapPropertyToIncrease))*1048576
-
-            if hbase_heapsize/1048576 > 2048:
-              xmn_size = '512'
-            elif hbase_heapsize/1048576 > 1024:
-              xmn_size = '256'
-            else:
-              xmn_size = '128'
+            hbase_heapsize = min(32*1024*1024*1024, hbase_heapsize) #Make sure heapsize < 32GB
+            xmn_size = round_to_n(0.12*hbase_heapsize,128)
 
             msg = "{0} MB RAM is unused on the host {1} based on components " \
                   "assigned. Consider allocating  {2} MB to " \
@@ -946,7 +980,9 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
       {"config-name": "hbase_regionserver_heapsize", "item": regionServerItem},
       {"config-name": "hbase_master_heapsize", "item": masterItem},
       {"config-name": "hbase_master_heapsize", "item": masterHostItem},
-      {"config-name": "hbase_log_dir", "item": logDirItem}
+      {"config-name": "hbase_log_dir", "item": logDirItem},
+      {"config-name": "hbase_master_xmn_size", "item": masterXmnItem},
+      {"config-name": "regionserver_xmn_size", "item": regionServerXmnItem}
     ])
     return self.toConfigurationValidationProblems(validationItems, "ams-hbase-env")
 

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

@@ -328,6 +328,11 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
     putKafkaLog4jProperty = self.putProperty(configurations, "kafka-log4j", services)
     putKafkaBrokerAttributes = self.putPropertyAttribute(configurations, "kafka-broker")
 
+    #If AMS is part of Services, use the KafkaTimelineMetricsReporter for metric reporting. Default is ''.
+    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
+    if "AMBARI_METRICS" in servicesList:
+      putKafkaBrokerProperty('kafka.metrics.reporters', 'org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter')
+
     if "ranger-env" in services["configurations"] and "ranger-kafka-plugin-properties" in services["configurations"] and \
         "ranger-kafka-plugin-enabled" in services["configurations"]["ranger-env"]["properties"]:
       putKafkaRangerPluginProperty = self.putProperty(configurations, "ranger-kafka-plugin-properties", services)

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

@@ -2048,6 +2048,7 @@ class TestHDP22StackAdvisor(TestCase):
       }
     ]
     expected["ams-hbase-env"]['properties']['hbase_master_heapsize'] = '1408'
+    expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '320'
     expected["ams-env"]['properties']['metrics_collector_heapsize'] = '512'
 
     self.stackAdvisor.recommendAmsConfigurations(configurations, clusterData, services, hosts)
@@ -2115,7 +2116,7 @@ class TestHDP22StackAdvisor(TestCase):
 
     ]
     expected["ams-hbase-env"]['properties']['hbase_master_heapsize'] = '2432'
-    expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '256'
+    expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '448'
     expected["ams-env"]['properties']['metrics_collector_heapsize'] = '640'
 
     self.stackAdvisor.recommendAmsConfigurations(configurations, clusterData, services, hosts)
@@ -2182,7 +2183,8 @@ class TestHDP22StackAdvisor(TestCase):
     expected['ams-site']['properties']['timeline.metrics.host.aggregate.splitpoints'] = ' '
     expected['ams-site']['properties']['timeline.metrics.cluster.aggregate.splitpoints'] = ' '
     expected['ams-hbase-env']['properties']['hbase_regionserver_heapsize'] = '512'
-    expected['ams-hbase-env']['properties']['regionserver_xmn_size'] = '256'
+    expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '102'
+    expected['ams-hbase-env']['properties']['regionserver_xmn_size'] = '384'
     self.stackAdvisor.recommendAmsConfigurations(configurations, clusterData, services, hosts)
     self.assertEquals(configurations, expected)
 
@@ -3582,4 +3584,3 @@ class TestHDP22StackAdvisor(TestCase):
     res_expected = [{'config-type': 'ranger-env', 'message': 'Ranger Storm plugin should not be enabled in non-kerberos environment.', 'type': 'configuration', 'config-name': 'ranger-storm-plugin-enabled', 'level': 'WARN'}]
     res = self.stackAdvisor.validateRangerConfigurationsEnv(properties, recommendedDefaults, configurations, services, {})
     self.assertEquals(res, res_expected)
-

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

@@ -377,6 +377,24 @@ class TestHDP23StackAdvisor(TestCase):
               "service_name" : "KAFKA",
               "service_version" : "2.6.0.2.2"
             }
+          },
+          {
+            "StackServices": {
+              "service_name": "AMBARI_METRICS"
+            },
+            "components": [{
+              "StackServiceComponents": {
+                "component_name": "METRICS_COLLECTOR",
+                "hostnames": ["host1"]
+              }
+
+            }, {
+              "StackServiceComponents": {
+                "component_name": "METRICS_MONITOR",
+                "hostnames": ["host1"]
+              }
+
+            }]
           }
         ],
       "Versions": {
@@ -446,6 +464,10 @@ class TestHDP23StackAdvisor(TestCase):
     expectedLog4jContent = log4jContent + newRangerLog4content
     self.assertEquals(configurations['kafka-log4j']['properties']['content'], expectedLog4jContent, "Test kafka-log4j content when Ranger plugin for Kafka is enabled")
 
+    # Test kafka.metrics.reporters when AMBARI_METRICS is present in services
+    self.stackAdvisor.recommendKAFKAConfigurations(configurations, clusterData, services, None)
+    self.assertEqual(configurations['kafka-broker']['properties']['kafka.metrics.reporters'],
+                                              'org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter')
 
   def test_recommendHBASEConfigurations(self):
     configurations = {}