[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT. URL: https://github.com/apache/flink/pull/11284#discussion_r396896951 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/KvStateService.java ## @@ -170,7 +170,7 @@ public static KvStateService fromConfiguration(TaskManagerServicesConfiguration int numProxyServerQueryThreads = qsConfig.numProxyQueryThreads() == 0 ? taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numProxyQueryThreads(); kvClientProxy = QueryableStateUtils.createKvStateClientProxy( - taskManagerServicesConfiguration.getTaskManagerAddress(), + taskManagerServicesConfiguration.getExternalAddress(), Review comment: As mentioned in the JIRA ticket, I'd like to exclude queryable state from the scope of this PR. Because I'm not familiar with how it works, and it seems does not have many use cases. I think this should not affect other use cases, because the code path should only be executed if queryable state is enabled. I think we should mention in the release notes that the new NAT support does not cover queryable state. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r396893625
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
##
@@ -571,9 +572,16 @@ private void releaseEmptyTaskManager(ResourceID
resourceId) {
@Override
public CompletableFuture registerTaskManager(
final String taskManagerRpcAddress,
- final TaskManagerLocation taskManagerLocation,
+ final TaskManagerUnresolvedLocation
taskManagerUnresolvedLocation,
final Time timeout) {
+ final TaskManagerLocation taskManagerLocation;
+ try {
+ taskManagerLocation =
TaskManagerLocation.fromUnresolvedLocation(taskManagerUnresolvedLocation);
Review comment:
For data transmission, yes. But I think the TM's address is required to be
resolvable for JM process anyway. Because we still need them for RPC
communications.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT. URL: https://github.com/apache/flink/pull/11284#discussion_r396891883 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/KvStateService.java ## @@ -181,7 +181,7 @@ public static KvStateService fromConfiguration(TaskManagerServicesConfiguration int numStateServerQueryThreads = qsConfig.numStateQueryThreads() == 0 ? taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numStateQueryThreads(); kvStateServer = QueryableStateUtils.createKvStateServer( Review comment: I have no opinion on this. I'm not familiar with how the queryable state works and its use cases. I've consulted @zhijiangW about this, and he suggested that queryable state does not have much use cases and may not be necessary for the NAT scenarios. In addition, if we want to support this for queryable state, we could use some help from people who are more familiar with the queryable state. 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: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r396889856
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
##
@@ -261,26 +258,30 @@ public void start() throws Exception {
// bring up all the RPC services
LOG.info("Starting RPC Service(s)");
- AkkaRpcServiceConfiguration
akkaRpcServiceConfig =
AkkaRpcServiceConfiguration.fromConfiguration(configuration);
-
final RpcServiceFactory
dispatcherResourceManagreComponentRpcServiceFactory;
if (useSingleRpcService) {
// we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, false, null);
+ commonRpcService =
createLocalRpcService(configuration);
final CommonRpcServiceFactory
commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
taskManagerRpcServiceFactory =
commonRpcServiceFactory;
dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
} else {
- // we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, true, null);
// start a new service per component,
possibly with custom bind addresses
final String jobManagerBindAddress =
miniClusterConfiguration.getJobManagerBindAddress();
final String taskManagerBindAddress =
miniClusterConfiguration.getTaskManagerBindAddress();
+ final String jobManagerBindPort =
miniClusterConfiguration.getJobManagerBindPortRange();
+ final String taskManagerBindPort =
miniClusterConfiguration.getTaskManagerBindPortRange();
+
+
dispatcherResourceManagreComponentRpcServiceFactory =
+ new
DedicatedRpcServiceFactory(configuration, jobManagerBindAddress,
jobManagerBindPort);
Review comment:
I think you are right. A `DedicatedRpcServiceFactory` might be used to
create multiple RPC services. In such cases, the current implementation will
lead to conflicts on binding ports. To that end, I would agree with not
supporting the binding port option and rely on the random port.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395549344
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##
@@ -351,21 +351,21 @@ public static TaskExecutor startTaskManager(
LOG.info("Starting TaskManager with ResourceID: {}",
resourceID);
- InetAddress remoteAddress =
InetAddress.getByName(rpcService.getAddress());
+ InetAddress externalAddress =
InetAddress.getByName(rpcService.getAddress());
Review comment:
Alright, thanks for the explanation. I'll try to fix it.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395464715
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##
@@ -351,21 +351,21 @@ public static TaskExecutor startTaskManager(
LOG.info("Starting TaskManager with ResourceID: {}",
resourceID);
- InetAddress remoteAddress =
InetAddress.getByName(rpcService.getAddress());
+ InetAddress externalAddress =
InetAddress.getByName(rpcService.getAddress());
Review comment:
After looking more into this, I'm not entirely sure whether we should
support unresolvable external addresses.
The prerequisites are unclear to me. If an external address cannot be
resolved by the TaskManager itself, then where can it be resolved? Do we assume
all other components (JobManager and other TaskManagers) except for itself can
resolve the external address? That sounds like a strong assumption to me, which
may not always be true. E.g., deploying a Flink cluster over machines from two
different sub-network, each sub-network has multiple machines/TMs. TMs in the
same sub-network might not be able to resolve each other's external address.
Another example, multiple TMs on the same machine.
I don't see a good way to cover all such cases. If we can only support the
special case where external addresses can be resolved by all other components,
I'm not sure whether we should do it given the efforts needed. Alternatively,
user might use IP addresses to workaround in such cases.
WDYT?
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395464715
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##
@@ -351,21 +351,21 @@ public static TaskExecutor startTaskManager(
LOG.info("Starting TaskManager with ResourceID: {}",
resourceID);
- InetAddress remoteAddress =
InetAddress.getByName(rpcService.getAddress());
+ InetAddress externalAddress =
InetAddress.getByName(rpcService.getAddress());
Review comment:
After looking more into this, I'm not entirely sure whether we should
support unresolvable external addresses.
The prerequisites are unclear to me. If an external address cannot be
resolved by the TaskManager itself, then where can it be resolved? Do we assume
all other components (JobManager and other TaskManagers) except for itself can
resolve the external address? That sounds like a strong assumption to me, which
may not always be true. E.g., deploying a Flink cluster over machines from two
different sub-network, each sub-network has multiple machines/TMs. TMs in the
same sub-network might not be able to resolve each other's external address.
Another example, multiple TMs on the same machine.
I don't see a good way to cover all such cases. If we can only support the
special case where external addresses can be resolved by all other components,
I'm not sure whether we should do it given the efforts needed. Alternatively,
user might use IP addresses to workaround in such cases.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395435948
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
##
@@ -351,21 +351,21 @@ public static TaskExecutor startTaskManager(
LOG.info("Starting TaskManager with ResourceID: {}",
resourceID);
- InetAddress remoteAddress =
InetAddress.getByName(rpcService.getAddress());
+ InetAddress externalAddress =
InetAddress.getByName(rpcService.getAddress());
Review comment:
True. I'll try to fix the problem and update the e2e test case.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395435813
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
##
@@ -102,21 +102,21 @@ public int getNumTaskManagers() {
public String getJobManagerBindAddress() {
return commonBindAddress != null ?
commonBindAddress :
-
configuration.getString(JobManagerOptions.ADDRESS, "localhost");
+
configuration.getString(JobManagerOptions.BIND_HOST, "localhost");
}
public String getTaskManagerBindAddress() {
return commonBindAddress != null ?
commonBindAddress :
-
configuration.getString(TaskManagerOptions.HOST, "localhost");
+
configuration.getString(TaskManagerOptions.BIND_HOST, "localhost");
}
- public String getJobManagerBindPortRange() {
- return
String.valueOf(configuration.getInteger(JobManagerOptions.PORT, 0));
+ public int getJobManagerBindPort() {
+ return
configuration.getInteger(JobManagerOptions.RPC_BIND_PORT, 0);
}
- public String getTaskManagerBindPortRange() {
- return configuration.getString(TaskManagerOptions.RPC_PORT);
+ public int getTaskManagerBindPort() {
+ return
configuration.getInteger(TaskManagerOptions.RPC_BIND_PORT, 0);
Review comment:
Not sure if I fully understand your concern.
I think the RPC services are meant to be used by Flink components
internally. Then how would user be affected if the configured address is not
used internally?
I have no problem preserving the external address/ports if configured. Just
trying to understand.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395430171
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
##
@@ -248,10 +247,14 @@ protected void initializeServices(Configuration
configuration) throws Exception
LOG.info("Initializing cluster services.");
synchronized (lock) {
- final String bindAddress =
configuration.getString(JobManagerOptions.ADDRESS);
- final String portRange = getRPCPortRange(configuration);
-
- commonRpcService = createRpcService(configuration,
bindAddress, portRange);
+ commonRpcService =
AkkaRpcServiceUtils.createRemoteRpcService(
+ configuration,
+
configuration.getString(JobManagerOptions.ADDRESS),
+ getRPCPortRange(configuration),
+
configuration.getString(JobManagerOptions.BIND_HOST),
+
configuration.contains(JobManagerOptions.RPC_BIND_PORT) ?
+
configuration.getInteger(JobManagerOptions.RPC_BIND_PORT) :
+ null);
Review comment:
Same here. I also think using Optional will express the contract better. The
question is do we need to strictly follow the code style guide.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395429730
##
File path:
flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##
@@ -94,19 +94,28 @@
" or if it has been quarantined by another
actor system.");
/**
-* The config parameter defining the task manager's hostname.
+* The external address of the network interface where the TaskManager
is exposed.
* Overrides {@link #HOST_BIND_POLICY} automatic address binding.
*/
@Documentation.Section({Documentation.Sections.COMMON_HOST_PORT,
Documentation.Sections.ALL_TASK_MANAGER})
public static final ConfigOption HOST =
key("taskmanager.host")
.stringType()
.noDefaultValue()
- .withDescription("The address of the network interface
that the TaskManager binds to." +
- " This option can be used to define explicitly
a binding address. Because" +
- " different TaskManagers need different values
for this option, usually it is specified in an" +
+ .withDescription("The external address of the network
interface where the TaskManager is exposed." +
+ " Because different TaskManagers need different
values for this option, usually it is specified in an" +
" additional non-shared TaskManager-specific
config file.");
+ /**
+* The local address of the network interface that the task manager
binds to.
+*/
+ public static final ConfigOption BIND_HOST =
+ key("taskmanager.bind-host")
Review comment:
Same here. TM address and bind-address are shared by RPC service, netty
shuffle service and queryable kv state.
I'm actually thinking about changing "jobmanager.rpc.address" to
"jobmanager.host" as a follow up, to make the config keys consistently using
"host".
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395428193
##
File path:
flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
##
@@ -54,6 +54,16 @@
" leader-election service (like ZooKeeper) is used to
elect and discover the JobManager" +
" leader from potentially multiple standby
JobManagers.");
+ /**
+* The local address of the network interface that the job manager
binds to.
+*/
+ public static final ConfigOption BIND_HOST =
+ key("jobmanager.bind-host")
Review comment:
This address is not only used by the RPC service, but also shared by the
blob server, rest server, and potentially any other service in future that
needs to bind to the network interface.
To that end, I think the config option for TM ("taskmanager.host") makes
more sense. I guess that might be the reason we use "host" in the config key at
the first place.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395426701
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
##
@@ -261,26 +258,30 @@ public void start() throws Exception {
// bring up all the RPC services
LOG.info("Starting RPC Service(s)");
- AkkaRpcServiceConfiguration
akkaRpcServiceConfig =
AkkaRpcServiceConfiguration.fromConfiguration(configuration);
-
final RpcServiceFactory
dispatcherResourceManagreComponentRpcServiceFactory;
if (useSingleRpcService) {
// we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, false, null);
+ commonRpcService =
createLocalRpcService(configuration);
final CommonRpcServiceFactory
commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
taskManagerRpcServiceFactory =
commonRpcServiceFactory;
dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
} else {
- // we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, true, null);
// start a new service per component,
possibly with custom bind addresses
final String jobManagerBindAddress =
miniClusterConfiguration.getJobManagerBindAddress();
final String taskManagerBindAddress =
miniClusterConfiguration.getTaskManagerBindAddress();
+ final String jobManagerBindPort =
miniClusterConfiguration.getJobManagerBindPortRange();
+ final String taskManagerBindPort =
miniClusterConfiguration.getTaskManagerBindPortRange();
Review comment:
Please see my explanation above.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395426615
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
##
@@ -261,26 +258,30 @@ public void start() throws Exception {
// bring up all the RPC services
LOG.info("Starting RPC Service(s)");
- AkkaRpcServiceConfiguration
akkaRpcServiceConfig =
AkkaRpcServiceConfiguration.fromConfiguration(configuration);
-
final RpcServiceFactory
dispatcherResourceManagreComponentRpcServiceFactory;
if (useSingleRpcService) {
// we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, false, null);
+ commonRpcService =
createLocalRpcService(configuration);
final CommonRpcServiceFactory
commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
taskManagerRpcServiceFactory =
commonRpcServiceFactory;
dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
} else {
- // we always need the
'commonRpcService' for auxiliary calls
- commonRpcService =
createRpcService(akkaRpcServiceConfig, true, null);
// start a new service per component,
possibly with custom bind addresses
final String jobManagerBindAddress =
miniClusterConfiguration.getJobManagerBindAddress();
final String taskManagerBindAddress =
miniClusterConfiguration.getTaskManagerBindAddress();
+ final String jobManagerBindPort =
miniClusterConfiguration.getJobManagerBindPortRange();
Review comment:
No, and I don't think we should.
Whenever the user needs to specify a bind port apart form the external port,
the user also needs to set port forwarding rules in the environment, e.g. NAT
gateway. For each external port, it can only be forwarded to one internal port.
Therefore, it does not make sense to support a range of bind ports, because the
user needs to know which internal port is bound anyway.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395425038
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
##
@@ -82,92 +82,80 @@
.build();
/**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
+* Starts a remote ActorSystem at given address and specific port range.
* @param configuration The Flink configuration
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
+* @param externalAddress The external address to access the
ActorSystem.
+* @param externalPortRange The choosing range of the external port to
access the ActorSystem.
* @param logger The logger to output log information.
* @return The ActorSystem which has been started
* @throws Exception Thrown when actor system cannot be started in
specified port range
*/
- public static ActorSystem startActorSystem(
+ @VisibleForTesting
+ public static ActorSystem startRemoteActorSystem(
Configuration configuration,
- String listeningAddress,
- String portRangeDefinition,
+ String externalAddress,
+ String externalPortRange,
Logger logger) throws Exception {
- return startActorSystem(
- configuration,
- listeningAddress,
- portRangeDefinition,
- logger,
-
ForkJoinExecutorConfiguration.fromConfiguration(configuration));
- }
-
- /**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
-*
-* @param configuration The Flink configuration
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
-* @param logger The logger to output log information.
-* @param actorSystemExecutorConfiguration configuration for the
ActorSystem's underlying executor
-* @return The ActorSystem which has been started
-* @throws Exception Thrown when actor system cannot be started in
specified port range
-*/
- public static ActorSystem startActorSystem(
- Configuration configuration,
- String listeningAddress,
- String portRangeDefinition,
- Logger logger,
- @Nonnull ActorSystemExecutorConfiguration
actorSystemExecutorConfiguration) throws Exception {
- return startActorSystem(
+ return startRemoteActorSystem(
configuration,
AkkaUtils.getFlinkActorSystemName(),
- listeningAddress,
- portRangeDefinition,
+ externalAddress,
+ externalPortRange,
+ NetUtils.getWildcardIPAddress(),
+ -1,
logger,
- actorSystemExecutorConfiguration);
+
ForkJoinExecutorConfiguration.fromConfiguration(configuration),
+ null);
}
/**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
+* Starts a remote ActorSystem at given address and specific port range.
*
* @param configuration The Flink configuration
* @param actorSystemName Name of the started {@link ActorSystem}
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
+* @param externalAddress The external address to access the
ActorSystem.
+* @param externalPortRange The choosing range of the external port to
access the ActorSystem.
+* @param bindAddress The local address to bind to.
+* @param bindPort The local port to bind to. If negative, external
port will be used.
* @param logger The logger to output log information.
* @param actorSystemExecutorConfiguration configuration for the
ActorSystem's underlying executor
+* @param customConfig Custom Akka config to be combined with the
config derived from Flink configuration.
* @return The ActorSystem which has been started
* @throws Exception Thrown when actor system cannot be started in
specified port range
*/
- public static ActorSystem startActorSystem(
+ public static ActorSystem startRemoteActorSystem(
Configuration configuration,
String acto
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r395425038
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
##
@@ -82,92 +82,80 @@
.build();
/**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
+* Starts a remote ActorSystem at given address and specific port range.
* @param configuration The Flink configuration
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
+* @param externalAddress The external address to access the
ActorSystem.
+* @param externalPortRange The choosing range of the external port to
access the ActorSystem.
* @param logger The logger to output log information.
* @return The ActorSystem which has been started
* @throws Exception Thrown when actor system cannot be started in
specified port range
*/
- public static ActorSystem startActorSystem(
+ @VisibleForTesting
+ public static ActorSystem startRemoteActorSystem(
Configuration configuration,
- String listeningAddress,
- String portRangeDefinition,
+ String externalAddress,
+ String externalPortRange,
Logger logger) throws Exception {
- return startActorSystem(
- configuration,
- listeningAddress,
- portRangeDefinition,
- logger,
-
ForkJoinExecutorConfiguration.fromConfiguration(configuration));
- }
-
- /**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
-*
-* @param configuration The Flink configuration
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
-* @param logger The logger to output log information.
-* @param actorSystemExecutorConfiguration configuration for the
ActorSystem's underlying executor
-* @return The ActorSystem which has been started
-* @throws Exception Thrown when actor system cannot be started in
specified port range
-*/
- public static ActorSystem startActorSystem(
- Configuration configuration,
- String listeningAddress,
- String portRangeDefinition,
- Logger logger,
- @Nonnull ActorSystemExecutorConfiguration
actorSystemExecutorConfiguration) throws Exception {
- return startActorSystem(
+ return startRemoteActorSystem(
configuration,
AkkaUtils.getFlinkActorSystemName(),
- listeningAddress,
- portRangeDefinition,
+ externalAddress,
+ externalPortRange,
+ NetUtils.getWildcardIPAddress(),
+ -1,
logger,
- actorSystemExecutorConfiguration);
+
ForkJoinExecutorConfiguration.fromConfiguration(configuration),
+ null);
}
/**
-* Starts an ActorSystem with the given configuration listening at the
address/ports.
+* Starts a remote ActorSystem at given address and specific port range.
*
* @param configuration The Flink configuration
* @param actorSystemName Name of the started {@link ActorSystem}
-* @param listeningAddress The address to listen at.
-* @param portRangeDefinition The port range to choose a port from.
+* @param externalAddress The external address to access the
ActorSystem.
+* @param externalPortRange The choosing range of the external port to
access the ActorSystem.
+* @param bindAddress The local address to bind to.
+* @param bindPort The local port to bind to. If negative, external
port will be used.
* @param logger The logger to output log information.
* @param actorSystemExecutorConfiguration configuration for the
ActorSystem's underlying executor
+* @param customConfig Custom Akka config to be combined with the
config derived from Flink configuration.
* @return The ActorSystem which has been started
* @throws Exception Thrown when actor system cannot be started in
specified port range
*/
- public static ActorSystem startActorSystem(
+ public static ActorSystem startRemoteActorSystem(
Configuration configuration,
String acto
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386828354
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceBuilder.java
##
@@ -0,0 +1,136 @@
+/*
+ * 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.rpc.akka;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.util.NetUtils;
+import org.apache.flink.util.Preconditions;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.Config;
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import java.net.InetAddress;
+
+/**
+ * Builder for {@link AkkaRpcService}.
+ */
+public class AkkaRpcServiceBuilder {
Review comment:
I moved `AkkaRpcServiceBuilder` into `AkkaRpcServiceUtils`, and made its
constructors private. Also annotated constructor of `AkkaRpcService`
`VisibleForTesting`.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386818991
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceBuilder.java
##
@@ -0,0 +1,136 @@
+/*
+ * 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.rpc.akka;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.util.NetUtils;
+import org.apache.flink.util.Preconditions;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.Config;
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import java.net.InetAddress;
+
+/**
+ * Builder for {@link AkkaRpcService}.
+ */
+public class AkkaRpcServiceBuilder {
Review comment:
I think whether we want to make a constructor private, depends on whether
allowing it to be accessed from outside may cause problem. E.g., the none
argument constructor of `ResourceProfile` must be private because the only case
we need it is for constructing the singleton instance `UNKNOWN`. If we allow to
create another instance of `ResourceProfile` with this none argument
constructor, the created instance may not work properly.
In this case, I don't see such problem in allowing direct access to the
constructors. That makes hiding the constructors a nice but not must to have
approach, by which we suggest people to choose the more convenient builders.
On the other hand, it is not about how many efforts needed that we do not
use builders in the test cases. Currently, the builder does two things: create
a `ActorSystem`, and use it to create a `AkkaRpcService`. However, in many test
cases, we need to use a custom testing actor system, which is different from
the one we created for production. That's way we cannot easily use builders in
the tests.
We could also make the builder to support custom actor system. But that
complicates the builder logics (what should we do if both custom actor system
and the fields for creating the production actor system are provided), and I
think it doesn't make sense to do for pure testing purpose.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386807612
##
File path: flink-end-to-end-tests/test-scripts/test_nat.sh
##
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+source "$(dirname "$0")"/common.sh
+
+DOCKER_MODULE_DIR=${END_TO_END_DIR}/../flink-container/docker
+DOCKER_SCRIPTS=${END_TO_END_DIR}/test-scripts/container-scripts
+DOCKER_IMAGE_BUILD_RETRIES=3
+BUILD_BACKOFF_TIME=5
+
+export FLINK_JOB=org.apache.flink.examples.java.wordcount.WordCount
+export FLINK_DOCKER_IMAGE_NAME=test_nat
+export INPUT_VOLUME=${END_TO_END_DIR}/test-scripts/test-data
+export OUTPUT_VOLUME=${TEST_DATA_DIR}/out
+export INPUT_PATH=/data/test/input
+export OUTPUT_PATH=/data/test/output
+
+export HOST_ADDR=$(get_node_ip | awk '{print $1}')
+export JM_RPC_EX_PORT=1
+export JM_RPC_IN_PORT=1
+export TM_1_RPC_EX_PORT=10001
+export TM_2_RPC_EX_PORT=10002
+export TM_RPC_IN_PORT=1
+export TM_1_DATA_EX_PORT=11001
+export TM_2_DATA_EX_PORT=11002
+export TM_DATA_IN_PORT=11000
+
+RESULT_HASH="72a690412be8928ba239c2da967328a5"
+INPUT_ARGS="--input ${INPUT_PATH}/words"
+
+export FLINK_JOB_ARGUMENTS="${INPUT_ARGS} --output
${OUTPUT_PATH}/docker_wc_out"
+
+build_image() {
+./build.sh --from-local-dist --job-artifacts
${FLINK_DIR}/examples/batch/WordCount.jar --image-name
${FLINK_DOCKER_IMAGE_NAME}
+}
+
+# user inside the container must be able to create files, this is a workaround
in-container permissions
+mkdir -p $OUTPUT_VOLUME
+chmod 777 $OUTPUT_VOLUME
Review comment:
Turns out `666` does not work. Flink throws the following error:
> Output directory '/data/test/output/docker_wc_out' could not be created.
And I found from google that:
> For directories, execute permission allows you to enter the directory
(i.e., cd into it), and to access any of its files.
http://web.mit.edu/sipb/doc/working/afs/html/subsection3.1.html
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386804627
##
File path: flink-end-to-end-tests/test-scripts/test_nat.sh
##
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+source "$(dirname "$0")"/common.sh
+
+DOCKER_MODULE_DIR=${END_TO_END_DIR}/../flink-container/docker
+DOCKER_SCRIPTS=${END_TO_END_DIR}/test-scripts/container-scripts
+DOCKER_IMAGE_BUILD_RETRIES=3
+BUILD_BACKOFF_TIME=5
+
+export FLINK_JOB=org.apache.flink.examples.java.wordcount.WordCount
+export FLINK_DOCKER_IMAGE_NAME=test_nat
+export INPUT_VOLUME=${END_TO_END_DIR}/test-scripts/test-data
+export OUTPUT_VOLUME=${TEST_DATA_DIR}/out
+export INPUT_PATH=/data/test/input
+export OUTPUT_PATH=/data/test/output
+
+export HOST_ADDR=$(get_node_ip | awk '{print $1}')
+export JM_RPC_EX_PORT=1
+export JM_RPC_IN_PORT=1
+export TM_1_RPC_EX_PORT=10001
+export TM_2_RPC_EX_PORT=10002
+export TM_RPC_IN_PORT=1
+export TM_1_DATA_EX_PORT=11001
+export TM_2_DATA_EX_PORT=11002
+export TM_DATA_IN_PORT=11000
+
+RESULT_HASH="72a690412be8928ba239c2da967328a5"
+INPUT_ARGS="--input ${INPUT_PATH}/words"
+
+export FLINK_JOB_ARGUMENTS="${INPUT_ARGS} --output
${OUTPUT_PATH}/docker_wc_out"
+
+build_image() {
+./build.sh --from-local-dist --job-artifacts
${FLINK_DIR}/examples/batch/WordCount.jar --image-name
${FLINK_DOCKER_IMAGE_NAME}
+}
+
+# user inside the container must be able to create files, this is a workaround
in-container permissions
+mkdir -p $OUTPUT_VOLUME
+chmod 777 $OUTPUT_VOLUME
Review comment:
I would argue that this is a new created empty directory dedicated for the
test case to use, and will be be cleaned up when the test is finished. But
never mind, I'll make it `666` if it bothers you.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386803470
##
File path: flink-end-to-end-tests/test-scripts/test_nat.sh
##
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+source "$(dirname "$0")"/common.sh
+
+DOCKER_MODULE_DIR=${END_TO_END_DIR}/../flink-container/docker
+DOCKER_SCRIPTS=${END_TO_END_DIR}/test-scripts/container-scripts
+DOCKER_IMAGE_BUILD_RETRIES=3
+BUILD_BACKOFF_TIME=5
+
+export FLINK_JOB=org.apache.flink.examples.java.wordcount.WordCount
+export FLINK_DOCKER_IMAGE_NAME=test_nat
+export INPUT_VOLUME=${END_TO_END_DIR}/test-scripts/test-data
+export OUTPUT_VOLUME=${TEST_DATA_DIR}/out
+export INPUT_PATH=/data/test/input
+export OUTPUT_PATH=/data/test/output
+
+export HOST_ADDR=$(get_node_ip | awk '{print $1}')
+export JM_RPC_EX_PORT=1
+export JM_RPC_IN_PORT=1
+export TM_1_RPC_EX_PORT=10001
+export TM_2_RPC_EX_PORT=10002
+export TM_RPC_IN_PORT=1
+export TM_1_DATA_EX_PORT=11001
+export TM_2_DATA_EX_PORT=11002
+export TM_DATA_IN_PORT=11000
+
+RESULT_HASH="72a690412be8928ba239c2da967328a5"
+INPUT_ARGS="--input ${INPUT_PATH}/words"
+
+export FLINK_JOB_ARGUMENTS="${INPUT_ARGS} --output
${OUTPUT_PATH}/docker_wc_out"
+
+build_image() {
+./build.sh --from-local-dist --job-artifacts
${FLINK_DIR}/examples/batch/WordCount.jar --image-name
${FLINK_DOCKER_IMAGE_NAME}
+}
+
+# user inside the container must be able to create files, this is a workaround
in-container permissions
+mkdir -p $OUTPUT_VOLUME
+chmod 777 $OUTPUT_VOLUME
+
+pushd "$DOCKER_MODULE_DIR"
+if ! retry_times $DOCKER_IMAGE_BUILD_RETRIES ${BUILD_BACKOFF_TIME}
build_image; then
+echo "Failed to build docker image. Aborting..."
+exit 1
+fi
+popd
+
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml up
--abort-on-container-exit --exit-code-from job-cluster &> /dev/null
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs job-cluster >
${FLINK_DIR}/log/jobmanager.log
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs taskmanager1 >
${FLINK_DIR}/log/taskmanager1.log
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs taskmanager2 >
${FLINK_DIR}/log/taskmanager2.log
+
+cat $OUTPUT_VOLUME/docker_wc_out/1 $OUTPUT_VOLUME/docker_wc_out/2 | sort >
$OUTPUT_VOLUME/docker_wc_out/merged
Review comment:
You're right.
I think I tried this before, but forgot to add the '/' in the prefix so it
failed.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386790513
##
File path: flink-end-to-end-tests/test-scripts/test_nat.sh
##
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+source "$(dirname "$0")"/common.sh
+
+DOCKER_MODULE_DIR=${END_TO_END_DIR}/../flink-container/docker
+DOCKER_SCRIPTS=${END_TO_END_DIR}/test-scripts/container-scripts
+DOCKER_IMAGE_BUILD_RETRIES=3
+BUILD_BACKOFF_TIME=5
+
+export FLINK_JOB=org.apache.flink.examples.java.wordcount.WordCount
+export FLINK_DOCKER_IMAGE_NAME=test_nat
+export INPUT_VOLUME=${END_TO_END_DIR}/test-scripts/test-data
+export OUTPUT_VOLUME=${TEST_DATA_DIR}/out
+export INPUT_PATH=/data/test/input
+export OUTPUT_PATH=/data/test/output
+
+export HOST_ADDR=$(get_node_ip | awk '{print $1}')
+export JM_RPC_EX_PORT=1
+export JM_RPC_IN_PORT=1
+export TM_1_RPC_EX_PORT=10001
+export TM_2_RPC_EX_PORT=10002
+export TM_RPC_IN_PORT=1
+export TM_1_DATA_EX_PORT=11001
+export TM_2_DATA_EX_PORT=11002
+export TM_DATA_IN_PORT=11000
+
+RESULT_HASH="72a690412be8928ba239c2da967328a5"
+INPUT_ARGS="--input ${INPUT_PATH}/words"
+
+export FLINK_JOB_ARGUMENTS="${INPUT_ARGS} --output
${OUTPUT_PATH}/docker_wc_out"
+
+build_image() {
+./build.sh --from-local-dist --job-artifacts
${FLINK_DIR}/examples/batch/WordCount.jar --image-name
${FLINK_DOCKER_IMAGE_NAME}
+}
+
+# user inside the container must be able to create files, this is a workaround
in-container permissions
+mkdir -p $OUTPUT_VOLUME
+chmod 777 $OUTPUT_VOLUME
+
+pushd "$DOCKER_MODULE_DIR"
+if ! retry_times $DOCKER_IMAGE_BUILD_RETRIES ${BUILD_BACKOFF_TIME}
build_image; then
+echo "Failed to build docker image. Aborting..."
+exit 1
+fi
+popd
+
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml up
--abort-on-container-exit --exit-code-from job-cluster &> /dev/null
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs job-cluster >
${FLINK_DIR}/log/jobmanager.log
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs taskmanager1 >
${FLINK_DIR}/log/taskmanager1.log
+docker-compose -f ${DOCKER_SCRIPTS}/docker-compose.nat.yml logs taskmanager2 >
${FLINK_DIR}/log/taskmanager2.log
+
+cat $OUTPUT_VOLUME/docker_wc_out/1 $OUTPUT_VOLUME/docker_wc_out/2 | sort >
$OUTPUT_VOLUME/docker_wc_out/merged
Review comment:
`check_result_hash_no_exit` requires a common prefix of the output files,
which the outputs of WordCount do not have.
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386790141
##
File path: flink-end-to-end-tests/test-scripts/test_nat.sh
##
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+source "$(dirname "$0")"/common.sh
+
+DOCKER_MODULE_DIR=${END_TO_END_DIR}/../flink-container/docker
+DOCKER_SCRIPTS=${END_TO_END_DIR}/test-scripts/container-scripts
+DOCKER_IMAGE_BUILD_RETRIES=3
+BUILD_BACKOFF_TIME=5
+
+export FLINK_JOB=org.apache.flink.examples.java.wordcount.WordCount
+export FLINK_DOCKER_IMAGE_NAME=test_nat
+export INPUT_VOLUME=${END_TO_END_DIR}/test-scripts/test-data
+export OUTPUT_VOLUME=${TEST_DATA_DIR}/out
+export INPUT_PATH=/data/test/input
+export OUTPUT_PATH=/data/test/output
+
+export HOST_ADDR=$(get_node_ip | awk '{print $1}')
+export JM_RPC_EX_PORT=1
+export JM_RPC_IN_PORT=1
+export TM_1_RPC_EX_PORT=10001
+export TM_2_RPC_EX_PORT=10002
+export TM_RPC_IN_PORT=1
+export TM_1_DATA_EX_PORT=11001
+export TM_2_DATA_EX_PORT=11002
+export TM_DATA_IN_PORT=11000
+
+RESULT_HASH="72a690412be8928ba239c2da967328a5"
+INPUT_ARGS="--input ${INPUT_PATH}/words"
+
+export FLINK_JOB_ARGUMENTS="${INPUT_ARGS} --output
${OUTPUT_PATH}/docker_wc_out"
+
+build_image() {
+./build.sh --from-local-dist --job-artifacts
${FLINK_DIR}/examples/batch/WordCount.jar --image-name
${FLINK_DOCKER_IMAGE_NAME}
+}
+
+# user inside the container must be able to create files, this is a workaround
in-container permissions
+mkdir -p $OUTPUT_VOLUME
+chmod 777 $OUTPUT_VOLUME
Review comment:
I think here we simply give the container full access to the path, not
assuming what it does with the path. What is the problem?
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:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #11284: [FLINK-15911][runtime] Make Flink work with NAT.
xintongsong commented on a change in pull request #11284:
[FLINK-15911][runtime] Make Flink work with NAT.
URL: https://github.com/apache/flink/pull/11284#discussion_r386789347
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceBuilder.java
##
@@ -0,0 +1,136 @@
+/*
+ * 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.rpc.akka;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.util.NetUtils;
+import org.apache.flink.util.Preconditions;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.Config;
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import java.net.InetAddress;
+
+/**
+ * Builder for {@link AkkaRpcService}.
+ */
+public class AkkaRpcServiceBuilder {
Review comment:
I don't get it.
- Constructor of `AkkaRpcService` is used in various of test cases. Unless
we want to modify them all, I don't think we can make the constructor private.
Maybe it makes sense to annotate the constructor as `VisibleForTesting`,
because without the test cases it can be package visibility.
- If we cannot make constructor of `AkkaRpcService` private anyway, then I
would prefer to put `AkkaRpcServiceBuilder` in a separate file. The file
`AkkaRpcService.java` is already quite long. No need to further complicate it.
- I don't see how `AkkaRpcService#local(remote)RpcServiceBuilder` is more
readable compared to `AkkaRpcServiceUtils#local(remote)ServiceBuilder`.
- It probably make sense to make the constructor of the builder private,
since we already have `AkkaRpcServiceUtils#local(remote)ServiceBuilder`. In
that case, I would tend to put the builder inside `AkkaRpcServiceUtils` rather
than `AkkaRpcService`.
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:
us...@infra.apache.org
With regards,
Apache Git Services
