[hotfix][network] Remove unused IOMode from NetworkEnvironmentConfiguration

Author: zhijiangW

flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java

@@ -284,9 +284,9 @@ public final class ConfigConstants {
 	public static final String TASK_MANAGER_MEMORY_SEGMENT_SIZE_KEY = "taskmanager.memory.segment-size";
 
 	/**
-	 * The implementation to use for spillable/spilled intermediate results, which have both
-	 * synchronous and asynchronous implementations: "sync" or "async".
+	 * @deprecated Not used anymore
 	 */
+	@Deprecated
 	public static final String TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "taskmanager.network.defaultIOMode";
 
 	/**
@@ -1435,9 +1435,9 @@ public final class ConfigConstants {
 	public static final int DEFAULT_TASK_MANAGER_MEMORY_SEGMENT_SIZE = 32768;
 
 	/**
-	 * The implementation to use for spillable/spilled intermediate results, which have both
-	 * synchronous and asynchronous implementations: "sync" or "async".
+	 * @deprecated Not used anymore
 	 */
+	@Deprecated
 	public static final String DEFAULT_TASK_MANAGER_NETWORK_DEFAULT_IO_MODE = "sync";
 
 	/**

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java

@@ -36,22 +36,6 @@
  * The facade for the provided I/O manager services.
  */
 public abstract class IOManager {
-
-	public enum IOMode {
-
-		SYNC(true), ASYNC(false);
-
-		private final boolean isSynchronous;
-
-		IOMode(boolean isSynchronous) {
-			this.isSynchronous = isSynchronous;
-		}
-
-		public boolean isSynchronous() {
-			return isSynchronous;
-		}
-	}
-
 	/** Logging */
 	protected static final Logger LOG = LoggerFactory.getLogger(IOManager.class);
 

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java

@@ -20,7 +20,6 @@
 
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.CheckpointingOptions;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.IllegalConfigurationException;
@@ -29,7 +28,6 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.memory.MemoryType;
 import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
 import org.apache.flink.runtime.io.network.netty.NettyConfig;
 import org.apache.flink.runtime.memory.MemoryManager;
 import org.apache.flink.runtime.registration.RetryingRegistrationConfiguration;
@@ -379,18 +377,6 @@ private static NetworkEnvironmentConfiguration parseNetworkEnvironmentConfigurat
 			nettyConfig = null;
 		}
 
-		// Default spill I/O mode for intermediate results
-		final String syncOrAsync = configuration.getString(
-			ConfigConstants.TASK_MANAGER_NETWORK_DEFAULT_IO_MODE,
-			ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_DEFAULT_IO_MODE);
-
-		final IOManager.IOMode ioMode;
-		if (syncOrAsync.equals("async")) {
-			ioMode = IOManager.IOMode.ASYNC;
-		} else {
-			ioMode = IOManager.IOMode.SYNC;
-		}
-
 		int initialRequestBackoff = configuration.getInteger(
 			TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL);
 		int maxRequestBackoff = configuration.getInteger(
@@ -406,7 +392,6 @@ private static NetworkEnvironmentConfiguration parseNetworkEnvironmentConfigurat
 			networkBufMin,
 			networkBufMax,
 			pageSize,
-			ioMode,
 			initialRequestBackoff,
 			maxRequestBackoff,
 			buffersPerChannel,

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.taskmanager;
 
-import org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode;
 import org.apache.flink.runtime.io.network.netty.NettyConfig;
 
 import javax.annotation.Nullable;
@@ -36,8 +35,6 @@ public class NetworkEnvironmentConfiguration {
 
 	private final int networkBufferSize;
 
-	private final IOMode ioMode;
-
 	private final int partitionRequestInitialBackoff;
 
 	private final int partitionRequestMaxBackoff;
@@ -48,34 +45,11 @@ public class NetworkEnvironmentConfiguration {
 
 	private final NettyConfig nettyConfig;
 
-	/**
-	 * Constructor for a setup with purely local communication (no netty).
-	 */
 	public NetworkEnvironmentConfiguration(
 			float networkBufFraction,
 			long networkBufMin,
 			long networkBufMax,
 			int networkBufferSize,
-			IOMode ioMode,
-			int partitionRequestInitialBackoff,
-			int partitionRequestMaxBackoff,
-			int networkBuffersPerChannel,
-			int floatingNetworkBuffersPerGate) {
-
-		this(networkBufFraction, networkBufMin, networkBufMax, networkBufferSize,
-				ioMode,
-				partitionRequestInitialBackoff, partitionRequestMaxBackoff,
-				networkBuffersPerChannel, floatingNetworkBuffersPerGate,
-				null);
-		
-	}
-
-	public NetworkEnvironmentConfiguration(
-			float networkBufFraction,
-			long networkBufMin,
-			long networkBufMax,
-			int networkBufferSize,
-			IOMode ioMode,
 			int partitionRequestInitialBackoff,
 			int partitionRequestMaxBackoff,
 			int networkBuffersPerChannel,
@@ -86,7 +60,6 @@ public NetworkEnvironmentConfiguration(
 		this.networkBufMin = networkBufMin;
 		this.networkBufMax = networkBufMax;
 		this.networkBufferSize = networkBufferSize;
-		this.ioMode = ioMode;
 		this.partitionRequestInitialBackoff = partitionRequestInitialBackoff;
 		this.partitionRequestMaxBackoff = partitionRequestMaxBackoff;
 		this.networkBuffersPerChannel = networkBuffersPerChannel;
@@ -112,10 +85,6 @@ public int networkBufferSize() {
 		return networkBufferSize;
 	}
 
-	public IOMode ioMode() {
-		return ioMode;
-	}
-
 	public int partitionRequestInitialBackoff() {
 		return partitionRequestInitialBackoff;
 	}
@@ -142,7 +111,6 @@ public NettyConfig nettyConfig() {
 	public int hashCode() {
 		int result = 1;
 		result = 31 * result + networkBufferSize;
-		result = 31 * result + ioMode.hashCode();
 		result = 31 * result + partitionRequestInitialBackoff;
 		result = 31 * result + partitionRequestMaxBackoff;
 		result = 31 * result + networkBuffersPerChannel;
@@ -170,7 +138,6 @@ else if (obj == null || getClass() != obj.getClass()) {
 					this.partitionRequestMaxBackoff == that.partitionRequestMaxBackoff &&
 					this.networkBuffersPerChannel == that.networkBuffersPerChannel &&
 					this.floatingNetworkBuffersPerGate == that.floatingNetworkBuffersPerGate &&
-					this.ioMode == that.ioMode && 
 					(nettyConfig != null ? nettyConfig.equals(that.nettyConfig) : that.nettyConfig == null);
 		}
 	}
@@ -182,7 +149,6 @@ public String toString() {
 				", networkBufMin=" + networkBufMin +
 				", networkBufMax=" + networkBufMax +
 				", networkBufferSize=" + networkBufferSize +
-				", ioMode=" + ioMode +
 				", partitionRequestInitialBackoff=" + partitionRequestInitialBackoff +
 				", partitionRequestMaxBackoff=" + partitionRequestMaxBackoff +
 				", networkBuffersPerChannel=" + networkBuffersPerChannel +

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NetworkBufferCalculationTest.java

@@ -91,7 +91,6 @@ private static TaskManagerServicesConfiguration getTmConfig(
 			networkBufMin,
 			networkBufMax,
 			checkedDownCast(MemorySize.parse(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()).getBytes()),
-			null,
 			TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL.defaultValue(),
 			TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX.defaultValue(),
 			TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue(),