[GitHub] tillrohrmann commented on a change in pull request #7595: [FLINK-11363][tests] Port TaskManagerConfigurationTest to new code base

[GitBox <gi...@apache.org>]

tillrohrmann commented on a change in pull request #7595: [FLINK-11363][tests] Port TaskManagerConfigurationTest to new code base
URL: https://github.com/apache/flink/pull/7595#discussion_r252242716
 
 

 ##########
 File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerConfigurationTest.java
 ##########
 @@ -0,0 +1,186 @@
+/*
+ * 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.GlobalConfiguration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.concurrent.Executors;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.IOUtils;
+
+import net.jcip.annotations.NotThreadSafe;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.isEmptyOrNullString;
+import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeNoException;
+
+/**
+ * Validates that the TaskManagerRunner startup properly obeys the configuration
+ * values.
+ *
+ * <p>NOTE: at least {@link #testDefaultFsParameterLoading()} should not be run in parallel to other
+ * tests in the same JVM as it modifies a static (private) member of the {@link FileSystem} class
+ * and verifies its content.
+ */
+@NotThreadSafe
+public class TaskManagerRunnerConfigurationTest {
+
+	private static final int TEST_TIMEOUT_SECONDS = 10;
+
+	@Rule
+	public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	@Test
+	public void testTaskManagerRpcServiceShouldBindToConfiguredTaskManagerHostname() throws Exception {
+		final String taskmanagerHost = "testhostname";
+		final Configuration config = createFlinkConfigWithPredefinedTaskManagerHostname(taskmanagerHost);
+		final HighAvailabilityServices highAvailabilityServices = createHighAvailabilityServices(config);
+
+		RpcService taskManagerRpcService = null;
+		try {
+			taskManagerRpcService = TaskManagerRunner.createRpcService(config, highAvailabilityServices);
+
+			assertThat(taskManagerRpcService.getPort(), is(greaterThanOrEqualTo(0)));
+			assertThat(taskManagerRpcService.getAddress(), is(equalTo(taskmanagerHost)));
+		} finally {
+			maybeCloseRpcService(taskManagerRpcService);
+			highAvailabilityServices.closeAndCleanupAllData();
+		}
+	}
+
+	@Test
+	public void testTaskManagerRpcServiceShouldBindToAddressDeterminedByConnectingToResourceManager() throws Exception {
+		final ServerSocket testJobManagerSocket = openServerSocket();
+		final Configuration config = createFlinkConfigWithJobManagerPort(testJobManagerSocket.getLocalPort());
+		final HighAvailabilityServices highAvailabilityServices = createHighAvailabilityServices(config);
+
+		RpcService taskManagerRpcService = null;
+		try {
+			taskManagerRpcService = TaskManagerRunner.createRpcService(config, highAvailabilityServices);
+			assertThat(taskManagerRpcService.getAddress(), not(isEmptyOrNullString()));
+		} finally {
+			maybeCloseRpcService(taskManagerRpcService);
+			highAvailabilityServices.closeAndCleanupAllData();
+			IOUtils.closeQuietly(testJobManagerSocket);
+		}
+	}
+
+	@Test
+	public void testCreatingTaskManagerRpcServiceShouldFailIfRpcPortRangeIsInvalid() throws Exception {
+		final Configuration config = new Configuration(createFlinkConfigWithPredefinedTaskManagerHostname("example.org"));
+		config.setString(TaskManagerOptions.RPC_PORT, "-1");
+
+		final HighAvailabilityServices highAvailabilityServices = createHighAvailabilityServices(config);
+
+		try {
+			TaskManagerRunner.createRpcService(config, highAvailabilityServices);
+			fail("Should fail because -1 is not a valid port range");
+		} catch (final IllegalArgumentException e) {
+			assertThat(e.getMessage(),  containsString("Invalid port range definition: -1"));
+		} finally {
+			highAvailabilityServices.closeAndCleanupAllData();
+		}
+	}
+
+	@Test
+	public void testDefaultFsParameterLoading() throws Exception {
+		try {
+			final File tmpDir = temporaryFolder.newFolder();
+			final File confFile = new File(tmpDir, GlobalConfiguration.FLINK_CONF_FILENAME);
+
+			final URI defaultFS = new URI("otherFS", null, "localhost", 1234, null, null, null);
+
+			final PrintWriter pw1 = new PrintWriter(confFile);
+			pw1.println("fs.default-scheme: " + defaultFS);
+			pw1.close();
+
+			String[] args = new String[] {"--configDir", tmpDir.toString()};
+			Configuration configuration = TaskManagerRunner.loadConfiguration(args);
+			FileSystem.initialize(configuration);
+
+			assertEquals(defaultFS, FileSystem.getDefaultFsUri());
+		}
+		finally {
+			// reset FS settings
+			FileSystem.initialize(new Configuration());
+		}
+	}
+
+	private static Configuration createFlinkConfigWithPredefinedTaskManagerHostname(
+			final String taskmanagerHost) {
+		final Configuration config = new Configuration();
+		config.setString(TaskManagerOptions.HOST, taskmanagerHost);
+		config.setString(JobManagerOptions.ADDRESS, "localhost");
+		return new UnmodifiableConfiguration(config);
+	}
+
+	private static Configuration createFlinkConfigWithJobManagerPort(final int port) {
+		Configuration config = new Configuration();
+		config.setString(JobManagerOptions.ADDRESS, "localhost");
+		config.setInteger(JobManagerOptions.PORT, port);
+		return config;
 
 Review comment:
   Maybe this could be symmetric to `createFlinkConfigWithPredefinedTaskManagerHostname` and also return an `UnmodifiableConfiguration`.

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


With regards,
Apache Git Services