flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/2] flink git commit: [FLINK-8198] [core] Fix condition for parsing ConnectionLimitingSettings
Date Tue, 05 Dec 2017 12:13:01 GMT
[FLINK-8198] [core] Fix condition for parsing ConnectionLimitingSettings


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/15514985
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/15514985
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/15514985

Branch: refs/heads/release-1.4
Commit: 155149855ba39582ffe7507bd808259dbe33ce81
Parents: 95ac289
Author: Stephan Ewen <sewen@apache.org>
Authored: Tue Dec 5 12:21:08 2017 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Tue Dec 5 12:32:41 2017 +0100

----------------------------------------------------------------------
 .../core/fs/LimitedConnectionsFileSystem.java   |  2 +-
 .../fs/LimitedConnectionsConfigurationTest.java | 56 ++++++++++++++++++++
 2 files changed, 57 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/15514985/flink-core/src/main/java/org/apache/flink/core/fs/LimitedConnectionsFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/LimitedConnectionsFileSystem.java
b/flink-core/src/main/java/org/apache/flink/core/fs/LimitedConnectionsFileSystem.java
index 5353563..fdf54e0 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/LimitedConnectionsFileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/LimitedConnectionsFileSystem.java
@@ -1074,7 +1074,7 @@ public class LimitedConnectionsFileSystem extends FileSystem {
 			checkLimit(limitOut, limitOutOption);
 
 			// create the settings only, if at least one limit is configured
-			if (totalLimit <= 0 || limitIn <= 0 || limitOut <= 0) {
+			if (totalLimit <= 0 && limitIn <= 0 && limitOut <= 0) {
 				// no limit configured
 				return null;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/15514985/flink-core/src/test/java/org/apache/flink/core/fs/LimitedConnectionsConfigurationTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/fs/LimitedConnectionsConfigurationTest.java
b/flink-core/src/test/java/org/apache/flink/core/fs/LimitedConnectionsConfigurationTest.java
index 4742a7e..2c30ce8 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/LimitedConnectionsConfigurationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/LimitedConnectionsConfigurationTest.java
@@ -19,6 +19,8 @@
 package org.apache.flink.core.fs;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.core.fs.LimitedConnectionsFileSystem.ConnectionLimitingSettings;
 import org.apache.flink.testutils.TestFileSystem;
 
 import org.junit.Rule;
@@ -29,6 +31,8 @@ import java.net.URI;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -40,6 +44,10 @@ public class LimitedConnectionsConfigurationTest {
 	@Rule
 	public final TemporaryFolder tempDir = new TemporaryFolder();
 
+	/**
+	 * This test validates that the File System is correctly wrapped by the
+	 * file system factories when the corresponding entries are in the configuration.
+	 */
 	@Test
 	public void testConfiguration() throws Exception {
 		final String fsScheme = TestFileSystem.SCHEME;
@@ -81,4 +89,52 @@ public class LimitedConnectionsConfigurationTest {
 			FileSystem.initialize(new Configuration());
 		}
 	}
+
+	/**
+	 * This test checks that the file system connection limiting configuration object
+	 * is properly created.
+	 */
+	@Test
+	public void testConnectionLimitingSettings() {
+		final String scheme = "testscheme";
+
+		// empty config
+		assertNull(ConnectionLimitingSettings.fromConfig(new Configuration(), scheme));
+
+		// only total limit set
+		{
+			Configuration conf = new Configuration();
+			conf.setInteger(CoreOptions.fileSystemConnectionLimit(scheme), 10);
+
+			ConnectionLimitingSettings settings = ConnectionLimitingSettings.fromConfig(conf, scheme);
+			assertNotNull(settings);
+			assertEquals(10, settings.limitTotal);
+			assertEquals(0, settings.limitInput);
+			assertEquals(0, settings.limitOutput);
+		}
+
+		// only input limit set
+		{
+			Configuration conf = new Configuration();
+			conf.setInteger(CoreOptions.fileSystemConnectionLimitIn(scheme), 10);
+
+			ConnectionLimitingSettings settings = ConnectionLimitingSettings.fromConfig(conf, scheme);
+			assertNotNull(settings);
+			assertEquals(0, settings.limitTotal);
+			assertEquals(10, settings.limitInput);
+			assertEquals(0, settings.limitOutput);
+		}
+
+		// only output limit set
+		{
+			Configuration conf = new Configuration();
+			conf.setInteger(CoreOptions.fileSystemConnectionLimitOut(scheme), 10);
+
+			ConnectionLimitingSettings settings = ConnectionLimitingSettings.fromConfig(conf, scheme);
+			assertNotNull(settings);
+			assertEquals(0, settings.limitTotal);
+			assertEquals(0, settings.limitInput);
+			assertEquals(10, settings.limitOutput);
+		}
+	}
 }


Mime
View raw message