[samza] branch master updated: Fix NPE when thread pool size is 0 and thread pool is null (#1481)

[bharathkk]

This is an automated email from the ASF dual-hosted git repository.

bharathkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git


The following commit(s) were added to refs/heads/master by this push:
     new a2827db  Fix NPE when thread pool size is 0 and thread pool is null 
(#1481)
a2827db is described below

commit a2827db501afc98d420ada50cf7a914436b8dd66
Author: Ziting <9065044+zitings...@users.noreply.github.com>
AuthorDate: Fri Mar 26 04:39:50 2021 -0700

    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
---
 .../main/scala/org/apache/samza/container/SamzaContainer.scala   | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

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 @@ object SamzaContainer extends Logging {
         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)
       }