Misc. minor cleanup.

1. Added a meaningful name for the container thread pool threads.
2. Made the thread names for framework threads consistent.
3. Made a couple of monitoring/metrics threads daemon.
4. Fixed a few checkstyle warning about missing param/throws documentation.

Author: Prateek Maheshwari <pmaheshw@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Jacob M <jmakes@apache.org>

Closes apache#433 from prateekm/container-thread-pool-name

Author: prateekm

samza-core/src/main/java/org/apache/samza/container/disk/PollingScanDiskSpaceMonitor.java

@@ -18,6 +18,8 @@
  */
 package org.apache.samza.container.disk;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,9 +36,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * An implementation of {@link DiskSpaceMonitor} that polls for disk usage based on a specified
@@ -47,7 +47,6 @@
 public class PollingScanDiskSpaceMonitor implements DiskSpaceMonitor {
   private enum State { INIT, RUNNING, STOPPED }
 
-  private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryImpl();
   private static final Logger log = LoggerFactory.getLogger(PollingScanDiskSpaceMonitor.class);
 
   // Note: we use this as a set where the value is always Boolean.TRUE.
@@ -57,7 +56,11 @@ private enum State { INIT, RUNNING, STOPPED }
   private final Object lock = new Object();
 
   private final ScheduledExecutorService schedulerService =
-      Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);
+      Executors.newSingleThreadScheduledExecutor(
+          new ThreadFactoryBuilder()
+              .setNameFormat("Samza PollingScanDiskSpaceMonitor Thread-%d")
+              .setDaemon(true)
+              .build());
   private final Set<Path> watchPaths;
   private final long pollingIntervalMillis;
 
@@ -197,13 +200,4 @@ private void updateSample() {
       }
     }
   }
-
-  private static class ThreadFactoryImpl implements ThreadFactory  {
-    private static final String PREFIX = "Samza-" + PollingScanDiskSpaceMonitor.class.getSimpleName() + "-";
-    private static final AtomicInteger INSTANCE_NUM = new AtomicInteger();
-
-    public Thread newThread(Runnable runnable) {
-      return new Thread(runnable, PREFIX + INSTANCE_NUM.getAndIncrement());
-    }
-  }
 }

samza-core/src/main/java/org/apache/samza/container/host/StatisticsMonitorImpl.java

@@ -18,16 +18,16 @@
  */
 package org.apache.samza.container.host;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * An implementation of {@link SystemStatisticsMonitor} for unix and mac platforms. Users can implement their own
@@ -42,8 +42,6 @@
  * This class is thread-safe.
  */
 public class StatisticsMonitorImpl implements SystemStatisticsMonitor {
-
-  private static final ThreadFactory THREAD_FACTORY = new StatisticsMonitorThreadFactory();
   private static final Logger LOG = LoggerFactory.getLogger(StatisticsMonitorImpl.class);
 
   /**
@@ -60,7 +58,8 @@ public class StatisticsMonitorImpl implements SystemStatisticsMonitor {
 
   // Single threaded executor to handle callback invocations.
   private final ScheduledExecutorService schedulerService =
-      Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);
+      Executors.newSingleThreadScheduledExecutor(
+          new ThreadFactoryBuilder().setNameFormat("Samza StatisticsMonitor Thread-%d").setDaemon(true).build());
 
   // Use this as a set with value always set to True
   private final ConcurrentMap<StatisticsMonitorImpl.Listener, Boolean> listenerSet = new ConcurrentHashMap<>();
@@ -174,15 +173,4 @@ public boolean registerListener(Listener listener) {
       }
     }
   }
-
-  // A convenience class that provides named threads
-  private static class StatisticsMonitorThreadFactory implements ThreadFactory {
-    private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger();
-    private static final String PREFIX = "Samza-StatisticsMonitor-Thread-";
-
-    @Override
-    public Thread newThread(Runnable runnable) {
-      return new Thread(runnable, PREFIX + INSTANCE_COUNT.getAndIncrement());
-    }
-  }
 }

samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamManager.java

@@ -79,7 +79,7 @@ public CoordinatorStreamManager(Config coordinatorSystemConfig, MetricsRegistry
   /**
    * Register source with the coordinator stream.
    *
-   * @param source
+   * @param source source to register with the coordinator stream
    */
   public void register(String source) {
     if (coordinatorStreamConsumer != null) {

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

@@ -26,6 +26,7 @@ import java.util
 import java.util.Base64
 import java.util.concurrent.{ExecutorService, Executors, TimeUnit}
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder
 import org.apache.samza.checkpoint.{CheckpointListener, CheckpointManagerFactory, OffsetManager, OffsetManagerMetrics}
 import org.apache.samza.config.JobConfig.Config2Job
 import org.apache.samza.config.MetricsConfig.Config2Metrics
@@ -405,10 +406,14 @@ object SamzaContainer extends Logging {
     val threadPoolSize = config.getThreadPoolSize
     info("Got thread pool size: " + threadPoolSize)
 
-    val taskThreadPool = if (!singleThreadMode && threadPoolSize > 0)
-      Executors.newFixedThreadPool(threadPoolSize)
-    else
+
+    val taskThreadPool = if (!singleThreadMode && threadPoolSize > 0) {
+      Executors.newFixedThreadPool(threadPoolSize,
+        new ThreadFactoryBuilder().setNameFormat("Samza Container Thread-%d").build())
+    } else {
       null
+    }
+
 
     val finalTaskFactory = TaskFactoryUtil.finalizeTaskFactory(
       taskFactory,

samza-core/src/main/scala/org/apache/samza/metrics/JvmMetrics.scala

@@ -19,23 +19,18 @@
 
 package org.apache.samza.metrics
 
-import scala.collection._
-import scala.collection.JavaConverters._
-import java.lang.management.ManagementFactory
+import com.google.common.util.concurrent.ThreadFactoryBuilder
+import com.sun.management.OperatingSystemMXBean
+import com.sun.management.UnixOperatingSystemMXBean
+import org.apache.samza.util.Logging
+
 import java.lang.Thread.State._
+import java.lang.management.ManagementFactory
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 
-import com.sun.management.{OperatingSystemMXBean, UnixOperatingSystemMXBean}
-import org.apache.samza.util.Logging
-import org.apache.samza.util.DaemonThreadFactory
-
-/**
- *  Companion object for class JvmMetrics encapsulating various constants
- */
-object JvmMetrics {
-  val JVM_METRICS_THREAD_NAME_PREFIX = "JVM-METRICS"
-}
+import scala.collection.JavaConverters._
+import scala.collection._
 
 /**
  * Straight up ripoff of Hadoop's metrics2 JvmMetrics class.
@@ -49,7 +44,8 @@ class JvmMetrics(val registry: MetricsRegistry) extends MetricsHelper with Runna
   val threadMXBean = ManagementFactory.getThreadMXBean()
   val osMXBean = ManagementFactory.getOperatingSystemMXBean()
   var gcBeanCounters = Map[String, (Counter, Counter)]()
-  val executor = Executors.newScheduledThreadPool(1, new DaemonThreadFactory(JvmMetrics.JVM_METRICS_THREAD_NAME_PREFIX))
+  val executor = Executors.newSingleThreadScheduledExecutor(
+    new ThreadFactoryBuilder().setNameFormat("Samza JvmMetrics Thread-%d").setDaemon(true).build())
 
   // jvm metrics
   val gMemNonHeapUsedM = newGauge("mem-non-heap-used-mb", 0.0F)

samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala

@@ -19,30 +19,20 @@
 
 package org.apache.samza.metrics.reporter
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder
+import org.apache.samza.metrics._
+import org.apache.samza.serializers.Serializer
+import org.apache.samza.system.OutgoingMessageEnvelope
+import org.apache.samza.system.SystemProducer
+import org.apache.samza.system.SystemStream
+import org.apache.samza.util.Logging
+
 import java.util.HashMap
 import java.util.Map
-import scala.collection.JavaConverters._
-import org.apache.samza.util.Logging
-import org.apache.samza.metrics.Counter
-import org.apache.samza.metrics.Gauge
-import org.apache.samza.metrics.Timer
-import org.apache.samza.metrics.MetricsReporter
-import org.apache.samza.metrics.MetricsVisitor
-import org.apache.samza.metrics.ReadableMetricsRegistry
 import java.util.concurrent.Executors
-import org.apache.samza.util.DaemonThreadFactory
 import java.util.concurrent.TimeUnit
-import org.apache.samza.serializers.Serializer
-import org.apache.samza.system.SystemProducer
-import org.apache.samza.system.SystemStream
-import org.apache.samza.system.OutgoingMessageEnvelope
 
-/**
- *  Companion object for class MetricsSnapshotReporter encapsulating various constants
- */
-object MetricsSnapshotReporter {
-  val METRIC_SNAPSHOT_REPORTER_THREAD_NAME_PREFIX = "METRIC-SNAPSHOT-REPORTER"
-}
+import scala.collection.JavaConverters._
 
 /**
  * MetricsSnapshotReporter is a generic metrics reporter that sends metrics to a stream.
@@ -66,7 +56,8 @@ class MetricsSnapshotReporter(
   serializer: Serializer[MetricsSnapshot] = null,
   clock: () => Long = () => { System.currentTimeMillis }) extends MetricsReporter with Runnable with Logging {
 
-  val executor = Executors.newScheduledThreadPool(1, new DaemonThreadFactory(MetricsSnapshotReporter.METRIC_SNAPSHOT_REPORTER_THREAD_NAME_PREFIX))
+  val executor = Executors.newSingleThreadScheduledExecutor(
+    new ThreadFactoryBuilder().setNameFormat("Samza MetricsSnapshotReporter Thread-%d").setDaemon(true).build())
   val resetTime = clock()
   var registries = List[(String, ReadableMetricsRegistry)]()
 

samza-core/src/main/scala/org/apache/samza/system/filereader/FileReaderSystemConsumer.scala

@@ -19,24 +19,20 @@
 
 package org.apache.samza.system.filereader
 
-import org.apache.samza.util.BlockingEnvelopeMap
+import com.google.common.util.concurrent.ThreadFactoryBuilder
 import org.apache.samza.metrics.MetricsRegistry
+import org.apache.samza.system.IncomingMessageEnvelope
 import org.apache.samza.system.SystemStreamPartition
-import scala.collection.mutable.Map
+import org.apache.samza.util.BlockingEnvelopeMap
+import org.apache.samza.util.Logging
+
 import java.io.RandomAccessFile
-import org.apache.samza.system.IncomingMessageEnvelope
-import java.util.concurrent.LinkedBlockingQueue
-import java.util.concurrent.Executors
 import java.util.concurrent.ExecutorService
-import org.apache.samza.util.DaemonThreadFactory
-import org.apache.samza.util.Logging
+import java.util.concurrent.Executors
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.collection.mutable.Map
 
-object FileReaderSystemConsumer {
-  /**
-   * prefix for the file reader system thread names
-   */
-  val FILE_READER_SYSTEM_THREAD_PREFIX = "filereader-"
-}
 
 class FileReaderSystemConsumer(
   systemName: String,
@@ -77,8 +73,9 @@ class FileReaderSystemConsumer(
    * start one thread for each file reader
    */
   override def start {
-    pool = Executors.newFixedThreadPool(systemStreamPartitionAndStartingOffset.size, new DaemonThreadFactory(FileReaderSystemConsumer.FILE_READER_SYSTEM_THREAD_PREFIX))
-    systemStreamPartitionAndStartingOffset.map { case (ssp, offset) => pool.execute(readInputFiles(ssp, offset)) }
+    pool = Executors.newFixedThreadPool(systemStreamPartitionAndStartingOffset.size,
+      new ThreadFactoryBuilder().setNameFormat("Samza FileReader Thread-%d").setDaemon(true).build())
+    systemStreamPartitionAndStartingOffset.foreach { case (ssp, offset) => pool.execute(readInputFiles(ssp, offset)) }
   }
 
   /**

samza-core/src/main/scala/org/apache/samza/util/DaemonThreadFactory.scala

@@ -31,19 +31,11 @@ import kafka.consumer.ConsumerConfig
 import kafka.message.MessageSet
 import org.apache.samza.SamzaException
 import org.apache.samza.util.ExponentialSleepStrategy
+import org.apache.samza.util.KafkaUtil
 import org.apache.samza.util.Logging
-import org.apache.samza.util.ThreadNamePrefix.SAMZA_THREAD_NAME_PREFIX
 
 import scala.collection.JavaConverters._
 import scala.collection.concurrent
-import org.apache.samza.util.KafkaUtil
-
-/**
- *  Companion object for class JvmMetrics encapsulating various constants
- */
-object BrokerProxy {
-  val BROKER_PROXY_THREAD_NAME_PREFIX = "BROKER-PROXY-"
-}
 
 /**
  * A BrokerProxy consolidates Kafka fetches meant for a particular broker and retrieves them all at once, providing
@@ -294,7 +286,7 @@ class BrokerProxy(
     if (!thread.isAlive) {
       info("Starting " + toString)
       thread.setDaemon(true)
-      thread.setName(SAMZA_THREAD_NAME_PREFIX + BrokerProxy.BROKER_PROXY_THREAD_NAME_PREFIX + thread.getName)
+      thread.setName("Samza BrokerProxy " + thread.getName)
       thread.setUncaughtExceptionHandler(new UncaughtExceptionHandler {
         override def uncaughtException(t: Thread, e: Throwable) = error("Uncaught exception in broker proxy:", e)
       })