[GitHub] [flink] tillrohrmann commented on a change in pull request #13706: [FLINK-19677][runtime] Make JobManager lazily resolve hostname of TaskManager and provide an option to turn off reverse resolution entirely

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

tillrohrmann commented on a change in pull request #13706:
URL: https://github.com/apache/flink/pull/13706#discussion_r509330828



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerLocation.java
##########
@@ -150,28 +165,48 @@ public String addressString() {
 
 	/**
 	 * Returns the fully-qualified domain name the TaskManager. If the name could not be
-	 * determined, the return value will be a textual representation of the TaskManager's IP address.
+	 * determined or {@link #retrieveHostName} is set to false,
+	 * the return value will be a textual representation of the TaskManager's IP address.
 	 * 
 	 * @return The fully-qualified domain name of the TaskManager.
 	 */
 	public String getFQDNHostname() {
+		// Lazily retrieve FQDN host name of this TaskManager
+		if (fqdnHostName == null) {
+			if (!retrieveHostName) {
+				fqdnHostName = inetAddress.getHostAddress();
+			} else {
+				fqdnHostName = getFqdnHostName(inetAddress);
+			}
+		}

Review comment:
       I am wondering whether it wouldn't make sense to move the resolution logic out of the `TaskManagerLocation` class in order to make more like a simple data transfer object. For example, we could have a method `fromUnresolvedLocation(UnresolvedTaskManagerLocation, ResolutionMode)` which either resolves the hostname or not depending on `ResolutionMode`.
   
   As a second step we could think about whether we want to support laziness. If this is the case, then we could provide a `Supplier<String> fqdnHostNameSupplier` which we call to give us the fqdn hostname and then store it some field.
   
   Separating the `TaskManagerLocation` from the way it is resolved might simplify the individual classes a bit.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
##########
@@ -916,6 +918,56 @@ public void testCheckpointPrecedesSavepointRecovery() throws Exception {
 		}
 	}
 
+	/**
+	 * Tests that the JobMaster does not retrieve TaskManager host name during registration,
+	 * so that no reverse DNS lookups would be performed.
+	 */
+	@Test
+	public void testDoNotRetrieveTaskManagerHostName() throws Exception {
+		final JobGraph restartingJobGraph = createSingleVertexJobWithRestartStrategy();
+
+		configuration.setBoolean(JobManagerOptions.RETRIEVE_TASK_MANAGER_HOSTNAME, false);
+
+		final JobMaster jobMaster = createJobMaster(
+				configuration,
+				restartingJobGraph,
+				haServices,
+				new TestingJobManagerSharedServicesBuilder().build(),
+				heartbeatServices);
+
+		final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
+		final TestingTaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGatewayBuilder()
+				.setSubmitTaskConsumer((tdd, ignored) -> CompletableFuture.completedFuture(Acknowledge.get()))
+				.setAddress("127.0.0.1:42")
+				.createTestingTaskExecutorGateway();
+
+		rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);
+
+		try {
+			jobMaster.start(JobMasterId.generate()).get();
+
+			final LocalUnresolvedTaskManagerLocation taskManagerUnresolvedLocation = new LocalUnresolvedTaskManagerLocation();
+
+			RegistrationResponse registrationResult = jobMasterGateway.registerTaskManager(taskExecutorGateway.getAddress(), taskManagerUnresolvedLocation, testingTimeout).get();
+			Map<ResourceID, Tuple2<TaskManagerLocation, TaskExecutorGateway>> registeredTaskManagers = jobMaster.getRegisteredTaskManagers();
+
+			assertTrue(registrationResult instanceof JMTMRegistrationSuccess);
+			assertEquals(1, registeredTaskManagers.size());
+
+			TaskManagerLocation taskManagerLocation = registeredTaskManagers.values().iterator().next().f0;
+
+			assertNotNull(taskManagerLocation.getHostname());
+			assertNotNull(taskManagerLocation.getFQDNHostname());
+			assertNotNull(taskManagerLocation.toString());
+
+			assertTrue(taskManagerLocation.getHostname().contains("127.0.0.1"));
+			assertTrue(taskManagerLocation.getFQDNHostname().contains("127.0.0.1"));
+			assertTrue(taskManagerLocation.toString().contains("127.0.0.1"));
+		} finally {
+			RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
+		}
+	}

Review comment:
       From and end-to-end perspective it is right that we do this test. On the other hand, what we are effectively testing here is that we are reading the `RETRIEVE_TASK_MANAGER_HOSTNAME` config option and passing it to the `TaskManagerLocation`. I am not sure whether the effort of starting a `JobManager` justifies the testing duration.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerLocationTest.java
##########
@@ -198,4 +199,18 @@ public void testGetHostname2() {
 			fail(e.getMessage());
 		}
 	}
+
+	@Test
+	public void testNotRetrieveHostName() {
+		InetAddress address = mock(InetAddress.class);
+		when(address.getCanonicalHostName()).thenReturn("worker10");
+		when(address.getHostName()).thenReturn("worker10");
+		when(address.getHostAddress()).thenReturn("127.0.0.1");

Review comment:
       Instead of mocking, could we use `127.0.0.1` which should resolve to `"localhost"` or are these assumption too strong?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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