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:
[email protected]
