You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/06/11 10:04:16 UTC
[flink] 03/03: [FLINK-11392][network] Rename NetworkEnviroment to
NettyShuffleEnviroment
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c104dafd26f32d939461cb880ddfc881a8d14d5f
Author: Andrey Zagrebin <az...@gmail.com>
AuthorDate: Thu May 30 15:16:16 2019 +0200
[FLINK-11392][network] Rename NetworkEnviroment to NettyShuffleEnviroment
This closes #8608.
---
.../flink/addons/hbase/HBaseConnectorITCase.java | 4 +-
.../flink/configuration/ConfigConstants.java | 12 +-
...ns.java => NettyShuffleEnvironmentOptions.java} | 4 +-
...TaskManagerHeapSizeCalculationJavaBashTest.java | 28 +-
...vironment.java => NettyShuffleEnvironment.java} | 30 +-
.../io/network/buffer/NetworkBufferPool.java | 14 +-
.../runtime/io/network/netty/NettyConfig.java | 18 +-
.../network/partition/ResultPartitionFactory.java | 3 +-
.../partition/consumer/SingleInputGateFactory.java | 7 +-
.../runtime/taskexecutor/TaskManagerServices.java | 8 +-
...a => NettyShuffleEnvironmentConfiguration.java} | 102 +++----
.../ContaineredTaskManagerParametersTest.java | 4 +-
...er.java => NettyShuffleEnvironmentBuilder.java} | 36 +--
...tTest.java => NettyShuffleEnvironmentTest.java} | 17 +-
.../network/netty/NettyConnectionManagerTest.java | 8 +-
.../PartialConsumePipelinedResultTest.java | 4 +-
.../io/network/partition/PartitionTestUtils.java | 6 +-
.../network/partition/ResultPartitionBuilder.java | 4 +-
.../io/network/partition/ResultPartitionTest.java | 6 +-
.../partition/consumer/InputChannelBuilder.java | 4 +-
.../partition/consumer/InputGateTestBase.java | 4 +-
.../partition/consumer/SingleInputGateBuilder.java | 8 +-
.../partition/consumer/SingleInputGateTest.java | 28 +-
.../NettyShuffleEnvironmentConfigurationTest.java | 334 +++++++++++++++++++++
.../taskexecutor/NetworkBufferCalculationTest.java | 28 +-
.../NetworkEnvironmentConfigurationTest.java | 334 ---------------------
.../taskexecutor/TaskExecutorSubmissionTest.java | 12 +-
.../runtime/taskexecutor/TaskExecutorTest.java | 30 +-
.../taskexecutor/TaskManagerRunnerStartupTest.java | 4 +-
.../TaskSubmissionTestEnvironment.java | 12 +-
.../runtime/taskmanager/TaskAsyncCallTest.java | 4 +-
.../TaskCancelAsyncProducerConsumerITCase.java | 4 +-
.../apache/flink/runtime/taskmanager/TaskTest.java | 4 +-
.../runtime/util/JvmExitOnFatalErrorTest.java | 4 +-
.../streaming/runtime/io/InputProcessorUtil.java | 4 +-
.../StreamNetworkBenchmarkEnvironment.java | 20 +-
.../StreamNetworkThroughputBenchmarkTest.java | 8 +-
.../tasks/InterruptSensitiveRestoreTest.java | 4 +-
.../runtime/tasks/StreamTaskTerminationTest.java | 4 +-
.../streaming/runtime/tasks/StreamTaskTest.java | 4 +-
.../runtime/tasks/SynchronousCheckpointITCase.java | 4 +-
.../tasks/TaskCheckpointingBehaviourTest.java | 4 +-
.../flink/test/cancelling/CancelingTestBase.java | 4 +-
.../EventTimeWindowCheckpointingITCase.java | 4 +-
.../manual/StreamingScalabilityAndLatency.java | 4 +-
.../SuccessAfterNetworkBuffersFailureITCase.java | 4 +-
...tractTaskManagerProcessFailureRecoveryTest.java | 4 +-
.../JobManagerHAProcessFailureRecoveryITCase.java | 4 +-
.../recovery/ProcessFailureCancelingITCase.java | 4 +-
.../flink/test/runtime/NettyEpollITCase.java | 4 +-
.../yarn/YARNSessionCapacitySchedulerITCase.java | 4 +-
.../apache/flink/yarn/YarnConfigurationITCase.java | 6 +-
52 files changed, 613 insertions(+), 610 deletions(-)
diff --git a/flink-connectors/flink-hbase/src/test/java/org/apache/flink/addons/hbase/HBaseConnectorITCase.java b/flink-connectors/flink-hbase/src/test/java/org/apache/flink/addons/hbase/HBaseConnectorITCase.java
index 20ba45e..bd992c5 100644
--- a/flink-connectors/flink-hbase/src/test/java/org/apache/flink/addons/hbase/HBaseConnectorITCase.java
+++ b/flink-connectors/flink-hbase/src/test/java/org/apache/flink/addons/hbase/HBaseConnectorITCase.java
@@ -27,7 +27,7 @@ import org.apache.flink.api.java.ExecutionEnvironmentFactory;
import org.apache.flink.api.java.LocalEnvironment;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.java.BatchTableEnvironment;
@@ -362,7 +362,7 @@ public class HBaseConnectorITCase extends HBaseTestingClusterAutostarter {
public static void setAsContext() {
Configuration config = new Configuration();
// the default network buffers size (10% of heap max =~ 150MB) seems to much for this test case
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(80L << 20)); // 80 MB
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(80L << 20)); // 80 MB
final LocalEnvironment le = new LocalEnvironment(config);
initializeContextEnvironment(new ExecutionEnvironmentFactory() {
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 b1b9713..a4de6cc 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
@@ -204,7 +204,7 @@ public final class ConfigConstants {
public static final String TASK_MANAGER_IPC_PORT_KEY = "taskmanager.rpc.port";
/**
- * @deprecated use {@link NetworkEnvironmentOptions#DATA_PORT} instead
+ * @deprecated use {@link NettyShuffleEnvironmentOptions#DATA_PORT} instead
*/
@Deprecated
public static final String TASK_MANAGER_DATA_PORT_KEY = "taskmanager.data.port";
@@ -212,7 +212,7 @@ public final class ConfigConstants {
/**
* Config parameter to override SSL support for taskmanager's data transport.
*
- * @deprecated use {@link NetworkEnvironmentOptions#DATA_SSL_ENABLED} instead
+ * @deprecated use {@link NettyShuffleEnvironmentOptions#DATA_SSL_ENABLED} instead
*/
@Deprecated
public static final String TASK_MANAGER_DATA_SSL_ENABLED = "taskmanager.data.ssl.enabled";
@@ -270,7 +270,7 @@ public final class ConfigConstants {
* The config parameter defining the number of buffers used in the network stack. This defines the
* number of possible tasks and shuffles.
*
- * @deprecated Use {@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS} instead
+ * @deprecated Use {@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS} instead
*/
@Deprecated
public static final String TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY = "taskmanager.network.numberOfBuffers";
@@ -1392,7 +1392,7 @@ public final class ConfigConstants {
* The default network port the task manager expects to receive transfer envelopes on. The {@code 0} means that
* the TaskManager searches for a free port.
*
- * @deprecated use {@link NetworkEnvironmentOptions#DATA_PORT} instead
+ * @deprecated use {@link NettyShuffleEnvironmentOptions#DATA_PORT} instead
*/
@Deprecated
public static final int DEFAULT_TASK_MANAGER_DATA_PORT = 0;
@@ -1400,7 +1400,7 @@ public final class ConfigConstants {
/**
* The default value to override ssl support for task manager's data transport.
*
- * @deprecated use {@link NetworkEnvironmentOptions#DATA_SSL_ENABLED} instead
+ * @deprecated use {@link NettyShuffleEnvironmentOptions#DATA_SSL_ENABLED} instead
*/
@Deprecated
public static final boolean DEFAULT_TASK_MANAGER_DATA_SSL_ENABLED = true;
@@ -1424,7 +1424,7 @@ public final class ConfigConstants {
/**
* Config key has been deprecated. Therefore, no default value required.
*
- * @deprecated {@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS} provides the default value now
+ * @deprecated {@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS} provides the default value now
*/
@Deprecated
public static final int DEFAULT_TASK_MANAGER_NETWORK_NUM_BUFFERS = 2048;
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/NetworkEnvironmentOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
similarity index 99%
rename from flink-core/src/main/java/org/apache/flink/configuration/NetworkEnvironmentOptions.java
rename to flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
index 920cf5b..699bd90 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/NetworkEnvironmentOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
@@ -29,7 +29,7 @@ import static org.apache.flink.configuration.ConfigOptions.key;
*/
@PublicEvolving
@ConfigGroups(groups = @ConfigGroup(name = "NetworkNetty", keyPrefix = "taskmanager.network.netty"))
-public class NetworkEnvironmentOptions {
+public class NettyShuffleEnvironmentOptions {
// ------------------------------------------------------------------------
// Network General Options
@@ -212,5 +212,5 @@ public class NetworkEnvironmentOptions {
// ------------------------------------------------------------------------
/** Not intended to be instantiated. */
- private NetworkEnvironmentOptions() {}
+ private NettyShuffleEnvironmentOptions() {}
}
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
index 9d6b638..87cf194 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
@@ -20,10 +20,10 @@ package org.apache.flink.dist;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.OperatingSystem;
import org.apache.flink.util.TestLogger;
@@ -71,7 +71,7 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
}
/**
- * Tests that {@link NetworkEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)}
+ * Tests that {@link NettyShuffleEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)}
* has the same result as the shell script.
*/
@Test
@@ -159,9 +159,9 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
config.setLong(KEY_TASKM_MEM_SIZE, javaMemMB);
config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, useOffHeap);
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, netBufMemFrac);
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(netBufMemMin));
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(netBufMemMax));
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, netBufMemFrac);
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(netBufMemMin));
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(netBufMemMax));
if (managedMemSizeMB == 0) {
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");
@@ -202,7 +202,7 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
Configuration config = getConfig(javaMemMB, useOffHeap, frac, min, max, managedMemSize, managedMemFrac);
long totalJavaMemorySize = ((long) javaMemMB) << 20; // megabytes to bytes
final int networkBufMB =
- (int) (NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(totalJavaMemorySize, config) >> 20);
+ (int) (NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(totalJavaMemorySize, config) >> 20);
// max (exclusive): total - netbuf
managedMemSize = Math.min(javaMemMB - networkBufMB - 1, ran.nextInt(Integer.MAX_VALUE));
} else {
@@ -228,14 +228,14 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
final long totalJavaMemorySizeMB = config.getLong(KEY_TASKM_MEM_SIZE, 0L);
- long javaNetworkBufMem = NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(
+ long javaNetworkBufMem = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
totalJavaMemorySizeMB << 20, config);
String[] command = {"src/test/bin/calcTMNetBufMem.sh",
totalJavaMemorySizeMB + "m",
- String.valueOf(config.getFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION)),
- config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN),
- config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX)};
+ String.valueOf(config.getFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION)),
+ config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN),
+ config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX)};
String scriptOutput = executeScript(command);
@@ -272,9 +272,9 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
String[] command = {"src/test/bin/calcTMHeapSizeMB.sh",
totalJavaMemorySizeMB + "m",
String.valueOf(config.getBoolean(TaskManagerOptions.MEMORY_OFF_HEAP)),
- String.valueOf(config.getFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION)),
- config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN),
- config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX),
+ String.valueOf(config.getFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION)),
+ config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN),
+ config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX),
config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE),
String.valueOf(config.getFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION))};
String scriptOutput = executeScript(command);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironment.java
similarity index 92%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironment.java
index 8483f84..9cf86cb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironment.java
@@ -51,7 +51,7 @@ import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.shuffle.ShuffleEnvironmentContext;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -72,9 +72,9 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* The network environment contains the data structures that keep track of all intermediate results
* and shuffle data exchanges.
*/
-public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, SingleInputGate> {
+public class NettyShuffleEnvironment implements ShuffleEnvironment<ResultPartition, SingleInputGate> {
- private static final Logger LOG = LoggerFactory.getLogger(NetworkEnvironment.class);
+ private static final Logger LOG = LoggerFactory.getLogger(NettyShuffleEnvironment.class);
private static final String METRIC_GROUP_NETWORK = "Network";
private static final String METRIC_TOTAL_MEMORY_SEGMENT = "TotalMemorySegments";
@@ -89,7 +89,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
private final ResourceID taskExecutorLocation;
- private final NetworkEnvironmentConfiguration config;
+ private final NettyShuffleEnvironmentConfiguration config;
private final NetworkBufferPool networkBufferPool;
@@ -105,9 +105,9 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
private boolean isClosed;
- private NetworkEnvironment(
+ private NettyShuffleEnvironment(
ResourceID taskExecutorLocation,
- NetworkEnvironmentConfiguration config,
+ NettyShuffleEnvironmentConfiguration config,
NetworkBufferPool networkBufferPool,
ConnectionManager connectionManager,
ResultPartitionManager resultPartitionManager,
@@ -124,8 +124,8 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
this.isClosed = false;
}
- public static NetworkEnvironment create(
- NetworkEnvironmentConfiguration config,
+ public static NettyShuffleEnvironment create(
+ NettyShuffleEnvironmentConfiguration config,
ResourceID taskExecutorLocation,
TaskEventPublisher taskEventPublisher,
MetricGroup metricGroup,
@@ -165,7 +165,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
taskEventPublisher,
networkBufferPool);
- return new NetworkEnvironment(
+ return new NettyShuffleEnvironment(
taskExecutorLocation,
config,
networkBufferPool,
@@ -205,7 +205,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
}
@VisibleForTesting
- public NetworkEnvironmentConfiguration getConfiguration() {
+ public NettyShuffleEnvironmentConfiguration getConfiguration() {
return config;
}
@@ -244,7 +244,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
MetricGroup outputGroup,
MetricGroup buffersGroup) {
synchronized (lock) {
- Preconditions.checkState(!isClosed, "The NetworkEnvironment has already been shut down.");
+ Preconditions.checkState(!isClosed, "The NettyShuffleEnvironment has already been shut down.");
ResultPartition[] resultPartitions = new ResultPartition[resultPartitionDeploymentDescriptors.size()];
int counter = 0;
@@ -267,7 +267,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
MetricGroup inputGroup,
MetricGroup buffersGroup) {
synchronized (lock) {
- Preconditions.checkState(!isClosed, "The NetworkEnvironment has already been shut down.");
+ Preconditions.checkState(!isClosed, "The NettyShuffleEnvironment has already been shut down.");
InputChannelMetrics inputChannelMetrics = new InputChannelMetrics(parentGroup);
SingleInputGate[] inputGates = new SingleInputGate[inputGateDeploymentDescriptors.size()];
@@ -331,7 +331,7 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
@Override
public int start() throws IOException {
synchronized (lock) {
- Preconditions.checkState(!isClosed, "The NetworkEnvironment has already been shut down.");
+ Preconditions.checkState(!isClosed, "The NettyShuffleEnvironment has already been shut down.");
LOG.info("Starting the network environment and its components.");
@@ -395,8 +395,8 @@ public class NetworkEnvironment implements ShuffleEnvironment<ResultPartition, S
}
}
- public static NetworkEnvironment fromShuffleContext(ShuffleEnvironmentContext shuffleEnvironmentContext) {
- NetworkEnvironmentConfiguration networkConfig = NetworkEnvironmentConfiguration.fromConfiguration(
+ public static NettyShuffleEnvironment fromShuffleContext(ShuffleEnvironmentContext shuffleEnvironmentContext) {
+ NettyShuffleEnvironmentConfiguration networkConfig = NettyShuffleEnvironmentConfiguration.fromConfiguration(
shuffleEnvironmentContext.getConfiguration(),
shuffleEnvironmentContext.getMaxJvmHeapMemory(),
shuffleEnvironmentContext.isLocalCommunicationOnly(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
index 2053100..51c6aa1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.io.network.buffer;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.core.memory.MemorySegmentProvider;
@@ -149,9 +149,9 @@ public class NetworkBufferPool implements BufferPoolFactory, MemorySegmentProvid
totalNumberOfMemorySegments - numTotalRequiredBuffers,
totalNumberOfMemorySegments,
memorySegmentSize,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key()));
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key()));
}
this.numTotalRequiredBuffers += numberOfSegmentsToRequest;
@@ -284,9 +284,9 @@ public class NetworkBufferPool implements BufferPoolFactory, MemorySegmentProvid
totalNumberOfMemorySegments - numTotalRequiredBuffers,
totalNumberOfMemorySegments,
memorySegmentSize,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key()));
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key()));
}
this.numTotalRequiredBuffers += numRequiredBuffers;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
index f0a96fe..9730907 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyConfig.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.net.SSLUtils;
import org.slf4j.Logger;
@@ -98,37 +98,37 @@ public class NettyConfig {
// ------------------------------------------------------------------------
public int getServerConnectBacklog() {
- return config.getInteger(NetworkEnvironmentOptions.CONNECT_BACKLOG);
+ return config.getInteger(NettyShuffleEnvironmentOptions.CONNECT_BACKLOG);
}
public int getNumberOfArenas() {
// default: number of slots
- final int configValue = config.getInteger(NetworkEnvironmentOptions.NUM_ARENAS);
+ final int configValue = config.getInteger(NettyShuffleEnvironmentOptions.NUM_ARENAS);
return configValue == -1 ? numberOfSlots : configValue;
}
public int getServerNumThreads() {
// default: number of task slots
- final int configValue = config.getInteger(NetworkEnvironmentOptions.NUM_THREADS_SERVER);
+ final int configValue = config.getInteger(NettyShuffleEnvironmentOptions.NUM_THREADS_SERVER);
return configValue == -1 ? numberOfSlots : configValue;
}
public int getClientNumThreads() {
// default: number of task slots
- final int configValue = config.getInteger(NetworkEnvironmentOptions.NUM_THREADS_CLIENT);
+ final int configValue = config.getInteger(NettyShuffleEnvironmentOptions.NUM_THREADS_CLIENT);
return configValue == -1 ? numberOfSlots : configValue;
}
public int getClientConnectTimeoutSeconds() {
- return config.getInteger(NetworkEnvironmentOptions.CLIENT_CONNECT_TIMEOUT_SECONDS);
+ return config.getInteger(NettyShuffleEnvironmentOptions.CLIENT_CONNECT_TIMEOUT_SECONDS);
}
public int getSendAndReceiveBufferSize() {
- return config.getInteger(NetworkEnvironmentOptions.SEND_RECEIVE_BUFFER_SIZE);
+ return config.getInteger(NettyShuffleEnvironmentOptions.SEND_RECEIVE_BUFFER_SIZE);
}
public TransportType getTransportType() {
- String transport = config.getString(NetworkEnvironmentOptions.TRANSPORT_TYPE);
+ String transport = config.getString(NettyShuffleEnvironmentOptions.TRANSPORT_TYPE);
switch (transport) {
case "nio":
@@ -155,7 +155,7 @@ public class NettyConfig {
}
public boolean getSSLEnabled() {
- return config.getBoolean(NetworkEnvironmentOptions.DATA_SSL_ENABLED)
+ return config.getBoolean(NettyShuffleEnvironmentOptions.DATA_SSL_ENABLED)
&& SSLUtils.isInternalSSLEnabled(config);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index 1d0a040..1fccf1d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferPoolFactory;
import org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
@@ -38,7 +39,7 @@ import java.io.IOException;
import java.util.Optional;
/**
- * Factory for {@link ResultPartition} to use in {@link org.apache.flink.runtime.io.network.NetworkEnvironment}.
+ * Factory for {@link ResultPartition} to use in {@link NettyShuffleEnvironment}.
*/
public class ResultPartitionFactory {
private static final Logger LOG = LoggerFactory.getLogger(ResultPartition.class);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
index 554d723..60bc70d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.io.network.ConnectionManager;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventPublisher;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferPoolFactory;
@@ -33,7 +34,7 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.function.SupplierWithException;
import org.slf4j.Logger;
@@ -46,7 +47,7 @@ import java.io.IOException;
import static org.apache.flink.runtime.shuffle.ShuffleUtils.applyWithShuffleTypeCheck;
/**
- * Factory for {@link SingleInputGate} to use in {@link org.apache.flink.runtime.io.network.NetworkEnvironment}.
+ * Factory for {@link SingleInputGate} to use in {@link NettyShuffleEnvironment}.
*/
public class SingleInputGateFactory {
private static final Logger LOG = LoggerFactory.getLogger(SingleInputGate.class);
@@ -78,7 +79,7 @@ public class SingleInputGateFactory {
public SingleInputGateFactory(
@Nonnull ResourceID taskExecutorLocation,
- @Nonnull NetworkEnvironmentConfiguration networkConfig,
+ @Nonnull NettyShuffleEnvironmentConfiguration networkConfig,
@Nonnull ConnectionManager connectionManager,
@Nonnull ResultPartitionManager partitionManager,
@Nonnull TaskEventPublisher taskEventPublisher,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index a8e909a..5d62076 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -30,7 +30,7 @@ import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
@@ -38,7 +38,7 @@ import org.apache.flink.runtime.shuffle.ShuffleEnvironmentContext;
import org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager;
import org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable;
import org.apache.flink.runtime.taskexecutor.slot.TimerService;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.ConfigurationParserUtils;
import org.apache.flink.util.ExceptionUtils;
@@ -320,7 +320,7 @@ public class TaskManagerServices {
taskManagerMetricGroup,
ioManager);
- return NetworkEnvironment.fromShuffleContext(shuffleEnvironmentContext);
+ return NettyShuffleEnvironment.fromShuffleContext(shuffleEnvironmentContext);
}
/**
@@ -426,7 +426,7 @@ public class TaskManagerServices {
Preconditions.checkArgument(totalJavaMemorySizeMB > 0);
// subtract the Java memory used for network buffers (always off-heap)
- final long networkBufMB = NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(
+ final long networkBufMB = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
totalJavaMemorySizeMB << 20, // megabytes to bytes
config) >> 20; // bytes to megabytes
final long remainingJavaMemorySizeMB = totalJavaMemorySizeMB - networkBufMB;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
similarity index 79%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
index 6767acd..a6ff17b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
@@ -39,8 +39,8 @@ import java.net.InetSocketAddress;
/**
* Configuration object for the network stack.
*/
-public class NetworkEnvironmentConfiguration {
- private static final Logger LOG = LoggerFactory.getLogger(NetworkEnvironmentConfiguration.class);
+public class NettyShuffleEnvironmentConfiguration {
+ private static final Logger LOG = LoggerFactory.getLogger(NettyShuffleEnvironmentConfiguration.class);
private final int numNetworkBuffers;
@@ -62,7 +62,7 @@ public class NetworkEnvironmentConfiguration {
private final NettyConfig nettyConfig;
- public NetworkEnvironmentConfiguration(
+ public NettyShuffleEnvironmentConfiguration(
int numNetworkBuffers,
int networkBufferSize,
int partitionRequestInitialBackoff,
@@ -132,9 +132,9 @@ public class NetworkEnvironmentConfiguration {
* @param maxJvmHeapMemory the maximum JVM heap size (in bytes)
* @param localTaskManagerCommunication true, to skip initializing the network stack
* @param taskManagerAddress identifying the IP address under which the TaskManager will be accessible
- * @return NetworkEnvironmentConfiguration
+ * @return NettyShuffleEnvironmentConfiguration
*/
- public static NetworkEnvironmentConfiguration fromConfiguration(
+ public static NettyShuffleEnvironmentConfiguration fromConfiguration(
Configuration configuration,
long maxJvmHeapMemory,
boolean localTaskManagerCommunication,
@@ -148,17 +148,17 @@ public class NetworkEnvironmentConfiguration {
final NettyConfig nettyConfig = createNettyConfig(configuration, localTaskManagerCommunication, taskManagerAddress, dataport);
- int initialRequestBackoff = configuration.getInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL);
- int maxRequestBackoff = configuration.getInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX);
+ int initialRequestBackoff = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL);
+ int maxRequestBackoff = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX);
- int buffersPerChannel = configuration.getInteger(NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL);
- int extraBuffersPerGate = configuration.getInteger(NetworkEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
+ int buffersPerChannel = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL);
+ int extraBuffersPerGate = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
- boolean isCreditBased = nettyConfig != null && configuration.getBoolean(NetworkEnvironmentOptions.NETWORK_CREDIT_MODEL);
+ boolean isCreditBased = nettyConfig != null && configuration.getBoolean(NettyShuffleEnvironmentOptions.NETWORK_CREDIT_MODEL);
- boolean isNetworkDetailedMetrics = configuration.getBoolean(NetworkEnvironmentOptions.NETWORK_DETAILED_METRICS);
+ boolean isNetworkDetailedMetrics = configuration.getBoolean(NettyShuffleEnvironmentOptions.NETWORK_DETAILED_METRICS);
- return new NetworkEnvironmentConfiguration(
+ return new NettyShuffleEnvironmentConfiguration(
numberOfNetworkBuffers,
pageSize,
initialRequestBackoff,
@@ -182,10 +182,10 @@ public class NetworkEnvironmentConfiguration {
* <ul>
* <li>{@link TaskManagerOptions#MANAGED_MEMORY_SIZE},</li>
* <li>{@link TaskManagerOptions#MANAGED_MEMORY_FRACTION},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}, and</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS} (fallback if the ones above do not exist)</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}, and</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS} (fallback if the ones above do not exist)</li>
* </ul>.
*
* @param config configuration object
@@ -220,7 +220,7 @@ public class NetworkEnvironmentConfiguration {
// jvmHeapNoNet = jvmHeap - networkBufBytes
// = jvmHeap - Math.min(networkBufMax, Math.max(networkBufMin, jvmHeap * netFraction)
// jvmHeap = jvmHeapNoNet / (1.0 - networkBufFraction)
- float networkBufFraction = config.getFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
+ float networkBufFraction = config.getFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
long networkBufSize = (long) (jvmHeapNoNet / (1.0 - networkBufFraction) * networkBufFraction);
return calculateNewNetworkBufferMemory(config, networkBufSize, maxJvmHeapMemory);
}
@@ -231,10 +231,10 @@ public class NetworkEnvironmentConfiguration {
*
* <p>The following configuration parameters are involved:
* <ul>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}, and</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS} (fallback if the ones above do not exist)</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}, and</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS} (fallback if the ones above do not exist)</li>
* </ul>.
*
* @param totalJavaMemorySize overall available memory to use (in bytes)
@@ -248,18 +248,18 @@ public class NetworkEnvironmentConfiguration {
final long networkBufBytes;
if (hasNewNetworkConfig(config)) {
- float networkBufFraction = config.getFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
+ float networkBufFraction = config.getFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
long networkBufSize = (long) (totalJavaMemorySize * networkBufFraction);
networkBufBytes = calculateNewNetworkBufferMemory(config, networkBufSize, totalJavaMemorySize);
} else {
// use old (deprecated) network buffers parameter
- int numNetworkBuffers = config.getInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS);
+ int numNetworkBuffers = config.getInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS);
networkBufBytes = (long) numNetworkBuffers * (long) segmentSize;
checkOldNetworkConfig(numNetworkBuffers);
ConfigurationParserUtils.checkConfigParameter(networkBufBytes < totalJavaMemorySize,
- networkBufBytes, NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS.key(),
+ networkBufBytes, NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS.key(),
"Network buffer memory size too large: " + networkBufBytes + " >= " +
totalJavaMemorySize + " (total JVM memory size)");
}
@@ -273,9 +273,9 @@ public class NetworkEnvironmentConfiguration {
*
* <p>The following configuration parameters are involved:
* <ul>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
- * <li>{@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN},</li>
+ * <li>{@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}</li>
* </ul>.
*
* @param config configuration object
@@ -285,9 +285,9 @@ public class NetworkEnvironmentConfiguration {
* @return memory to use for network buffers (in bytes)
*/
private static long calculateNewNetworkBufferMemory(Configuration config, long networkBufSize, long maxJvmHeapMemory) {
- float networkBufFraction = config.getFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
- long networkBufMin = MemorySize.parse(config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN)).getBytes();
- long networkBufMax = MemorySize.parse(config.getString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX)).getBytes();
+ float networkBufFraction = config.getFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
+ long networkBufMin = MemorySize.parse(config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN)).getBytes();
+ long networkBufMax = MemorySize.parse(config.getString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX)).getBytes();
int pageSize = ConfigurationParserUtils.getPageSize(config);
@@ -297,9 +297,9 @@ public class NetworkEnvironmentConfiguration {
ConfigurationParserUtils.checkConfigParameter(networkBufBytes < maxJvmHeapMemory,
"(" + networkBufFraction + ", " + networkBufMin + ", " + networkBufMax + ")",
- "(" + NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key() + ", " +
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key() + ", " +
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key() + ")",
+ "(" + NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key() + ", " +
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key() + ", " +
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key() + ")",
"Network buffer memory size too large: " + networkBufBytes + " >= " +
maxJvmHeapMemory + " (maximum JVM memory size)");
@@ -316,7 +316,7 @@ public class NetworkEnvironmentConfiguration {
@SuppressWarnings("deprecation")
private static void checkOldNetworkConfig(final int numNetworkBuffers) {
ConfigurationParserUtils.checkConfigParameter(numNetworkBuffers > 0, numNetworkBuffers,
- NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS.key(),
"Must have at least one network buffer");
}
@@ -337,23 +337,23 @@ public class NetworkEnvironmentConfiguration {
final long networkBufMax) throws IllegalConfigurationException {
ConfigurationParserUtils.checkConfigParameter(networkBufFraction > 0.0f && networkBufFraction < 1.0f, networkBufFraction,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(),
"Network buffer memory fraction of the free memory must be between 0.0 and 1.0");
ConfigurationParserUtils.checkConfigParameter(networkBufMin >= pageSize, networkBufMin,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(),
"Minimum memory for network buffers must allow at least one network " +
"buffer with respect to the memory segment size");
ConfigurationParserUtils.checkConfigParameter(networkBufMax >= pageSize, networkBufMax,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(),
"Maximum memory for network buffers must allow at least one network " +
"buffer with respect to the memory segment size");
ConfigurationParserUtils.checkConfigParameter(networkBufMax >= networkBufMin, networkBufMax,
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(),
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(),
"Maximum memory for network buffers must not be smaller than minimum memory (" +
- NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key() + ": " + networkBufMin + ")");
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key() + ": " + networkBufMin + ")");
}
/**
@@ -366,10 +366,10 @@ public class NetworkEnvironmentConfiguration {
@SuppressWarnings("deprecation")
@VisibleForTesting
public static boolean hasNewNetworkConfig(final Configuration config) {
- return config.contains(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION) ||
- config.contains(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN) ||
- config.contains(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX) ||
- !config.contains(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS);
+ return config.contains(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION) ||
+ config.contains(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN) ||
+ config.contains(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX) ||
+ !config.contains(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS);
}
/**
@@ -379,8 +379,8 @@ public class NetworkEnvironmentConfiguration {
* @return the data port
*/
private static int getDataport(Configuration configuration) {
- final int dataport = configuration.getInteger(NetworkEnvironmentOptions.DATA_PORT);
- ConfigurationParserUtils.checkConfigParameter(dataport >= 0, dataport, NetworkEnvironmentOptions.DATA_PORT.key(),
+ final int dataport = configuration.getInteger(NettyShuffleEnvironmentOptions.DATA_PORT);
+ ConfigurationParserUtils.checkConfigParameter(dataport >= 0, dataport, NettyShuffleEnvironmentOptions.DATA_PORT.key(),
"Leave config parameter empty or use 0 to let the system choose a port automatically.");
return dataport;
@@ -398,13 +398,13 @@ public class NetworkEnvironmentConfiguration {
final int numberOfNetworkBuffers;
if (!hasNewNetworkConfig(configuration)) {
// fallback: number of network buffers
- numberOfNetworkBuffers = configuration.getInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS);
+ numberOfNetworkBuffers = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS);
checkOldNetworkConfig(numberOfNetworkBuffers);
} else {
- if (configuration.contains(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS)) {
+ if (configuration.contains(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS)) {
LOG.info("Ignoring old (but still present) network buffer configuration via {}.",
- NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS.key());
+ NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS.key());
}
final long networkMemorySize = calculateNewNetworkBufferMemory(configuration, maxJvmHeapMemory);
@@ -479,7 +479,7 @@ public class NetworkEnvironmentConfiguration {
return false;
}
else {
- final NetworkEnvironmentConfiguration that = (NetworkEnvironmentConfiguration) obj;
+ final NettyShuffleEnvironmentConfiguration that = (NettyShuffleEnvironmentConfiguration) obj;
return this.numNetworkBuffers == that.numNetworkBuffers &&
this.networkBufferSize == that.networkBufferSize &&
@@ -494,7 +494,7 @@ public class NetworkEnvironmentConfiguration {
@Override
public String toString() {
- return "NetworkEnvironmentConfiguration{" +
+ return "NettyShuffleEnvironmentConfiguration{" +
", numNetworkBuffers=" + numNetworkBuffers +
", networkBufferSize=" + networkBufferSize +
", partitionRequestInitialBackoff=" + partitionRequestInitialBackoff +
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
index c6e50fa..c49f8e6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.clusterframework;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
@@ -52,7 +52,7 @@ public class ContaineredTaskManagerParametersTest extends TestLogger {
ConfigConstants.DEFAULT_YARN_HEAP_CUTOFF);
long cutoff = Math.max((long) (CONTAINER_MEMORY * memoryCutoffRatio), minCutoff);
- final long networkBufMB = NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(
+ final long networkBufMB = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
(CONTAINER_MEMORY - cutoff) << 20, // megabytes to bytes
conf) >> 20; // bytes to megabytes
assertEquals(networkBufMB + cutoff, params.taskManagerDirectMemoryLimitMB());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
similarity index 68%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentBuilder.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
index 03ec3a4..9238dda 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
@@ -24,12 +24,12 @@ import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
/**
- * Builder for the {@link NetworkEnvironment}.
+ * Builder for the {@link NettyShuffleEnvironment}.
*/
-public class NetworkEnvironmentBuilder {
+public class NettyShuffleEnvironmentBuilder {
private int numNetworkBuffers = 1024;
@@ -57,69 +57,69 @@ public class NetworkEnvironmentBuilder {
private IOManager ioManager = new IOManagerAsync();
- public NetworkEnvironmentBuilder setTaskManagerLocation(ResourceID taskManagerLocation) {
+ public NettyShuffleEnvironmentBuilder setTaskManagerLocation(ResourceID taskManagerLocation) {
this.taskManagerLocation = taskManagerLocation;
return this;
}
- public NetworkEnvironmentBuilder setNumNetworkBuffers(int numNetworkBuffers) {
+ public NettyShuffleEnvironmentBuilder setNumNetworkBuffers(int numNetworkBuffers) {
this.numNetworkBuffers = numNetworkBuffers;
return this;
}
- public NetworkEnvironmentBuilder setNetworkBufferSize(int networkBufferSize) {
+ public NettyShuffleEnvironmentBuilder setNetworkBufferSize(int networkBufferSize) {
this.networkBufferSize = networkBufferSize;
return this;
}
- public NetworkEnvironmentBuilder setPartitionRequestInitialBackoff(int partitionRequestInitialBackoff) {
+ public NettyShuffleEnvironmentBuilder setPartitionRequestInitialBackoff(int partitionRequestInitialBackoff) {
this.partitionRequestInitialBackoff = partitionRequestInitialBackoff;
return this;
}
- public NetworkEnvironmentBuilder setPartitionRequestMaxBackoff(int partitionRequestMaxBackoff) {
+ public NettyShuffleEnvironmentBuilder setPartitionRequestMaxBackoff(int partitionRequestMaxBackoff) {
this.partitionRequestMaxBackoff = partitionRequestMaxBackoff;
return this;
}
- public NetworkEnvironmentBuilder setNetworkBuffersPerChannel(int networkBuffersPerChannel) {
+ public NettyShuffleEnvironmentBuilder setNetworkBuffersPerChannel(int networkBuffersPerChannel) {
this.networkBuffersPerChannel = networkBuffersPerChannel;
return this;
}
- public NetworkEnvironmentBuilder setFloatingNetworkBuffersPerGate(int floatingNetworkBuffersPerGate) {
+ public NettyShuffleEnvironmentBuilder setFloatingNetworkBuffersPerGate(int floatingNetworkBuffersPerGate) {
this.floatingNetworkBuffersPerGate = floatingNetworkBuffersPerGate;
return this;
}
- public NetworkEnvironmentBuilder setIsCreditBased(boolean isCreditBased) {
+ public NettyShuffleEnvironmentBuilder setIsCreditBased(boolean isCreditBased) {
this.isCreditBased = isCreditBased;
return this;
}
- public NetworkEnvironmentBuilder setNettyConfig(NettyConfig nettyConfig) {
+ public NettyShuffleEnvironmentBuilder setNettyConfig(NettyConfig nettyConfig) {
this.nettyConfig = nettyConfig;
return this;
}
- public NetworkEnvironmentBuilder setTaskEventDispatcher(TaskEventDispatcher taskEventDispatcher) {
+ public NettyShuffleEnvironmentBuilder setTaskEventDispatcher(TaskEventDispatcher taskEventDispatcher) {
this.taskEventDispatcher = taskEventDispatcher;
return this;
}
- public NetworkEnvironmentBuilder setMetricGroup(MetricGroup metricGroup) {
+ public NettyShuffleEnvironmentBuilder setMetricGroup(MetricGroup metricGroup) {
this.metricGroup = metricGroup;
return this;
}
- public NetworkEnvironmentBuilder setIOManager(IOManager ioManager) {
+ public NettyShuffleEnvironmentBuilder setIOManager(IOManager ioManager) {
this.ioManager = ioManager;
return this;
}
- public NetworkEnvironment build() {
- return NetworkEnvironment.create(
- new NetworkEnvironmentConfiguration(
+ public NettyShuffleEnvironment build() {
+ return NettyShuffleEnvironment.create(
+ new NettyShuffleEnvironmentConfiguration(
numNetworkBuffers,
networkBufferSize,
partitionRequestInitialBackoff,
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/NettyShuffleEnvironmentTest.java
similarity index 94%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
index 4606318..d4f0727 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/NettyShuffleEnvironmentTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.io.network;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.core.memory.MemorySegmentProvider;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.partition.ResultPartition;
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBuilder;
import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.util.TestLogger;
import org.junit.Rule;
import org.junit.Test;
@@ -47,10 +48,10 @@ import static org.mockito.Mockito.verify;
import static org.powermock.api.mockito.PowerMockito.spy;
/**
- * Various tests for the {@link NetworkEnvironment} class.
+ * Various tests for the {@link NettyShuffleEnvironment} class.
*/
@RunWith(Parameterized.class)
-public class NetworkEnvironmentTest {
+public class NettyShuffleEnvironmentTest extends TestLogger {
@Parameterized.Parameter
public boolean enableCreditBasedFlowControl;
@@ -69,7 +70,7 @@ public class NetworkEnvironmentTest {
*/
@Test
public void testRegisterTaskUsesBoundedBuffers() throws Exception {
- final NetworkEnvironment network = new NetworkEnvironmentBuilder()
+ final NettyShuffleEnvironment network = new NettyShuffleEnvironmentBuilder()
.setIsCreditBased(enableCreditBasedFlowControl)
.build();
@@ -141,7 +142,7 @@ public class NetworkEnvironmentTest {
bufferCount = 20;
} else {
// incoming: 2 exclusive buffers per channel
- bufferCount = 10 + 10 * NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue();
+ bufferCount = 10 + 10 * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue();
}
testRegisterTaskWithLimitedBuffers(bufferCount);
@@ -160,7 +161,7 @@ public class NetworkEnvironmentTest {
bufferCount = 19;
} else {
// incoming: 2 exclusive buffers per channel
- bufferCount = 10 + 10 * NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() - 1;
+ bufferCount = 10 + 10 * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() - 1;
}
expectedException.expect(IOException.class);
@@ -169,7 +170,7 @@ public class NetworkEnvironmentTest {
}
private void testRegisterTaskWithLimitedBuffers(int bufferPoolSize) throws Exception {
- final NetworkEnvironment network = new NetworkEnvironmentBuilder()
+ final NettyShuffleEnvironment network = new NettyShuffleEnvironmentBuilder()
.setNumNetworkBuffers(bufferPoolSize)
.setIsCreditBased(enableCreditBasedFlowControl)
.build();
@@ -262,7 +263,7 @@ public class NetworkEnvironmentTest {
* @return input gate with some fake settings
*/
private SingleInputGate createSingleInputGate(
- NetworkEnvironment network, ResultPartitionType partitionType, int numberOfChannels) {
+ NettyShuffleEnvironment network, ResultPartitionType partitionType, int numberOfChannels) {
return spy(new SingleInputGateBuilder()
.setNumberOfChannels(numberOfChannels)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManagerTest.java
index 094426d..e8b079c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyConnectionManagerTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
import org.apache.flink.util.NetUtils;
@@ -111,9 +111,9 @@ public class NettyConnectionManagerTest {
// Expected number of threads
Configuration flinkConfig = new Configuration();
- flinkConfig.setInteger(NetworkEnvironmentOptions.NUM_ARENAS, numberOfArenas);
- flinkConfig.setInteger(NetworkEnvironmentOptions.NUM_THREADS_CLIENT, 3);
- flinkConfig.setInteger(NetworkEnvironmentOptions.NUM_THREADS_SERVER, 4);
+ flinkConfig.setInteger(NettyShuffleEnvironmentOptions.NUM_ARENAS, numberOfArenas);
+ flinkConfig.setInteger(NettyShuffleEnvironmentOptions.NUM_THREADS_CLIENT, 3);
+ flinkConfig.setInteger(NettyShuffleEnvironmentOptions.NUM_THREADS_SERVER, 4);
NettyConfig config = new NettyConfig(
InetAddress.getLocalHost(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
index a6efa79..12ab077 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.buffer.Buffer;
@@ -62,7 +62,7 @@ public class PartialConsumePipelinedResultTest extends TestLogger {
private static Configuration getFlinkConfiguration() {
final Configuration config = new Configuration();
config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
return config;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
index df07db7..ace58fa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.io.network.partition;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
/**
* This class should consolidate all mocking logic for ResultPartitions.
@@ -36,11 +36,11 @@ public class PartitionTestUtils {
}
public static ResultPartition createPartition(
- NetworkEnvironment environment,
+ NettyShuffleEnvironment environment,
ResultPartitionType partitionType,
int numChannels) {
return new ResultPartitionBuilder()
- .setupBufferPoolFactoryFromNetworkEnvironment(environment)
+ .setupBufferPoolFactoryFromNettyShuffleEnvironment(environment)
.setResultPartitionType(partitionType)
.setNumberOfSubpartitions(numChannels)
.build();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
index f34a598..45b25c0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
@@ -20,7 +20,7 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
@@ -85,7 +85,7 @@ public class ResultPartitionBuilder {
return this;
}
- public ResultPartitionBuilder setupBufferPoolFactoryFromNetworkEnvironment(NetworkEnvironment environment) {
+ public ResultPartitionBuilder setupBufferPoolFactoryFromNettyShuffleEnvironment(NettyShuffleEnvironment environment) {
return setNetworkBuffersPerChannel(environment.getConfiguration().networkBuffersPerChannel())
.setFloatingNetworkBuffersPerGate(environment.getConfiguration().floatingNetworkBuffersPerGate())
.setNetworkBufferPool(environment.getNetworkBufferPool());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
index de5eaf1..7545988 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
@@ -20,8 +20,8 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
@@ -231,7 +231,7 @@ public class ResultPartitionTest {
*/
private void testReleaseMemory(final ResultPartitionType resultPartitionType) throws Exception {
final int numAllBuffers = 10;
- final NetworkEnvironment network = new NetworkEnvironmentBuilder()
+ final NettyShuffleEnvironment network = new NettyShuffleEnvironmentBuilder()
.setNumNetworkBuffers(numAllBuffers).build();
final ResultPartition resultPartition = createPartition(network, resultPartitionType, 1);
try {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelBuilder.java
index 2e6fcbc..d8e9cc2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelBuilder.java
@@ -22,7 +22,7 @@ import org.apache.flink.core.memory.MemorySegmentProvider;
import org.apache.flink.runtime.io.network.ConnectionID;
import org.apache.flink.runtime.io.network.ConnectionManager;
import org.apache.flink.runtime.io.network.LocalConnectionManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.TaskEventPublisher;
import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics;
@@ -99,7 +99,7 @@ public class InputChannelBuilder {
return this;
}
- InputChannelBuilder setupFromNetworkEnvironment(NetworkEnvironment network) {
+ InputChannelBuilder setupFromNettyShuffleEnvironment(NettyShuffleEnvironment network) {
this.partitionManager = network.getResultPartitionManager();
this.connectionManager = network.getConnectionManager();
this.initialBackoff = network.getConfiguration().partitionRequestInitialBackoff();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputGateTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputGateTestBase.java
index 9f5a4a2..b32ba89 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputGateTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputGateTestBase.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.runtime.io.AsyncDataInput;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.junit.runner.RunWith;
@@ -98,7 +98,7 @@ public abstract class InputGateTestBase {
}
protected SingleInputGate createInputGate(
- NetworkEnvironment environment, int numberOfInputChannels, ResultPartitionType partitionType) {
+ NettyShuffleEnvironment environment, int numberOfInputChannels, ResultPartitionType partitionType) {
SingleInputGateBuilder builder = new SingleInputGateBuilder()
.setNumberOfChannels(numberOfInputChannels)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index 8c04d5f..b23af06 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -18,13 +18,13 @@
package org.apache.flink.runtime.io.network.partition.consumer;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvider;
import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvider.ResponseHandle;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.function.SupplierWithException;
import java.io.IOException;
@@ -75,8 +75,8 @@ public class SingleInputGateBuilder {
return this;
}
- public SingleInputGateBuilder setupBufferPoolFactory(NetworkEnvironment environment) {
- NetworkEnvironmentConfiguration config = environment.getConfiguration();
+ public SingleInputGateBuilder setupBufferPoolFactory(NettyShuffleEnvironment environment) {
+ NettyShuffleEnvironmentConfiguration config = environment.getConfiguration();
this.bufferPoolFactory = SingleInputGateFactory.createBufferPoolFactory(
environment.getNetworkBufferPool(),
config.isCreditBased(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index 03fd3b2..ebc9888 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -24,8 +24,8 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.event.TaskEvent;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
@@ -346,7 +346,7 @@ public class SingleInputGateTest extends InputGateTestBase {
int initialBackoff = 137;
int maxBackoff = 1001;
- final NetworkEnvironment netEnv = new NetworkEnvironmentBuilder()
+ final NettyShuffleEnvironment netEnv = new NettyShuffleEnvironmentBuilder()
.setPartitionRequestInitialBackoff(initialBackoff)
.setPartitionRequestMaxBackoff(maxBackoff)
.setIsCreditBased(enableCreditBasedFlowControl)
@@ -410,7 +410,7 @@ public class SingleInputGateTest extends InputGateTestBase {
*/
@Test
public void testRequestBuffersWithRemoteInputChannel() throws Exception {
- final NetworkEnvironment network = createNetworkEnvironment();
+ final NettyShuffleEnvironment network = createNettyShuffleEnvironment();
final SingleInputGate inputGate = createInputGate(network, 1, ResultPartitionType.PIPELINED_BOUNDED);
int buffersPerChannel = 2;
int extraNetworkBuffersPerGate = 8;
@@ -418,7 +418,7 @@ public class SingleInputGateTest extends InputGateTestBase {
try {
RemoteInputChannel remote =
InputChannelBuilder.newBuilder()
- .setupFromNetworkEnvironment(network)
+ .setupFromNettyShuffleEnvironment(network)
.buildRemoteAndSetToGate(inputGate);
inputGate.setup();
@@ -446,7 +446,7 @@ public class SingleInputGateTest extends InputGateTestBase {
*/
@Test
public void testRequestBuffersWithUnknownInputChannel() throws Exception {
- final NetworkEnvironment network = createNetworkEnvironment();
+ final NettyShuffleEnvironment network = createNettyShuffleEnvironment();
final SingleInputGate inputGate = createInputGate(network, 1, ResultPartitionType.PIPELINED_BOUNDED);
int buffersPerChannel = 2;
int extraNetworkBuffersPerGate = 8;
@@ -497,7 +497,7 @@ public class SingleInputGateTest extends InputGateTestBase {
*/
@Test
public void testUpdateUnknownInputChannel() throws Exception {
- final NetworkEnvironment network = createNetworkEnvironment();
+ final NettyShuffleEnvironment network = createNettyShuffleEnvironment();
final SingleInputGate inputGate = createInputGate(network, 2, ResultPartitionType.PIPELINED);
try {
@@ -564,8 +564,8 @@ public class SingleInputGateTest extends InputGateTestBase {
}
@Test
- public void testInputGateRemovalFromNetworkEnvironment() throws Exception {
- NetworkEnvironment network = createNetworkEnvironment();
+ public void testInputGateRemovalFromNettyShuffleEnvironment() throws Exception {
+ NettyShuffleEnvironment network = createNettyShuffleEnvironment();
try {
int numberOfGates = 10;
@@ -587,7 +587,7 @@ public class SingleInputGateTest extends InputGateTestBase {
// ---------------------------------------------------------------------------------------------
private static Map<InputGateID, SingleInputGate> createInputGateWithLocalChannels(
- NetworkEnvironment network,
+ NettyShuffleEnvironment network,
int numberOfGates,
@SuppressWarnings("SameParameterValue") int numberOfLocalChannels) {
ShuffleDescriptor[] channelDescs = new NettyShuffleDescriptor[numberOfLocalChannels];
@@ -624,19 +624,19 @@ public class SingleInputGateTest extends InputGateTestBase {
}
private void addUnknownInputChannel(
- NetworkEnvironment network,
+ NettyShuffleEnvironment network,
SingleInputGate inputGate,
ResultPartitionID partitionId,
int channelIndex) {
InputChannelBuilder.newBuilder()
.setChannelIndex(channelIndex)
.setPartitionId(partitionId)
- .setupFromNetworkEnvironment(network)
+ .setupFromNettyShuffleEnvironment(network)
.buildUnknownAndSetToGate(inputGate);
}
- private NetworkEnvironment createNetworkEnvironment() {
- return new NetworkEnvironmentBuilder()
+ private NettyShuffleEnvironment createNettyShuffleEnvironment() {
+ return new NettyShuffleEnvironmentBuilder()
.setIsCreditBased(enableCreditBasedFlowControl)
.build();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java
new file mode 100644
index 0000000..1aa4817
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java
@@ -0,0 +1,334 @@
+/*
+ * 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.runtime.taskexecutor;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Unit test for {@link NettyShuffleEnvironmentConfiguration}.
+ */
+public class NettyShuffleEnvironmentConfigurationTest extends TestLogger {
+
+ private static final long MEM_SIZE_PARAM = 128L * 1024 * 1024;
+
+ /**
+ * Verifies that {@link NettyShuffleEnvironmentConfiguration#fromConfiguration(Configuration, long, boolean, InetAddress)}
+ * returns the correct result for new configurations via
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_REQUEST_BACKOFF_INITIAL},
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_REQUEST_BACKOFF_MAX},
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_PER_CHANNEL} and
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_EXTRA_BUFFERS_PER_GATE}
+ */
+ @Test
+ public void testNetworkRequestBackoffAndBuffers() {
+
+ // set some non-default values
+ final Configuration config = new Configuration();
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, 10);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100);
+
+ final NettyShuffleEnvironmentConfiguration networkConfig = NettyShuffleEnvironmentConfiguration.fromConfiguration(
+ config,
+ MEM_SIZE_PARAM,
+ true,
+ InetAddress.getLoopbackAddress());
+
+ assertEquals(networkConfig.partitionRequestInitialBackoff(), 100);
+ assertEquals(networkConfig.partitionRequestMaxBackoff(), 200);
+ assertEquals(networkConfig.networkBuffersPerChannel(), 10);
+ assertEquals(networkConfig.floatingNetworkBuffersPerGate(), 100);
+ }
+
+ /**
+ * Test for {@link NettyShuffleEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using old
+ * configurations via {@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS}.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void calculateNetworkBufOld() {
+ Configuration config = new Configuration();
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
+
+ // note: actual network buffer memory size is independent of the totalJavaMemorySize
+ assertEquals(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(10L << 20, config));
+ assertEquals(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(64L << 20, config));
+
+ // test integer overflow in the memory size
+ int numBuffers = (int) ((2L << 32) / MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes()); // 2^33
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, numBuffers);
+ assertEquals(2L << 32, NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(2L << 33, config));
+ }
+
+ /**
+ * Test for {@link NettyShuffleEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using new
+ * configurations via {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN} and
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}.
+ */
+ @Test
+ public void calculateNetworkBufNew() throws Exception {
+ Configuration config = new Configuration();
+
+ // (1) defaults
+ final Float defaultFrac = NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue();
+ final Long defaultMin = MemorySize.parse(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes();
+ final Long defaultMax = MemorySize.parse(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue()).getBytes();
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config));
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config));
+
+ calculateNetworkBufNew(config);
+ }
+
+ /**
+ * Helper to test {@link NettyShuffleEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} with the
+ * new configuration parameters.
+ *
+ * @param config configuration object
+ */
+ private static void calculateNetworkBufNew(final Configuration config) {
+ // (2) fixed size memory
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(1L << 20)); // 1MB
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(1L << 20)); // 1MB
+
+
+ // note: actual network buffer memory size is independent of the totalJavaMemorySize
+ assertEquals(1 << 20, NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(10L << 20, config));
+ assertEquals(1 << 20, NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(64L << 20, config));
+ assertEquals(1 << 20, NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(1L << 30, config));
+
+ // (3) random fraction, min, and max values
+ Random ran = new Random();
+ for (int i = 0; i < 1_000; ++i){
+ float frac = Math.max(ran.nextFloat(), Float.MIN_VALUE);
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, frac);
+
+ long min = Math.max(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(), ran.nextLong());
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(min));
+
+
+ long max = Math.max(min, ran.nextLong());
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(max));
+
+ long javaMem = Math.max(max + 1, ran.nextLong());
+
+ final long networkBufMem = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(javaMem, config);
+
+ if (networkBufMem < min) {
+ fail("Lower bound not met with configuration: " + config.toString());
+ }
+
+ if (networkBufMem > max) {
+ fail("Upper bound not met with configuration: " + config.toString());
+ }
+
+ if (networkBufMem > min && networkBufMem < max) {
+ if ((javaMem * frac) != networkBufMem) {
+ fail("Wrong network buffer memory size with configuration: " + config.toString() +
+ ". Expected value: " + (javaMem * frac) + " actual value: " + networkBufMem + '.');
+ }
+ }
+ }
+ }
+
+ /**
+ * Test for {@link NettyShuffleEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using mixed
+ * old/new configurations.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void calculateNetworkBufMixed() throws Exception {
+ Configuration config = new Configuration();
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
+
+ final Float defaultFrac = NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue();
+ final Long defaultMin = MemorySize.parse(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes();
+ final Long defaultMax = MemorySize.parse(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue()).getBytes();
+
+
+ // old + 1 new parameter = new:
+ Configuration config1 = config.clone();
+ config1.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ assertEquals(enforceBounds((long) (0.1f * (10L << 20)), defaultMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config1));
+ assertEquals(enforceBounds((long) (0.1f * (10L << 30)), defaultMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
+
+ config1 = config.clone();
+ long newMin = MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(); // smallest value possible
+ config1.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(newMin));
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), newMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 20), config1));
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), newMin, defaultMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
+
+ config1 = config.clone();
+ long newMax = Math.max(64L << 20 + 1, MemorySize.parse(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes());
+ config1.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(newMax));
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, newMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config1));
+ assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, newMax),
+ NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config1));
+
+ // old + any new parameter = new:
+ calculateNetworkBufNew(config);
+ }
+
+ /**
+ * Returns the value or the lower/upper bound in case the value is less/greater than the lower/upper bound, respectively.
+ *
+ * @param value value to inspect
+ * @param lower lower bound
+ * @param upper upper bound
+ *
+ * @return <tt>min(upper, max(lower, value))</tt>
+ */
+ private static long enforceBounds(final long value, final long lower, final long upper) {
+ return Math.min(upper, Math.max(lower, value));
+ }
+
+ /**
+ * Test for {@link TaskManagerServices#calculateHeapSizeMB(long, Configuration)} with some
+ * manually calculated scenarios.
+ */
+ @Test
+ public void calculateHeapSizeMB() throws Exception {
+ Configuration config = new Configuration();
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(64L << 20)); // 64MB
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(1L << 30)); // 1GB
+
+ config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, false);
+ assertEquals(900, TaskManagerServices.calculateHeapSizeMB(1000, config));
+
+ config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, false);
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.2f);
+ assertEquals(800, TaskManagerServices.calculateHeapSizeMB(1000, config));
+
+ config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, true);
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "10m"); // 10MB
+ assertEquals(890, TaskManagerServices.calculateHeapSizeMB(1000, config));
+
+ config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0"); // use fraction of given memory
+ config.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION, 0.1f); // 10%
+ assertEquals(810, TaskManagerServices.calculateHeapSizeMB(1000, config));
+ }
+
+ /**
+ * Verifies that {@link NettyShuffleEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
+ * returns the correct result for old configurations via
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_NUM_BUFFERS}.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void hasNewNetworkBufConfOld() throws Exception {
+ Configuration config = new Configuration();
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
+
+ assertFalse(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+ }
+
+ /**
+ * Verifies that {@link NettyShuffleEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
+ * returns the correct result for new configurations via
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},
+ * {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN} and {@link
+ * NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}.
+ */
+ @Test
+ public void hasNewNetworkBufConfNew() throws Exception {
+ Configuration config = new Configuration();
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ // fully defined:
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "2048");
+
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ // partly defined:
+ config = new Configuration();
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ config = new Configuration();
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+ config.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ config = new Configuration();
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+ }
+
+ /**
+ * Verifies that {@link NettyShuffleEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
+ * returns the correct result for mixed old/new configurations.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void hasNewNetworkBufConfMixed() throws Exception {
+ Configuration config = new Configuration();
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
+ assertFalse(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config));
+
+ // old + 1 new parameter = new:
+ Configuration config1 = config.clone();
+ config1.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config1));
+
+ config1 = config.clone();
+ config1.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config1));
+
+ config1 = config.clone();
+ config1.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
+ assertTrue(NettyShuffleEnvironmentConfiguration.hasNewNetworkConfig(config1));
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkBufferCalculationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkBufferCalculationTest.java
index e6736fe..39b9d74 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkBufferCalculationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkBufferCalculationTest.java
@@ -21,8 +21,8 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.memory.MemoryType;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
@@ -35,8 +35,8 @@ import static org.junit.Assert.assertEquals;
public class NetworkBufferCalculationTest extends TestLogger {
/**
- * Test for {@link NetworkEnvironmentConfiguration#calculateNewNetworkBufferMemory(Configuration, long)}
- * using the same (manual) test cases as in {@link NetworkEnvironmentConfigurationTest#calculateHeapSizeMB()}.
+ * Test for {@link NettyShuffleEnvironmentConfiguration#calculateNewNetworkBufferMemory(Configuration, long)}
+ * using the same (manual) test cases as in {@link NettyShuffleEnvironmentConfigurationTest#calculateHeapSizeMB()}.
*/
@Test
public void calculateNetworkBufFromHeapSize() {
@@ -47,23 +47,23 @@ public class NetworkBufferCalculationTest extends TestLogger {
TaskManagerOptions.MANAGED_MEMORY_FRACTION.defaultValue(),
0.1f, 60L << 20, 1L << 30, MemoryType.HEAP);
assertEquals((100L << 20) + 1 /* one too many due to floating point imprecision */,
- NetworkEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 900L << 20)); // 900MB
+ NettyShuffleEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 900L << 20)); // 900MB
config = getConfig(
Long.valueOf(TaskManagerOptions.MANAGED_MEMORY_SIZE.defaultValue()),
TaskManagerOptions.MANAGED_MEMORY_FRACTION.defaultValue(),
0.2f, 60L << 20, 1L << 30, MemoryType.HEAP);
assertEquals((200L << 20) + 3 /* slightly too many due to floating point imprecision */,
- NetworkEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 800L << 20)); // 800MB
+ NettyShuffleEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 800L << 20)); // 800MB
config = getConfig(10, TaskManagerOptions.MANAGED_MEMORY_FRACTION.defaultValue(),
0.1f, 60L << 20, 1L << 30, MemoryType.OFF_HEAP);
assertEquals((100L << 20) + 1 /* one too many due to floating point imprecision */,
- NetworkEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 890L << 20)); // 890MB
+ NettyShuffleEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 890L << 20)); // 890MB
config = getConfig(0, 0.1f, 0.1f, 60L << 20, 1L << 30, MemoryType.OFF_HEAP);
assertEquals((100L << 20) + 1 /* one too many due to floating point imprecision */,
- NetworkEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 810L << 20)); // 810MB
+ NettyShuffleEnvironmentConfiguration.calculateNewNetworkBufferMemory(config, 810L << 20)); // 810MB
}
/**
@@ -71,9 +71,9 @@ public class NetworkBufferCalculationTest extends TestLogger {
*
* @param managedMemory see {@link TaskManagerOptions#MANAGED_MEMORY_SIZE}
* @param managedMemoryFraction see {@link TaskManagerOptions#MANAGED_MEMORY_FRACTION}
- * @param networkBufFraction see {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION}
- * @param networkBufMin see {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN}
- * @param networkBufMax see {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}
+ * @param networkBufFraction see {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION}
+ * @param networkBufMin see {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN}
+ * @param networkBufMax see {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}
* @param memoryType on-heap or off-heap
*
* @return configuration object
@@ -90,9 +90,9 @@ public class NetworkBufferCalculationTest extends TestLogger {
configuration.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE.key(), managedMemory);
configuration.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION.key(), managedMemoryFraction);
- configuration.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(), networkBufFraction);
- configuration.setLong(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(), networkBufMin);
- configuration.setLong(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(), networkBufMax);
+ configuration.setFloat(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key(), networkBufFraction);
+ configuration.setLong(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.key(), networkBufMin);
+ configuration.setLong(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.key(), networkBufMax);
configuration.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP.key(), memoryType == MemoryType.OFF_HEAP);
return configuration;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkEnvironmentConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkEnvironmentConfigurationTest.java
deleted file mode 100644
index 84b1238..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkEnvironmentConfigurationTest.java
+++ /dev/null
@@ -1,334 +0,0 @@
-/*
- * 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.runtime.taskexecutor;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.Test;
-
-import java.net.InetAddress;
-import java.util.Random;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/**
- * Unit test for {@link TaskManagerServices}.
- */
-public class NetworkEnvironmentConfigurationTest extends TestLogger {
-
- private static final long MEM_SIZE_PARAM = 128L * 1024 * 1024;
-
- /**
- * Verifies that {@link NetworkEnvironmentConfiguration#fromConfiguration(Configuration, long, boolean, InetAddress)}
- * returns the correct result for new configurations via
- * {@link NetworkEnvironmentOptions#NETWORK_REQUEST_BACKOFF_INITIAL},
- * {@link NetworkEnvironmentOptions#NETWORK_REQUEST_BACKOFF_MAX},
- * {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_PER_CHANNEL} and
- * {@link NetworkEnvironmentOptions#NETWORK_EXTRA_BUFFERS_PER_GATE}
- */
- @Test
- public void testNetworkRequestBackoffAndBuffers() {
-
- // set some non-default values
- final Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, 10);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100);
-
- final NetworkEnvironmentConfiguration networkConfig = NetworkEnvironmentConfiguration.fromConfiguration(
- config,
- MEM_SIZE_PARAM,
- true,
- InetAddress.getLoopbackAddress());
-
- assertEquals(networkConfig.partitionRequestInitialBackoff(), 100);
- assertEquals(networkConfig.partitionRequestMaxBackoff(), 200);
- assertEquals(networkConfig.networkBuffersPerChannel(), 10);
- assertEquals(networkConfig.floatingNetworkBuffersPerGate(), 100);
- }
-
- /**
- * Test for {@link NetworkEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using old
- * configurations via {@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS}.
- */
- @SuppressWarnings("deprecation")
- @Test
- public void calculateNetworkBufOld() {
- Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
-
- // note: actual network buffer memory size is independent of the totalJavaMemorySize
- assertEquals(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(10L << 20, config));
- assertEquals(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(64L << 20, config));
-
- // test integer overflow in the memory size
- int numBuffers = (int) ((2L << 32) / MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes()); // 2^33
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, numBuffers);
- assertEquals(2L << 32, NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(2L << 33, config));
- }
-
- /**
- * Test for {@link NetworkEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using new
- * configurations via {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},
- * {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN} and
- * {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}.
- */
- @Test
- public void calculateNetworkBufNew() throws Exception {
- Configuration config = new Configuration();
-
- // (1) defaults
- final Float defaultFrac = NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue();
- final Long defaultMin = MemorySize.parse(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes();
- final Long defaultMax = MemorySize.parse(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue()).getBytes();
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config));
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config));
-
- calculateNetworkBufNew(config);
- }
-
- /**
- * Helper to test {@link NetworkEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} with the
- * new configuration parameters.
- *
- * @param config configuration object
- */
- private static void calculateNetworkBufNew(final Configuration config) {
- // (2) fixed size memory
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(1L << 20)); // 1MB
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(1L << 20)); // 1MB
-
-
- // note: actual network buffer memory size is independent of the totalJavaMemorySize
- assertEquals(1 << 20, NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(10L << 20, config));
- assertEquals(1 << 20, NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(64L << 20, config));
- assertEquals(1 << 20, NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(1L << 30, config));
-
- // (3) random fraction, min, and max values
- Random ran = new Random();
- for (int i = 0; i < 1_000; ++i){
- float frac = Math.max(ran.nextFloat(), Float.MIN_VALUE);
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, frac);
-
- long min = Math.max(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(), ran.nextLong());
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(min));
-
-
- long max = Math.max(min, ran.nextLong());
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(max));
-
- long javaMem = Math.max(max + 1, ran.nextLong());
-
- final long networkBufMem = NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(javaMem, config);
-
- if (networkBufMem < min) {
- fail("Lower bound not met with configuration: " + config.toString());
- }
-
- if (networkBufMem > max) {
- fail("Upper bound not met with configuration: " + config.toString());
- }
-
- if (networkBufMem > min && networkBufMem < max) {
- if ((javaMem * frac) != networkBufMem) {
- fail("Wrong network buffer memory size with configuration: " + config.toString() +
- ". Expected value: " + (javaMem * frac) + " actual value: " + networkBufMem + '.');
- }
- }
- }
- }
-
- /**
- * Test for {@link NetworkEnvironmentConfiguration#calculateNetworkBufferMemory(long, Configuration)} using mixed
- * old/new configurations.
- */
- @SuppressWarnings("deprecation")
- @Test
- public void calculateNetworkBufMixed() throws Exception {
- Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
-
- final Float defaultFrac = NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue();
- final Long defaultMin = MemorySize.parse(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes();
- final Long defaultMax = MemorySize.parse(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue()).getBytes();
-
-
- // old + 1 new parameter = new:
- Configuration config1 = config.clone();
- config1.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- assertEquals(enforceBounds((long) (0.1f * (10L << 20)), defaultMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config1));
- assertEquals(enforceBounds((long) (0.1f * (10L << 30)), defaultMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
-
- config1 = config.clone();
- long newMin = MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes(); // smallest value possible
- config1.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(newMin));
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), newMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 20), config1));
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), newMin, defaultMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
-
- config1 = config.clone();
- long newMax = Math.max(64L << 20 + 1, MemorySize.parse(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()).getBytes());
- config1.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(newMax));
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, newMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((64L << 20 + 1), config1));
- assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, newMax),
- NetworkEnvironmentConfiguration.calculateNetworkBufferMemory((10L << 30), config1));
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config1));
-
- // old + any new parameter = new:
- calculateNetworkBufNew(config);
- }
-
- /**
- * Returns the value or the lower/upper bound in case the value is less/greater than the lower/upper bound, respectively.
- *
- * @param value value to inspect
- * @param lower lower bound
- * @param upper upper bound
- *
- * @return <tt>min(upper, max(lower, value))</tt>
- */
- private static long enforceBounds(final long value, final long lower, final long upper) {
- return Math.min(upper, Math.max(lower, value));
- }
-
- /**
- * Test for {@link TaskManagerServices#calculateHeapSizeMB(long, Configuration)} with some
- * manually calculated scenarios.
- */
- @Test
- public void calculateHeapSizeMB() throws Exception {
- Configuration config = new Configuration();
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(64L << 20)); // 64MB
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(1L << 30)); // 1GB
-
- config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, false);
- assertEquals(900, TaskManagerServices.calculateHeapSizeMB(1000, config));
-
- config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, false);
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.2f);
- assertEquals(800, TaskManagerServices.calculateHeapSizeMB(1000, config));
-
- config.setBoolean(TaskManagerOptions.MEMORY_OFF_HEAP, true);
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "10m"); // 10MB
- assertEquals(890, TaskManagerServices.calculateHeapSizeMB(1000, config));
-
- config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0"); // use fraction of given memory
- config.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION, 0.1f); // 10%
- assertEquals(810, TaskManagerServices.calculateHeapSizeMB(1000, config));
- }
-
- /**
- * Verifies that {@link NetworkEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
- * returns the correct result for old configurations via
- * {@link NetworkEnvironmentOptions#NETWORK_NUM_BUFFERS}.
- */
- @SuppressWarnings("deprecation")
- @Test
- public void hasNewNetworkBufConfOld() throws Exception {
- Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
-
- assertFalse(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
- }
-
- /**
- * Verifies that {@link NetworkEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
- * returns the correct result for new configurations via
- * {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_FRACTION},
- * {@link NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MIN} and {@link
- * NetworkEnvironmentOptions#NETWORK_BUFFERS_MEMORY_MAX}.
- */
- @Test
- public void hasNewNetworkBufConfNew() throws Exception {
- Configuration config = new Configuration();
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- // fully defined:
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "2048");
-
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- // partly defined:
- config = new Configuration();
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- config = new Configuration();
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
- config.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- config = new Configuration();
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
- }
-
- /**
- * Verifies that {@link NetworkEnvironmentConfiguration#hasNewNetworkConfig(Configuration)}
- * returns the correct result for mixed old/new configurations.
- */
- @SuppressWarnings("deprecation")
- @Test
- public void hasNewNetworkBufConfMixed() throws Exception {
- Configuration config = new Configuration();
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 1);
- assertFalse(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config));
-
- // old + 1 new parameter = new:
- Configuration config1 = config.clone();
- config1.setFloat(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config1));
-
- config1 = config.clone();
- config1.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config1));
-
- config1 = config.clone();
- config1.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, "1024");
- assertTrue(NetworkEnvironmentConfiguration.hasNewNetworkConfig(config1));
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
index 9673e13..5511d1b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -372,9 +372,9 @@ public class TaskExecutorSubmissionTest extends TestLogger {
public void testRemotePartitionNotFound() throws Exception {
final int dataPort = NetUtils.getAvailablePort();
Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.DATA_PORT, dataPort);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
+ config.setInteger(NettyShuffleEnvironmentOptions.DATA_PORT, dataPort);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
// Remote location (on the same TM though) for the partition
NettyShuffleDescriptor sdd =
@@ -464,8 +464,8 @@ public class TaskExecutorSubmissionTest extends TestLogger {
ExecutionAttemptID eid = tdd.getExecutionAttemptId();
Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>();
final CompletableFuture<Void> taskFailedFuture = new CompletableFuture<>();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index f1d607a..ff369e6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -57,8 +57,8 @@ import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -197,7 +197,7 @@ public class TaskExecutorTest extends TestLogger {
private SettableLeaderRetrievalService jobManagerLeaderRetriever;
- private NetworkEnvironment networkEnvironment;
+ private NettyShuffleEnvironment nettyShuffleEnvironment;
@Before
public void setup() throws IOException {
@@ -223,7 +223,7 @@ public class TaskExecutorTest extends TestLogger {
haServices.setResourceManagerLeaderRetriever(resourceManagerLeaderRetriever);
haServices.setJobMasterLeaderRetriever(jobId, jobManagerLeaderRetriever);
- networkEnvironment = new NetworkEnvironmentBuilder().build();
+ nettyShuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
}
@After
@@ -243,8 +243,8 @@ public class TaskExecutorTest extends TestLogger {
dummyBlobCacheService = null;
}
- if (networkEnvironment != null) {
- networkEnvironment.close();
+ if (nettyShuffleEnvironment != null) {
+ nettyShuffleEnvironment.close();
}
testingFatalErrorHandler.rethrowError();
@@ -270,7 +270,7 @@ public class TaskExecutorTest extends TestLogger {
MemoryType.HEAP,
false);
- networkEnvironment.start();
+ nettyShuffleEnvironment.start();
final KvStateService kvStateService = new KvStateService(new KvStateRegistry(), null, null);
kvStateService.start();
@@ -279,7 +279,7 @@ public class TaskExecutorTest extends TestLogger {
.setTaskManagerLocation(taskManagerLocation)
.setMemoryManager(memoryManager)
.setIoManager(ioManager)
- .setShuffleEnvironment(networkEnvironment)
+ .setShuffleEnvironment(nettyShuffleEnvironment)
.setKvStateService(kvStateService)
.setTaskSlotTable(taskSlotTable)
.setJobLeaderService(jobLeaderService)
@@ -295,7 +295,7 @@ public class TaskExecutorTest extends TestLogger {
}
assertThat(memoryManager.isShutdown(), is(true));
- assertThat(networkEnvironment.isClosed(), is(true));
+ assertThat(nettyShuffleEnvironment.isClosed(), is(true));
assertThat(ioManager.isProperlyShutDown(), is(true));
assertThat(kvStateService.isShutdown(), is(true));
}
@@ -709,7 +709,7 @@ public class TaskExecutorTest extends TestLogger {
final TaskExecutorLocalStateStoresManager localStateStoresManager = createTaskExecutorLocalStateStoresManager();
final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder()
- .setShuffleEnvironment(networkEnvironment)
+ .setShuffleEnvironment(nettyShuffleEnvironment)
.setTaskSlotTable(taskSlotTable)
.setJobManagerTable(jobManagerTable)
.setTaskStateManager(localStateStoresManager)
@@ -967,7 +967,7 @@ public class TaskExecutorTest extends TestLogger {
final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder()
.setTaskManagerLocation(taskManagerLocation)
- .setShuffleEnvironment(networkEnvironment)
+ .setShuffleEnvironment(nettyShuffleEnvironment)
.setTaskSlotTable(taskSlotTable)
.setJobLeaderService(jobLeaderService)
.setJobManagerTable(jobManagerTable)
@@ -1627,9 +1627,9 @@ public class TaskExecutorTest extends TestLogger {
public void testLogNotFoundHandling() throws Throwable {
final int dataPort = NetUtils.getAvailablePort();
Configuration config = new Configuration();
- config.setInteger(NetworkEnvironmentOptions.DATA_PORT, dataPort);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
- config.setInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
+ config.setInteger(NettyShuffleEnvironmentOptions.DATA_PORT, dataPort);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/i/dont/exist");
try (TaskSubmissionTestEnvironment env =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
index f48949d..58ef6a8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerStartupTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -163,7 +163,7 @@ public class TaskManagerRunnerStartupTest extends TestLogger {
try {
final Configuration cfg = new Configuration();
- cfg.setInteger(NetworkEnvironmentOptions.DATA_PORT, blocker.getLocalPort());
+ cfg.setInteger(NettyShuffleEnvironmentOptions.DATA_PORT, blocker.getLocalPort());
startTaskManager(
cfg,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
index 958696f..a802c71 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -34,7 +34,7 @@ import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
@@ -240,15 +240,15 @@ class TaskSubmissionTestEnvironment implements AutoCloseable {
shuffleEnvironment = mock(ShuffleEnvironment.class, Mockito.RETURNS_MOCKS);
} else {
final InetSocketAddress socketAddress = new InetSocketAddress(
- InetAddress.getByName(testingRpcService.getAddress()), configuration.getInteger(NetworkEnvironmentOptions.DATA_PORT));
+ InetAddress.getByName(testingRpcService.getAddress()), configuration.getInteger(NettyShuffleEnvironmentOptions.DATA_PORT));
final NettyConfig nettyConfig = new NettyConfig(socketAddress.getAddress(), socketAddress.getPort(),
ConfigurationParserUtils.getPageSize(configuration), ConfigurationParserUtils.getSlot(configuration), configuration);
- shuffleEnvironment = new NetworkEnvironmentBuilder()
+ shuffleEnvironment = new NettyShuffleEnvironmentBuilder()
.setTaskManagerLocation(taskManagerLocation)
- .setPartitionRequestInitialBackoff(configuration.getInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL))
- .setPartitionRequestMaxBackoff(configuration.getInteger(NetworkEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX))
+ .setPartitionRequestInitialBackoff(configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL))
+ .setPartitionRequestMaxBackoff(configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX))
.setNettyConfig(localCommunication ? null : nettyConfig)
.build();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
index bfcbd13..27e1f7e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -40,7 +40,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.taskexecutor.PartitionProducerStateChecker;
@@ -117,7 +117,7 @@ public class TaskAsyncCallTest extends TestLogger {
notifyCheckpointCompleteLatch = new OneShotLatch();
stopLatch = new OneShotLatch();
- shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
classLoaders.clear();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
index 33d89e5..7696976 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.taskmanager;
import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.execution.Environment;
@@ -74,7 +74,7 @@ public class TaskCancelAsyncProducerConsumerITCase extends TestLogger {
private static Configuration getFlinkConfiguration() {
Configuration config = new Configuration();
config.setString(TaskManagerOptions.MEMORY_SEGMENT_SIZE, "4096");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 9);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 9);
return config;
}
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 a1b73df..9a1b874 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
@@ -46,7 +46,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
@@ -134,7 +134,7 @@ public class TaskTest extends TestLogger {
awaitLatch = new OneShotLatch();
triggerLatch = new OneShotLatch();
- shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
}
@After
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
index 99db686..6f39813 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
@@ -43,7 +43,7 @@ import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.taskexecutor.PartitionProducerStateChecker;
@@ -168,7 +168,7 @@ public class JvmExitOnFatalErrorTest {
final MemoryManager memoryManager = new MemoryManager(1024 * 1024, 1);
final IOManager ioManager = new IOManagerAsync();
- final ShuffleEnvironment shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ final ShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
final TaskManagerRuntimeInfo tmInfo = TaskManagerConfiguration.fromConfiguration(taskManagerConfig);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
index da401a7..b77c7d0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
@@ -20,7 +20,7 @@ package org.apache.flink.streaming.runtime.io;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
@@ -53,7 +53,7 @@ public class InputProcessorUtil {
+ " must be positive or -1 (infinite)");
}
- if (taskManagerConfig.getBoolean(NetworkEnvironmentOptions.NETWORK_CREDIT_MODEL)) {
+ if (taskManagerConfig.getBoolean(NettyShuffleEnvironmentOptions.NETWORK_CREDIT_MODEL)) {
barrierHandler = new BarrierBuffer(
inputGate,
new CachedBufferBlocker(inputGate.getPageSize()),
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
index 8f50198..0c399f7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
@@ -26,8 +26,8 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironment;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.api.writer.RecordWriterBuilder;
@@ -79,8 +79,8 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
protected final JobID jobId = new JobID();
protected final IntermediateDataSetID dataSetID = new IntermediateDataSetID();
- protected NetworkEnvironment senderEnv;
- protected NetworkEnvironment receiverEnv;
+ protected NettyShuffleEnvironment senderEnv;
+ protected NettyShuffleEnvironment receiverEnv;
protected IOManager ioManager;
protected int channels;
@@ -144,13 +144,13 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
ioManager = new IOManagerAsync();
- senderEnv = createNettyNetworkEnvironment(senderBufferPoolSize, config);
+ senderEnv = createShuffleEnvironment(senderBufferPoolSize, config);
this.dataPort = senderEnv.start();
if (localMode && senderBufferPoolSize == receiverBufferPoolSize) {
receiverEnv = senderEnv;
}
else {
- receiverEnv = createNettyNetworkEnvironment(receiverBufferPoolSize, config);
+ receiverEnv = createShuffleEnvironment(receiverBufferPoolSize, config);
receiverEnv.start();
}
@@ -196,7 +196,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
}
}
- private NetworkEnvironment createNettyNetworkEnvironment(
+ private NettyShuffleEnvironment createShuffleEnvironment(
@SuppressWarnings("SameParameterValue") int bufferPoolSize, Configuration config) throws Exception {
final NettyConfig nettyConfig = new NettyConfig(
@@ -206,7 +206,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
// please note that the number of slots directly influences the number of netty threads!
ConfigurationParserUtils.getSlot(config),
config);
- return new NetworkEnvironmentBuilder()
+ return new NettyShuffleEnvironmentBuilder()
.setNumNetworkBuffers(bufferPoolSize)
.setNettyConfig(nettyConfig)
.build();
@@ -215,7 +215,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
protected ResultPartitionWriter createResultPartition(
JobID jobId,
ResultPartitionID partitionId,
- NetworkEnvironment environment,
+ NettyShuffleEnvironment environment,
int channels) throws Exception {
ResultPartitionWriter resultPartitionWriter = new ResultPartitionBuilder()
@@ -224,7 +224,7 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
.setNumberOfSubpartitions(channels)
.setResultPartitionManager(environment.getResultPartitionManager())
.setIOManager(ioManager)
- .setupBufferPoolFactoryFromNetworkEnvironment(environment)
+ .setupBufferPoolFactoryFromNettyShuffleEnvironment(environment)
.build();
ResultPartitionWriter consumableNotifyingPartitionWriter = new ConsumableNotifyingResultPartitionWriterDecorator(
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
index a15aca4..b311060 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.streaming.runtime.io.benchmark;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.junit.Rule;
import org.junit.Test;
@@ -82,7 +82,7 @@ public class StreamNetworkThroughputBenchmarkTest {
expectedException.expect(IOException.class);
expectedException.expectMessage("Insufficient number of network buffers");
- env.setUp(writers, channels, 100, false, writers * channels - 1, writers * channels * NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue());
+ env.setUp(writers, channels, 100, false, writers * channels - 1, writers * channels * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue());
}
@Test
@@ -94,7 +94,7 @@ public class StreamNetworkThroughputBenchmarkTest {
expectedException.expect(IOException.class);
expectedException.expectMessage("Insufficient number of network buffers");
- env.setUp(writers, channels, 100, false, writers * channels, writers * channels * NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() - 1);
+ env.setUp(writers, channels, 100, false, writers * channels, writers * channels * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue() - 1);
}
@Test
@@ -104,7 +104,7 @@ public class StreamNetworkThroughputBenchmarkTest {
int channels = 2;
env.setUp(writers, channels, 100, false, writers * channels, writers * channels *
- NetworkEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue());
+ NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue());
env.executeBenchmark(10_000);
env.tearDown();
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
index e0f3b52..62ab1d4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
@@ -43,7 +43,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -181,7 +181,7 @@ public class InterruptSensitiveRestoreTest {
StreamStateHandle state,
int mode) throws IOException {
- ShuffleEnvironment shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ ShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
Collection<KeyedStateHandle> keyedStateFromBackend = Collections.emptyList();
Collection<KeyedStateHandle> keyedStateFromStream = Collections.emptyList();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index 64b55fc..e92ba9c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -43,7 +43,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -150,7 +150,7 @@ public class StreamTaskTerminationTest extends TestLogger {
final TaskManagerRuntimeInfo taskManagerRuntimeInfo = new TestingTaskManagerRuntimeInfo();
- final ShuffleEnvironment shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ final ShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
BlobCacheService blobService =
new BlobCacheService(mock(PermanentBlobCache.class), mock(TransientBlobCache.class));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index c4d5965..f1fcfca 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -49,7 +49,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
@@ -902,7 +902,7 @@ public class StreamTaskTest extends TestLogger {
PartitionProducerStateChecker partitionProducerStateChecker = mock(PartitionProducerStateChecker.class);
Executor executor = mock(Executor.class);
- ShuffleEnvironment shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ ShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
JobInformation jobInformation = new JobInformation(
new JobID(),
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
index 5dbd5ce..ab3bcf8 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
@@ -42,7 +42,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
@@ -263,7 +263,7 @@ public class SynchronousCheckpointITCase {
ResultPartitionConsumableNotifier consumableNotifier = new NoOpResultPartitionConsumableNotifier();
PartitionProducerStateChecker partitionProducerStateChecker = mock(PartitionProducerStateChecker.class);
Executor executor = mock(Executor.class);
- ShuffleEnvironment shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ ShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
TaskMetricGroup taskMetricGroup = UnregisteredMetricGroups.createUnregisteredTaskMetricGroup();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
index 9f91a00..c02ff99 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
@@ -45,7 +45,7 @@ import org.apache.flink.runtime.executiongraph.JobInformation;
import org.apache.flink.runtime.executiongraph.TaskInformation;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.io.network.NetworkEnvironmentBuilder;
+import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -224,7 +224,7 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
TestStreamTask.class.getName(),
taskConfig);
- ShuffleEnvironment<?, ?> shuffleEnvironment = new NetworkEnvironmentBuilder().build();
+ ShuffleEnvironment<?, ?> shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
BlobCacheService blobService =
new BlobCacheService(mock(PermanentBlobCache.class), mock(TransientBlobCache.class));
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
index 08a5e1a..ab98e92 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
@@ -24,7 +24,7 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
@@ -80,7 +80,7 @@ public abstract class CancelingTestBase extends TestLogger {
config.setBoolean(CoreOptions.FILESYTEM_DEFAULT_OVERRIDE, true);
config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
config.setString(TaskManagerOptions.MEMORY_SEGMENT_SIZE, "4096");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 2048);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 2048);
return config;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
index 99c5a75..818eb13 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
@@ -28,7 +28,7 @@ import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.contrib.streaming.state.RocksDBOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
@@ -211,7 +211,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
Configuration config = new Configuration();
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "48m");
// the default network buffers size (10% of heap max =~ 150MB) seems to much for this test case
- config.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(80L << 20)); // 80 MB
+ config.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(80L << 20)); // 80 MB
config.setString(AkkaOptions.FRAMESIZE, String.valueOf(MAX_MEM_STATE_SIZE) + "b");
if (zkServer != null) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java b/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java
index 95aab44..38902c7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java
@@ -21,7 +21,7 @@ package org.apache.flink.test.manual;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.CheckpointingMode;
@@ -51,7 +51,7 @@ public class StreamingScalabilityAndLatency {
try {
Configuration config = new Configuration();
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "80m");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 20000);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 20000);
config.setInteger("taskmanager.net.server.numThreads", 1);
config.setInteger("taskmanager.net.client.numThreads", 1);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
index 642f601..7cef448 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.operators.DeltaIteration;
import org.apache.flink.api.java.operators.IterativeDataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.examples.java.clustering.KMeans;
import org.apache.flink.examples.java.clustering.util.KMeansData;
@@ -63,7 +63,7 @@ public class SuccessAfterNetworkBuffersFailureITCase extends TestLogger {
private static Configuration getConfiguration() {
Configuration config = new Configuration();
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "80m");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 800);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 800);
return config;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
index a8f6ffc..1677a88 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
@@ -24,7 +24,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HeartbeatManagerOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -103,7 +103,7 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test
config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, temporaryFolder.newFolder().getAbsolutePath());
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "4m");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
try (final StandaloneSessionClusterEntrypoint clusterEntrypoint = new StandaloneSessionClusterEntrypoint(config)) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
index 2ff3f59..39ee5d9 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
@@ -29,7 +29,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.FutureUtils;
@@ -251,7 +251,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
zooKeeper.getConnectString(), zookeeperStoragePath.getPath());
// Task manager configuration
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "4m");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
final RpcService rpcService = AkkaRpcServiceUtils.createRpcService("localhost", 0, config);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
index e4dd1f1..933d630 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
@@ -32,7 +32,7 @@ import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.client.JobStatusMessage;
@@ -115,7 +115,7 @@ public class ProcessFailureCancelingITCase extends TestLogger {
config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, temporaryFolder.newFolder().getAbsolutePath());
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "4m");
- config.setInteger(NetworkEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
+ config.setInteger(NettyShuffleEnvironmentOptions.NETWORK_NUM_BUFFERS, 100);
config.setInteger(RestOptions.PORT, 0);
final RpcService rpcService = AkkaRpcServiceUtils.createRpcService("localhost", 0, config);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
index 05fd914..1657d8b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
@@ -20,7 +20,7 @@ package org.apache.flink.test.runtime;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -73,7 +73,7 @@ public class NettyEpollITCase extends TestLogger {
private MiniClusterWithClientResource trySetUpCluster() throws Exception {
try {
Configuration config = new Configuration();
- config.setString(NetworkEnvironmentOptions.TRANSPORT_TYPE, "epoll");
+ config.setString(NettyShuffleEnvironmentOptions.TRANSPORT_TYPE, "epoll");
MiniClusterWithClientResource cluster = new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(config)
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 140b8c8..0c93748 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -35,7 +35,7 @@ import org.apache.flink.runtime.rest.messages.ClusterOverviewHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersInfo;
-import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
+import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.test.testdata.WordCountData;
import org.apache.flink.util.ExceptionUtils;
@@ -206,7 +206,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
// set memory constraints (otherwise this is the same test as perJobYarnCluster() above)
final long taskManagerMemoryMB = 1024;
//noinspection NumericOverflow if the calculation of the total Java memory size overflows, default configuration parameters are wrong in the first place, so we can ignore this inspection
- final long networkBuffersMB = NetworkEnvironmentConfiguration.calculateNetworkBufferMemory(
+ final long networkBuffersMB = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
(taskManagerMemoryMB - ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN.defaultValue()) << 20,
new Configuration()) >> 20;
final long offHeapMemory = taskManagerMemoryMB
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index 28653b6..86b0052 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -25,7 +25,7 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NetworkEnvironmentOptions;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -88,8 +88,8 @@ public class YarnConfigurationITCase extends YarnTestBase {
// disable heap cutoff min
configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
- configuration.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(1L << 20));
- configuration.setString(NetworkEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(4L << 20));
+ configuration.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MIN, String.valueOf(1L << 20));
+ configuration.setString(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_MEMORY_MAX, String.valueOf(4L << 20));
final YarnConfiguration yarnConfiguration = getYarnConfiguration();
final YarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(