You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2016/11/07 10:19:03 UTC
[1/2] flink git commit: [hotfix] Remove unnecessary
ConfigOptions#key() calls
Repository: flink
Updated Branches:
refs/heads/master 3a4fc537d -> 727484986
[hotfix] Remove unnecessary ConfigOptions#key() calls
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/72748498
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/72748498
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/72748498
Branch: refs/heads/master
Commit: 7274849861ac85dddb2d89f40f96b41d815836ec
Parents: 448d7f0
Author: Ufuk Celebi <uc...@apache.org>
Authored: Fri Nov 4 14:25:13 2016 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Nov 7 11:12:13 2016 +0100
----------------------------------------------------------------------
.../java/org/apache/flink/client/cli/DefaultCLI.java | 2 +-
.../org/apache/flink/runtime/taskmanager/TaskTest.java | 12 ++++++------
2 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/72748498/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java b/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
index 41737d0..e94c2f9 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
@@ -67,7 +67,7 @@ public class DefaultCLI implements CustomCommandLine<StandaloneClusterClient> {
if (commandLine.hasOption(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt())) {
String zkNamespace = commandLine.getOptionValue(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt());
- config.setString(HighAvailabilityOptions.HA_CLUSTER_ID.key(), zkNamespace);
+ config.setString(HighAvailabilityOptions.HA_CLUSTER_ID, zkNamespace);
}
StandaloneClusterDescriptor descriptor = new StandaloneClusterDescriptor(config);
http://git-wip-us.apache.org/repos/asf/flink/blob/72748498/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 761eaa2..382e5c1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -603,8 +603,8 @@ public class TaskTest extends TestLogger {
@Test
public void testInterruptableSharedLockInInvokeAndCancel() throws Exception {
Configuration config = new Configuration();
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL.key(), 5);
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.key(), 50);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL, 5);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT, 50);
Task task = createTask(InvokableInterruptableSharedLockInInvokeAndCancel.class, config);
task.startTaskThread();
@@ -627,8 +627,8 @@ public class TaskTest extends TestLogger {
@Test
public void testFatalErrorAfterUninterruptibleInvoke() throws Exception {
Configuration config = new Configuration();
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL.key(), 5);
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.key(), 50);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL, 5);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT, 50);
Task task = createTask(InvokableUninterruptibleBlockingInvoke.class, config);
@@ -664,8 +664,8 @@ public class TaskTest extends TestLogger {
long timeout = interval + 19292;
Configuration config = new Configuration();
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL.key(), interval);
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.key(), timeout);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL, interval);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT, timeout);
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.setTaskCancellationInterval(interval + 1337);
[2/2] flink git commit: [FLINK-5004] [runtime] Add option to disable
queryable state
Posted by uc...@apache.org.
[FLINK-5004] [runtime] Add option to disable queryable state
- By default, the queryable state server is enabled
- Via config option `query.server.start` it can be disabled
- In tests, we disable it by default
- Not part of the TaskManagerOptions as there are also client
options available
This closes #2748.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/448d7f08
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/448d7f08
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/448d7f08
Branch: refs/heads/master
Commit: 448d7f0832c8ac51c891bb467d0602fa938863d4
Parents: 3a4fc53
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu Nov 3 14:08:11 2016 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Nov 7 11:12:13 2016 +0100
----------------------------------------------------------------------
docs/setup/config.md | 8 +-
.../flink/configuration/ConfigConstants.java | 40 ----------
.../configuration/QueryableStateOptions.java | 79 ++++++++++++++++++++
.../runtime/query/QueryableStateClient.java | 14 +---
.../minicluster/LocalFlinkMiniCluster.scala | 8 +-
.../NetworkEnvironmentConfiguration.scala | 3 -
.../flink/runtime/taskmanager/TaskManager.scala | 41 ++++------
.../io/network/NetworkEnvironmentTest.java | 3 -
...askManagerComponentsStartupShutdownTest.java | 3 +-
.../flink/test/query/QueryableStateITCase.java | 7 +-
10 files changed, 113 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/docs/setup/config.md
----------------------------------------------------------------------
diff --git a/docs/setup/config.md b/docs/setup/config.md
index afe9fcc..5791945 100644
--- a/docs/setup/config.md
+++ b/docs/setup/config.md
@@ -95,7 +95,7 @@ These options are useful for debugging a Flink application for memory and garbag
### Kerberos
-Flink supports Kerberos authentication for the following services
+Flink supports Kerberos authentication for the following services
+ Hadoop Components: such as HDFS, YARN, or HBase.
+ Kafka Connectors (version 0.9+)
@@ -113,7 +113,7 @@ Hadoop components relies on the UserGroupInformation (UGI) implementation to han
While Hadoop uses Kerberos tickets to authenticate users with services initially, the authentication process continues differently afterwards. Instead of saving the ticket to authenticate on a later access, Hadoop creates its own security tokens (DelegationToken) that it passes around. These are authenticated to Kerberos periodically but are independent of the token renewal time. The tokens have a maximum life span identical to the Kerberos ticket maximum life span.
-While using ticket cache mode, please make sure to set the maximum ticket life span high long running jobs.
+While using ticket cache mode, please make sure to set the maximum ticket life span high long running jobs.
If you are on YARN, then it is sufficient to authenticate the client with Kerberos. On a Flink standalone cluster you need to ensure that, initially, all nodes are authenticated with Kerberos using the `kinit` tool.
@@ -437,7 +437,7 @@ Previously this key was named `recovery.mode` and the default value was `standal
- `zookeeper.sasl.disable`: (Default: `true`) Defines if SASL based authentication needs to be enabled or disabled. The configuration value can be set to "true" if ZooKeeper cluster is running in secure mode (Kerberos)
-- `zookeeper.sasl.service-name`: (Default: `zookeeper`) If the ZooKeeper server is configured with a different service name (default:"zookeeper") then it can be supplied using this configuration. A mismatch in service name between client and server configuration will cause the authentication to fail.
+- `zookeeper.sasl.service-name`: (Default: `zookeeper`) If the ZooKeeper server is configured with a different service name (default:"zookeeper") then it can be supplied using this configuration. A mismatch in service name between client and server configuration will cause the authentication to fail.
## Environment
@@ -447,6 +447,8 @@ Previously this key was named `recovery.mode` and the default value was `standal
### Server
+- `query.server.enable`: Enable queryable state (Default: `true`).
+
- `query.server.port`: Port to bind queryable state server to (Default: `0`, binds to random port).
- `query.server.network-threads`: Number of network (Netty's event loop) Threads for queryable state server (Default: `0`, picks number of slots).
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 10dbaaa..b377e54 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -1352,51 +1352,11 @@ public final class ConfigConstants {
/** ACL options supported "creator" or "open" */
public static final String DEFAULT_HA_ZOOKEEPER_CLIENT_ACL = "open";
-
- // ------------------------- Queryable state ------------------------------
-
- /** Port to bind KvState server to. */
- public static final String QUERYABLE_STATE_SERVER_PORT = "query.server.port";
-
- /** Number of network (event loop) threads for the KvState server. */
- public static final String QUERYABLE_STATE_SERVER_NETWORK_THREADS = "query.server.network-threads";
-
- /** Number of query threads for the KvState server. */
- public static final String QUERYABLE_STATE_SERVER_QUERY_THREADS = "query.server.query-threads";
-
- /** Default port to bind KvState server to (0 => pick random free port). */
- public static final int DEFAULT_QUERYABLE_STATE_SERVER_PORT = 0;
-
- /** Default Number of network (event loop) threads for the KvState server (0 => #slots). */
- public static final int DEFAULT_QUERYABLE_STATE_SERVER_NETWORK_THREADS = 0;
-
- /** Default number of query threads for the KvState server (0 => #slots). */
- public static final int DEFAULT_QUERYABLE_STATE_SERVER_QUERY_THREADS = 0;
-
- /** Number of network (event loop) threads for the KvState client. */
- public static final String QUERYABLE_STATE_CLIENT_NETWORK_THREADS = "query.client.network-threads";
-
- /** Number of retries on location lookup failures. */
- public static final String QUERYABLE_STATE_CLIENT_LOOKUP_RETRIES = "query.client.lookup.num-retries";
-
- /** Retry delay on location lookup failures (millis). */
- public static final String QUERYABLE_STATE_CLIENT_LOOKUP_RETRY_DELAY = "query.client.lookup.retry-delay";
-
- /** Default number of query threads for the KvState client (0 => #cores) */
- public static final int DEFAULT_QUERYABLE_STATE_CLIENT_NETWORK_THREADS = 0;
-
- /** Default number of retries on location lookup failures. */
- public static final int DEFAULT_QUERYABLE_STATE_CLIENT_LOOKUP_RETRIES = 3;
-
- /** Default retry delay on location lookup failures. */
- public static final int DEFAULT_QUERYABLE_STATE_CLIENT_LOOKUP_RETRY_DELAY = 1000;
-
// ----------------------------- Metrics ----------------------------
/** The default number of measured latencies to maintain at each operator */
public static final int DEFAULT_METRICS_LATENCY_HISTORY_SIZE = 128;
-
// ----------------------------- Environment Variables ----------------------------
/** The environment variable name which contains the location of the configuration directory */
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-core/src/main/java/org/apache/flink/configuration/QueryableStateOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/QueryableStateOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/QueryableStateOptions.java
new file mode 100644
index 0000000..8b17cfb
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/configuration/QueryableStateOptions.java
@@ -0,0 +1,79 @@
+/*
+ * 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 org.apache.flink.configuration;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * The set of configuration options relating to Queryable State.
+ */
+@PublicEvolving
+public class QueryableStateOptions {
+
+ // ------------------------------------------------------------------------
+ // Server Options
+ // ------------------------------------------------------------------------
+
+ /** Flag to indicate whether to start the queryable state server. */
+ public static final ConfigOption<Boolean> SERVER_ENABLE =
+ key("query.server.enable")
+ .defaultValue(true);
+
+ /** Port to bind KvState server to (0 => pick random available port). */
+ public static final ConfigOption<Integer> SERVER_PORT =
+ key("query.server.port")
+ .defaultValue(0);
+
+ /** Number of network (event loop) threads for the KvState server (0 => #slots). */
+ public static final ConfigOption<Integer> SERVER_NETWORK_THREADS =
+ key("query.server.network-threads")
+ .defaultValue(0);
+
+ /** Number of async query threads for the KvStateServerHandler (0 => #slots). */
+ public static final ConfigOption<Integer> SERVER_ASYNC_QUERY_THREADS =
+ key("query.server.query-threads")
+ .defaultValue(0);
+
+ // ------------------------------------------------------------------------
+ // Client Options
+ // ------------------------------------------------------------------------
+
+ /** Number of network (event loop) threads for the KvState client (0 => Use number of available cores). */
+ public static final ConfigOption<Integer> CLIENT_NETWORK_THREADS =
+ key("query.client.network-threads")
+ .defaultValue(0);
+
+ /** Number of retries on location lookup failures. */
+ public static final ConfigOption<Integer> CLIENT_LOOKUP_RETRIES =
+ key("query.client.lookup.num-retries")
+ .defaultValue(3);
+
+ /** Retry delay on location lookup failures (millis). */
+ public static final ConfigOption<Integer> CLIENT_LOOKUP_RETRY_DELAY =
+ key("query.client.lookup.retry-delay")
+ .defaultValue(1000);
+
+ // ------------------------------------------------------------------------
+
+ /** Not intended to be instantiated */
+ private QueryableStateOptions() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java
index 591c67d..98c3580 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.QueryableStateOptions;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.query.AkkaKvStateLocationLookupService.FixedDelayLookupRetryStrategyFactory;
@@ -123,13 +124,8 @@ public class QueryableStateClient {
FiniteDuration askTimeout = (FiniteDuration) timeout;
- int lookupRetries = config.getInteger(
- ConfigConstants.QUERYABLE_STATE_CLIENT_LOOKUP_RETRIES,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_CLIENT_LOOKUP_RETRIES);
-
- int lookupRetryDelayMillis = config.getInteger(
- ConfigConstants.QUERYABLE_STATE_CLIENT_LOOKUP_RETRY_DELAY,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_CLIENT_LOOKUP_RETRY_DELAY);
+ int lookupRetries = config.getInteger(QueryableStateOptions.CLIENT_LOOKUP_RETRIES);
+ int lookupRetryDelayMillis = config.getInteger(QueryableStateOptions.CLIENT_LOOKUP_RETRY_DELAY);
// Retries if no JobManager is around
LookupRetryStrategyFactory retryStrategy = new FixedDelayLookupRetryStrategyFactory(
@@ -147,9 +143,7 @@ public class QueryableStateClient {
askTimeout,
retryStrategy);
- int numEventLoopThreads = config.getInteger(
- ConfigConstants.QUERYABLE_STATE_CLIENT_NETWORK_THREADS,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_CLIENT_NETWORK_THREADS);
+ int numEventLoopThreads = config.getInteger(QueryableStateOptions.CLIENT_NETWORK_THREADS);
if (numEventLoopThreads == 0) {
numEventLoopThreads = Runtime.getRuntime().availableProcessors();
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
index cad2648..43ccda9 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
@@ -23,7 +23,7 @@ import java.util.concurrent.ExecutorService
import akka.actor.{ActorRef, ActorSystem, Props}
import org.apache.flink.api.common.JobID
import org.apache.flink.api.common.io.FileOutputFormat
-import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.configuration.{QueryableStateOptions, ConfigConstants, Configuration}
import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory
import org.apache.flink.runtime.clusterframework.FlinkResourceManager
import org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager
@@ -74,6 +74,12 @@ class LocalFlinkMiniCluster(
setMemory(config)
initializeIOFormatClasses(config)
+ // Disable queryable state server if nothing else is configured explicitly
+ if (!config.containsKey(QueryableStateOptions.SERVER_ENABLE.key())) {
+ LOG.info("Disabled queryable state server")
+ config.setBoolean(QueryableStateOptions.SERVER_ENABLE, false)
+ }
+
config
}
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
index 893eaa8..14589a1 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
@@ -27,9 +27,6 @@ case class NetworkEnvironmentConfiguration(
networkBufferSize: Int,
memoryType: MemoryType,
ioMode: IOMode,
- queryServerPort: Int,
- queryServerNetworkThreads: Int,
- queryServerQueryThreads: Int,
nettyConfig: Option[NettyConfig] = None,
partitionRequestInitialBackoff: Int = 500,
partitinRequestMaxBackoff: Int = 3000)
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 5073bb0..429d961 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1923,30 +1923,30 @@ object TaskManager {
val kvStateRegistry = new KvStateRegistry()
- val kvStateServer = netConfig.nettyConfig match {
- case Some(nettyConfig) =>
-
- val numNetworkThreads = if (netConfig.queryServerNetworkThreads == 0) {
- nettyConfig.getNumberOfSlots
- } else {
- netConfig.queryServerNetworkThreads
+ val tmConfig = taskManagerConfig.configuration
+ val kvStateServer = tmConfig.getBoolean(QueryableStateOptions.SERVER_ENABLE) match {
+ case true =>
+ val port = tmConfig.getInteger(QueryableStateOptions.SERVER_PORT)
+
+ var numNetworkThreads = tmConfig.getInteger(QueryableStateOptions.SERVER_NETWORK_THREADS)
+ if (numNetworkThreads == 0) {
+ numNetworkThreads = taskManagerConfig.numberOfSlots
}
- val numQueryThreads = if (netConfig.queryServerQueryThreads == 0) {
- nettyConfig.getNumberOfSlots
- } else {
- netConfig.queryServerQueryThreads
+ var numQueryThreads = tmConfig.getInteger(QueryableStateOptions.SERVER_ASYNC_QUERY_THREADS)
+ if (numQueryThreads == 0) {
+ numQueryThreads = taskManagerConfig.numberOfSlots
}
new KvStateServer(
taskManagerAddress.getAddress(),
- netConfig.queryServerPort,
+ port,
numNetworkThreads,
numQueryThreads,
kvStateRegistry,
new DisabledKvStateRequestStats())
- case None => null
+ case false => null
}
// we start the network first, to make sure it can allocate its buffers first
@@ -2235,26 +2235,11 @@ object TaskManager {
val ioMode : IOMode = if (syncOrAsync == "async") IOMode.ASYNC else IOMode.SYNC
- val queryServerPort = configuration.getInteger(
- ConfigConstants.QUERYABLE_STATE_SERVER_PORT,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_SERVER_PORT)
-
- val queryServerNetworkThreads = configuration.getInteger(
- ConfigConstants.QUERYABLE_STATE_SERVER_NETWORK_THREADS,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_SERVER_NETWORK_THREADS)
-
- val queryServerQueryThreads = configuration.getInteger(
- ConfigConstants.QUERYABLE_STATE_SERVER_QUERY_THREADS,
- ConfigConstants.DEFAULT_QUERYABLE_STATE_SERVER_QUERY_THREADS)
-
val networkConfig = NetworkEnvironmentConfiguration(
numNetworkBuffers,
pageSize,
memType,
ioMode,
- queryServerPort,
- queryServerNetworkThreads,
- queryServerQueryThreads,
nettyConfig)
// ----> timeouts, library caching, profiling
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
index a9ad75d..13da18e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -72,9 +72,6 @@ public class NetworkEnvironmentTest {
1024,
MemoryType.HEAP,
IOManager.IOMode.SYNC,
- 0,
- 0,
- 0,
Some.<NettyConfig>empty(),
0,
0);
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
index 1f93e9b..0bcd1ce 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
@@ -104,8 +104,7 @@ public class TaskManagerComponentsStartupShutdownTest {
config);
final NetworkEnvironmentConfiguration netConf = new NetworkEnvironmentConfiguration(
- 32, BUFFER_SIZE, MemoryType.HEAP, IOManager.IOMode.SYNC, 0, 0, 0,
- Option.<NettyConfig>empty(), 0, 0);
+ 32, BUFFER_SIZE, MemoryType.HEAP, IOManager.IOMode.SYNC, Option.<NettyConfig>empty(), 0, 0);
ResourceID taskManagerId = ResourceID.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/448d7f08/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java b/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java
index e0e6ecd..40daeaf 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java
@@ -39,6 +39,7 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.QueryableStateOptions;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
@@ -67,7 +68,6 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import org.junit.AfterClass;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import scala.concurrent.Await;
import scala.concurrent.Future;
@@ -115,8 +115,9 @@ public class QueryableStateITCase extends TestLogger {
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 4);
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_SLOTS_PER_TM);
- config.setInteger(ConfigConstants.QUERYABLE_STATE_CLIENT_NETWORK_THREADS, 1);
- config.setInteger(ConfigConstants.QUERYABLE_STATE_SERVER_NETWORK_THREADS, 1);
+ config.setInteger(QueryableStateOptions.CLIENT_NETWORK_THREADS, 1);
+ config.setBoolean(QueryableStateOptions.SERVER_ENABLE, true);
+ config.setInteger(QueryableStateOptions.SERVER_NETWORK_THREADS, 1);
cluster = new TestingCluster(config, false);
cluster.start(true);