[flink] 02/03: [FLINK-10282][runtime] Add builder for ExecutorThreadFactory

[ch...@apache.org]

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

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

commit 773e47398e6e2073333036bdc4b7d1b3475c5799
Author: zentol <ch...@apache.org>
AuthorDate: Mon Oct 8 14:42:54 2018 +0200

    [FLINK-10282][runtime] Add builder for ExecutorThreadFactory
---
 .../flink/runtime/util/ExecutorThreadFactory.java  | 50 ++++++++++++++++++----
 1 file changed, 41 insertions(+), 9 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ExecutorThreadFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ExecutorThreadFactory.java
index 7673111..5ee1bcf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ExecutorThreadFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ExecutorThreadFactory.java
@@ -18,6 +18,8 @@
 
 package org.apache.flink.runtime.util;
 
+import javax.annotation.Nullable;
+
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -51,6 +53,9 @@ public class ExecutorThreadFactory implements ThreadFactory {
 
 	private final String namePrefix;
 
+	private final int threadPriority;
+
+	@Nullable
 	private final UncaughtExceptionHandler exceptionHandler;
 
 	// ------------------------------------------------------------------------
@@ -81,14 +86,20 @@ public class ExecutorThreadFactory implements ThreadFactory {
 	 * @param exceptionHandler The uncaught exception handler for the threads
 	 */
 	public ExecutorThreadFactory(String poolName, UncaughtExceptionHandler exceptionHandler) {
-		checkNotNull(poolName, "poolName");
+		this(poolName, Thread.NORM_PRIORITY, exceptionHandler);
+	}
+
+	ExecutorThreadFactory(
+			final String poolName,
+			final int threadPriority,
+			@Nullable final UncaughtExceptionHandler exceptionHandler) {
+		this.namePrefix = checkNotNull(poolName, "poolName") + "-thread-";
+		this.threadPriority = threadPriority;
+		this.exceptionHandler = exceptionHandler;
 
 		SecurityManager securityManager = System.getSecurityManager();
 		this.group = (securityManager != null) ? securityManager.getThreadGroup() :
-				Thread.currentThread().getThreadGroup();
-
-		this.namePrefix = poolName + "-thread-";
-		this.exceptionHandler = exceptionHandler;
+			Thread.currentThread().getThreadGroup();
 	}
 
 	// ------------------------------------------------------------------------
@@ -98,10 +109,7 @@ public class ExecutorThreadFactory implements ThreadFactory {
 		Thread t = new Thread(group, runnable, namePrefix + threadNumber.getAndIncrement());
 		t.setDaemon(true);
 
-		// normalize the priority
-		if (t.getPriority() != Thread.NORM_PRIORITY) {
-			t.setPriority(Thread.NORM_PRIORITY);
-		}
+		t.setPriority(threadPriority);
 
 		// optional handler for uncaught exceptions
 		if (exceptionHandler != null) {
@@ -113,4 +121,28 @@ public class ExecutorThreadFactory implements ThreadFactory {
 
 	// --------------------------------------------------------------------------------------------
 
+	public static final class Builder {
+		private String poolName;
+		private int priority = Thread.NORM_PRIORITY;
+		private UncaughtExceptionHandler exceptionHandler = FatalExitExceptionHandler.INSTANCE;
+
+		public Builder setPoolName(final String poolName) {
+			this.poolName = poolName;
+			return this;
+		}
+		
+		public Builder setThreadPriority(final int priority) {
+			this.priority = priority;
+			return this;
+		}
+		
+		public Builder setExceptionHandler(final UncaughtExceptionHandler exceptionHandler) {
+			this.exceptionHandler = exceptionHandler;
+			return this;
+		}
+		
+		public ExecutorThreadFactory build() {
+			return new ExecutorThreadFactory(poolName, priority, exceptionHandler);
+		}
+	}
 }