Merge pull request #24646: Add thread prefix for StreamingDataflowWor…

…ker work executor
apache · Jan 6, 2023 · e54871d · e54871d
2 parents 6ad7540 + db1f9d2
commit e54871d
Showing 1 changed file with 5 additions and 7 deletions.
diff --git a/...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
@@ -667,11 +667,7 @@ public static StreamingDataflowWorker fromDataflowWorkerHarnessOptions(
     this.isDoneFuture = new CompletableFuture<>();
 
     this.threadFactory =
-        r -> {
-          Thread t = new Thread(r);
-          t.setDaemon(true);
-          return t;
-        };
+        new ThreadFactoryBuilder().setNameFormat("DataflowWorkUnits-%d").setDaemon(true).build();
     this.workUnitExecutor =
         new BoundedQueueExecutor(
             chooseMaximumNumberOfThreads(),
@@ -691,7 +687,7 @@ public static StreamingDataflowWorker fromDataflowWorkerHarnessOptions(
     memoryMonitorThread.setName("MemoryMonitor");
 
     dispatchThread =
-        threadFactory.newThread(
+        new Thread(
             new Runnable() {
               @Override
               public void run() {
@@ -704,11 +700,12 @@ public void run() {
                 LOG.info("Dispatch done");
               }
             });
+    dispatchThread.setDaemon(true);
     dispatchThread.setPriority(Thread.MIN_PRIORITY);
     dispatchThread.setName("DispatchThread");
 
     commitThread =
-        threadFactory.newThread(
+        new Thread(
             new Runnable() {
               @Override
               public void run() {
@@ -719,6 +716,7 @@ public void run() {
                 }
               }
             });
+    commitThread.setDaemon(true);
     commitThread.setPriority(Thread.MAX_PRIORITY);
     commitThread.setName("CommitThread");