commit a2827db501afc98d420ada50cf7a914436b8dd66
author Ziting <9065044+ZitingShen@users.noreply.github.com> Fri Mar 26 04:39:50 2021 -0700
committer GitHub <noreply@github.com> Fri Mar 26 04:39:50 2021 -0700
tree 2e842f21b9b6c5180b31e29c0fcf9e62bf27febe
parent 29deb426b3adf1d73e6c5e74330280efbebc974e

Fix NPE when thread pool size is 0 and thread pool is null (#1481)

Problem: when thread pool size is 0 and thread pool is null, hostStatisticsMonitor will throw NullPointerException when calculating containerActiveThreads and containerThreadPoolSize metrics. This will subsequently lead to the physical memory metrics to be missing as well, which can trigger alerts, lead to undesirable autosizing, etc.

Solution: add a null check for the thread pool before calculating containerActiveThreads and containerThreadPoolSize metrics

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 2e87f48..83f5c9d 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -624,12 +624,17 @@
         val physicalMemoryBytes : Long = sample.getPhysicalMemoryBytes
         val physicalMemoryMb : Float = physicalMemoryBytes / (1024.0F * 1024.0F)
         val memoryUtilization : Float = physicalMemoryMb.toFloat / containerMemoryMb
-        val containerThreadPoolSize : Long = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getPoolSize
-        val containerActiveThreads : Long = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getActiveCount
         logger.debug("Container physical memory utilization (mb): " + physicalMemoryMb)
         logger.debug("Container physical memory utilization: " + memoryUtilization)
         samzaContainerMetrics.physicalMemoryMb.set(physicalMemoryMb)
         samzaContainerMetrics.physicalMemoryUtilization.set(memoryUtilization)
+
+        var containerThreadPoolSize : Long = 0
+        var containerActiveThreads : Long = 0
+        if (taskThreadPool != null) {
+          containerThreadPoolSize = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getPoolSize
+          containerActiveThreads = taskThreadPool.asInstanceOf[ThreadPoolExecutor].getActiveCount
+        }
         samzaContainerMetrics.containerThreadPoolSize.set(containerThreadPoolSize)
         samzaContainerMetrics.containerActiveThreads.set(containerActiveThreads)
       }