flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] asfgit closed pull request #6839: [FLINK-10253] Run MetricQueryService with lower priority
Date Wed, 17 Oct 2018 13:18:56 GMT
asfgit closed pull request #6839: [FLINK-10253] Run MetricQueryService with lower priority
URL: https://github.com/apache/flink/pull/6839
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/docs/_includes/generated/metric_configuration.html b/docs/_includes/generated/metric_configuration.html
index 5c073f3ad59..420bb7f1da2 100644
--- a/docs/_includes/generated/metric_configuration.html
+++ b/docs/_includes/generated/metric_configuration.html
@@ -12,6 +12,11 @@
             <td style="word-wrap: break-word;">"0"</td>
             <td>The port range used for Flink's internal metric query service. Accepts
a list of ports (“50100,50101”), ranges(“50100-50200”) or a combination of both. It
is recommended to set a range of ports to avoid collisions when multiple Flink components
are running on the same machine. Per default Flink will pick a random port.</td>
         </tr>
+        <tr>
+            <td><h5>metrics.internal.query-service.thread-priority</h5></td>
+            <td style="word-wrap: break-word;">1</td>
+            <td>The thread priority used for Flink's internal metric query service.
The thread is created by Akka's thread pool executor. The range of the priority is from 1
(MIN_PRIORITY) to 10 (MAX_PRIORITY). Warning, increasing this value may bring the main Flink
components down.</td>
+        </tr>
         <tr>
             <td><h5>metrics.latency.granularity</h5></td>
             <td style="word-wrap: break-word;">"operator"</td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java
index 0e7268ee052..0785b347335 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java
@@ -156,6 +156,18 @@
 			"ports to avoid collisions when multiple Flink components are running on the same machine.
Per default " +
 			"Flink will pick a random port.");
 
+	/**
+	 * The thread priority for Flink's internal metric query service. The {@code 1} means the
min priority and the
+	 * {@code 10} means the max priority.
+	 */
+	public static final ConfigOption<Integer> QUERY_SERVICE_THREAD_PRIORITY =
+		key("metrics.internal.query-service.thread-priority")
+		.defaultValue(1)
+		.withDescription("The thread priority used for Flink's internal metric query service. The
thread is created" +
+			" by Akka's thread pool executor. " +
+			"The range of the priority is from 1 (MIN_PRIORITY) to 10 (MAX_PRIORITY). " +
+			"Warning, increasing this value may bring the main Flink components down.");
+
 	private MetricOptions() {
 	}
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
index 00b61737d20..430af98bc2e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
@@ -24,6 +24,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MetricOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.configuration.WebOptions;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -290,7 +291,8 @@ private static Config getExecutorConfigByExecutorMode(Configuration configuratio
 			case FORK_JOIN_EXECUTOR:
 				return AkkaUtils.getForkJoinExecutorConfig(configuration);
 			case FIXED_THREAD_POOL_EXECUTOR:
-				return AkkaUtils.getThreadPoolExecutorConfig();
+				return AkkaUtils.getThreadPoolExecutorConfig(
+					configuration.getInteger(MetricOptions.QUERY_SERVICE_THREAD_PRIORITY));
 			default:
 				throw new IllegalArgumentException(String.format("Unknown ActorSystemExecutorMode %s.",
executorMode));
 		}
diff --git a/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadFactory.scala b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadFactory.scala
new file mode 100644
index 00000000000..d6f6d76ec51
--- /dev/null
+++ b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadFactory.scala
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package akka.dispatch
+
+import java.util.concurrent.ThreadFactory
+
+/**
+  * Composition over the [[DispatcherPrerequisites.threadFactory]] that set priority
+  * for newly created threads.
+  *
+  * @param newThreadPriority priority that will be set to each newly created thread
+  *                          should be between Thread.MIN_PRIORITY and Thread.MAX_PRIORITY
+  */
+class PriorityThreadFactory(
+  prerequisites: DispatcherPrerequisites,
+  newThreadPriority: Int) extends ThreadFactory {
+    override def newThread(r: Runnable): Thread = {
+      val newThread = prerequisites.threadFactory.newThread(r)
+      newThread.setPriority(newThreadPriority)
+      newThread
+    }
+}
diff --git a/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcher.scala b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcher.scala
new file mode 100644
index 00000000000..06ef001f1a7
--- /dev/null
+++ b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcher.scala
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package akka.dispatch
+
+import com.typesafe.config.Config
+
+/**
+  * Akka Dispatcher that creates thread with configurable priority.
+  *
+  * Example of configuration:
+  *
+  *   low-priority-threads-dispatcher {
+  *     type = akka.dispatch.PriorityThreadsDispatcher
+  *     executor = "thread-pool-executor"
+  *     # should be between Thread.MIN_PRIORITY (which is 1) and Thread.MAX_PRIORITY (which
is 10)
+  *     threads-priority = 1
+  *     thread-pool-executor {
+  *       core-pool-size-min = 0
+  *       core-pool-size-factor = 2.0
+  *       core-pool-size-max = 10
+  *     }
+  *   }
+  *
+  * Two arguments constructor (the primary constructor) is automatically called by Akka
+  * when it founds:
+  *   abcde-dispatcher {
+  *     type = akka.dispatch.PriorityThreadsDispatcher <-- the class that Akka will instantiate
+  *     ...
+  *   }
+  *
+  * @param config passed automatically by Akka, should contains information about threads
priority
+  * @param prerequisites passed automatically by Akka
+  */
+class PriorityThreadsDispatcher(config: Config, prerequisites: DispatcherPrerequisites)
+  extends DispatcherConfigurator(
+    config,
+    new PriorityThreadsDispatcherPrerequisites(
+      prerequisites,
+      config.getInt(PriorityThreadsDispatcher.threadPriorityConfigKey)
+    )
+  )
+
+object PriorityThreadsDispatcher {
+  /**
+    * Configuration key under which int value should be placed.
+    */
+  val threadPriorityConfigKey = "thread-priority"
+}
diff --git a/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcherPrerequisites.scala
b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcherPrerequisites.scala
new file mode 100644
index 00000000000..a62f12185f6
--- /dev/null
+++ b/flink-runtime/src/main/scala/akka/dispatch/PriorityThreadsDispatcherPrerequisites.scala
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package akka.dispatch
+
+/**
+  * Composition over [[DefaultDispatcherPrerequisites]] that replaces thread factory with
one that
+  * allow to configure thread priority.
+  *
+  * @param newThreadPriority priority that will be set to each newly created thread
+  *                          should be between Thread.MIN_PRIORITY and Thread.MAX_PRIORITY
+  */
+class PriorityThreadsDispatcherPrerequisites(
+  prerequisites: DispatcherPrerequisites,
+  newThreadPriority: Int) extends DispatcherPrerequisites {
+
+  private val defaultDispatcherPrerequisites : DefaultDispatcherPrerequisites =
+    new DefaultDispatcherPrerequisites(
+      eventStream = prerequisites.eventStream,
+      scheduler = prerequisites.scheduler,
+      dynamicAccess = prerequisites.dynamicAccess,
+      settings = prerequisites.settings,
+      mailboxes = prerequisites.mailboxes,
+      defaultExecutionContext = prerequisites.defaultExecutionContext,
+      threadFactory = new PriorityThreadFactory(prerequisites, newThreadPriority)
+  )
+
+  override def threadFactory : java.util.concurrent.ThreadFactory = {
+    defaultDispatcherPrerequisites.threadFactory
+  }
+
+  override def eventStream : akka.event.EventStream = {
+    defaultDispatcherPrerequisites.eventStream
+  }
+
+  override def scheduler : akka.actor.Scheduler = {
+    defaultDispatcherPrerequisites.scheduler
+  }
+
+  override def dynamicAccess : akka.actor.DynamicAccess = {
+    defaultDispatcherPrerequisites.dynamicAccess
+  }
+
+  override def settings : akka.actor.ActorSystem.Settings = {
+    defaultDispatcherPrerequisites.settings
+  }
+
+  override def mailboxes : akka.dispatch.Mailboxes = {
+    defaultDispatcherPrerequisites.mailboxes
+  }
+
+  override def defaultExecutionContext : scala.Option[scala.concurrent.ExecutionContext]
= {
+    defaultDispatcherPrerequisites.defaultExecutionContext
+  }
+}
+
+object PriorityThreadsDispatcherPrerequisites {
+  def apply(prerequisites: DispatcherPrerequisites, newThreadPriority: Int):
+    PriorityThreadsDispatcherPrerequisites =
+      new PriorityThreadsDispatcherPrerequisites(prerequisites, newThreadPriority)
+}
+
+
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
index de2f35ef8fd..80c17cc7bfd 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
@@ -26,7 +26,7 @@ import akka.actor._
 import akka.pattern.{ask => akkaAsk}
 import com.typesafe.config.{Config, ConfigFactory}
 import org.apache.flink.api.common.time.Time
-import org.apache.flink.configuration.{AkkaOptions, Configuration, IllegalConfigurationException,
SecurityOptions}
+import org.apache.flink.configuration._
 import org.apache.flink.runtime.concurrent.FutureUtils
 import org.apache.flink.runtime.net.SSLUtils
 import org.apache.flink.util.NetUtils
@@ -291,12 +291,20 @@ object AkkaUtils {
     ConfigFactory.parseString(config)
   }
 
-  def getThreadPoolExecutorConfig: Config = {
+  def getThreadPoolExecutorConfig(threadPriority: Int): Config = {
+    if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY)
{
+      throw new IllegalConfigurationException("The threadPriority must be between "
+        + Thread.MIN_PRIORITY + " and " + Thread.MAX_PRIORITY +
+        ", but it is " + threadPriority)
+    }
+
     val configString = s"""
        |akka {
        |  actor {
        |    default-dispatcher {
+       |      type = akka.dispatch.PriorityThreadsDispatcher
        |      executor = "thread-pool-executor"
+       |      thread-priority = $threadPriority
        |      thread-pool-executor {
        |        core-pool-size-min = 2
        |        core-pool-size-factor = 2.0
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
b/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
index e5c1668df0a..ae6209abdd9 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
@@ -20,14 +20,17 @@ package org.apache.flink.runtime.akka
 
 import java.net.{InetAddress, InetSocketAddress}
 
-import org.apache.flink.configuration.{AkkaOptions, Configuration, IllegalConfigurationException}
+import org.apache.flink.configuration.{AkkaOptions, Configuration, IllegalConfigurationException,
MetricOptions}
 import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils.AddressResolution
+import org.apache.flink.runtime.metrics.util.MetricUtils
 import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils
 import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils.AkkaProtocol
 import org.apache.flink.util.NetUtils
+import org.junit.Assert.assertEquals
 import org.junit.runner.RunWith
 import org.scalatest.junit.JUnitRunner
 import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
+import org.slf4j.LoggerFactory
 
 @RunWith(classOf[JUnitRunner])
 class AkkaUtilsTest
@@ -184,13 +187,44 @@ class AkkaUtilsTest
   }
 
   test("getAkkaConfig respects executor config") {
-    val akkaConfig = AkkaUtils.getAkkaConfig(
+    var akkaConfig = AkkaUtils.getAkkaConfig(
       new Configuration(),
       "localhost",
       1234,
-      AkkaUtils.getThreadPoolExecutorConfig)
+      AkkaUtils.getThreadPoolExecutorConfig(Thread.MIN_PRIORITY))
 
     akkaConfig.getString("akka.actor.default-dispatcher.executor") should
       equal("thread-pool-executor")
+
+    akkaConfig.getInt("akka.actor.default-dispatcher.thread-priority") should
+      equal(Thread.MIN_PRIORITY)
+
+    akkaConfig = AkkaUtils.getAkkaConfig(
+      new Configuration(),
+      "localhost",
+      1234,
+      AkkaUtils.getThreadPoolExecutorConfig(Thread.MAX_PRIORITY))
+
+    akkaConfig.getInt("akka.actor.default-dispatcher.thread-priority") should
+      equal(Thread.MAX_PRIORITY)
+  }
+
+  test("thread priority for metrics ActorSystem ") {
+    var actorSystem = MetricUtils.startMetricsActorSystem(
+      new Configuration, "localhost", LoggerFactory.getLogger("AkkaUtilsTest"))
+    //test default thread priority
+    val defaultThreadPriority = actorSystem.settings.config.getInt(
+      "akka.actor.default-dispatcher.thread-priority")
+    //check default value
+    assertEquals(Thread.MIN_PRIORITY, defaultThreadPriority)
+
+    val config = new Configuration()
+    config.setInteger(MetricOptions.QUERY_SERVICE_THREAD_PRIORITY, Thread.MAX_PRIORITY)
+    actorSystem = MetricUtils.startMetricsActorSystem(
+      config, "localhost", LoggerFactory.getLogger("AkkaUtilsTest"))
+    val threadPriority = actorSystem.settings.config.getInt(
+      "akka.actor.default-dispatcher.thread-priority")
+    //check config value
+    assertEquals(Thread.MAX_PRIORITY, threadPriority)
   }
 }


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message