You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/02/05 09:47:56 UTC
[flink] 06/09: [hotfix] Simplify MiniCluster by using the
DispatcherResourceManagerComponent
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 95f3ea086fb80cd3966a2e72b0713f4404463818
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Feb 1 11:58:26 2019 +0100
[hotfix] Simplify MiniCluster by using the DispatcherResourceManagerComponent
This commit simplifies the MiniCluster by using the DispatcherResourceManagerComponent
to start a Dispatcher and a ResourceManager. Moreover it refactors the MiniCluster to
support support for multiple DispatcherResourceManagerComponents.
---
.../org/apache/flink/client/LocalExecutor.java | 2 +-
...tDispatcherResourceManagerComponentFactory.java | 11 +-
.../DispatcherResourceManagerComponent.java | 12 +-
.../flink/runtime/minicluster/MiniCluster.java | 374 +++++++++------------
.../minicluster/MiniClusterConfiguration.java | 6 -
.../runtime/testutils/MiniClusterResource.java | 4 +-
.../org/apache/flink/api/scala/FlinkShell.scala | 2 +-
.../api/environment/LocalStreamEnvironment.java | 2 +-
.../test/operators/RemoteEnvironmentITCase.java | 2 +-
9 files changed, 177 insertions(+), 238 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
index 14d3ee5..77bda09 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java
@@ -143,7 +143,7 @@ public class LocalExecutor extends PlanExecutor {
final MiniCluster miniCluster = new MiniCluster(miniClusterConfiguration);
miniCluster.start();
- configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().getPort());
+ configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());
return miniCluster;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
index fb514a3..a2e95b0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
@@ -160,9 +160,11 @@ public abstract class AbstractDispatcherResourceManagerComponentFactory<T extend
log.debug("Starting Dispatcher REST endpoint.");
webMonitorEndpoint.start();
+ final String hostname = getHostname(rpcService);
+
jobManagerMetricGroup = MetricUtils.instantiateJobManagerMetricGroup(
metricRegistry,
- rpcService.getAddress(),
+ hostname,
ConfigurationUtils.getSystemResourceMetricsProbingInterval(configuration));
resourceManager = resourceManagerFactory.createResourceManager(
@@ -173,7 +175,7 @@ public abstract class AbstractDispatcherResourceManagerComponentFactory<T extend
heartbeatServices,
metricRegistry,
fatalErrorHandler,
- new ClusterInformation(rpcService.getAddress(), blobServer.getPort()),
+ new ClusterInformation(hostname, blobServer.getPort()),
webMonitorEndpoint.getRestBaseUrl(),
jobManagerMetricGroup);
@@ -258,6 +260,11 @@ public abstract class AbstractDispatcherResourceManagerComponentFactory<T extend
}
}
+ protected String getHostname(RpcService rpcService) {
+ final String rpcServiceAddress = rpcService.getAddress();
+ return rpcServiceAddress == "" ? "localhost" : rpcServiceAddress;
+ }
+
protected abstract DispatcherResourceManagerComponent<T> createDispatcherResourceManagerComponent(
T dispatcher,
ResourceManager<?> resourceManager,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
index 6e28ab6..83f5252 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
import org.apache.flink.runtime.resourcemanager.ResourceManager;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
+import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import javax.annotation.Nonnull;
@@ -40,7 +41,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
* Component which starts a {@link Dispatcher}, {@link ResourceManager} and {@link WebMonitorEndpoint}
* in the same process.
*/
-public class DispatcherResourceManagerComponent<T extends Dispatcher> {
+public class DispatcherResourceManagerComponent<T extends Dispatcher> implements AutoCloseableAsync {
@Nonnull
private final T dispatcher;
@@ -141,7 +142,7 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> {
final CompletableFuture<Void> closeWebMonitorAndDeregisterAppFuture =
FutureUtils.composeAfterwards(webMonitorEndpoint.closeAsync(), () -> deregisterApplication(applicationStatus, diagnostics));
- return FutureUtils.composeAfterwards(closeWebMonitorAndDeregisterAppFuture, this::closeAsyncInternal);
+ return FutureUtils.composeAfterwards(closeWebMonitorAndDeregisterAppFuture, this::closeAsyncInteral);
} else {
return terminationFuture;
}
@@ -155,7 +156,7 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> {
return selfGateway.deregisterApplication(applicationStatus, diagnostics).thenApply(ack -> null);
}
- private CompletableFuture<Void> closeAsyncInternal() {
+ private CompletableFuture<Void> closeAsyncInteral() {
Exception exception = null;
final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);
@@ -198,4 +199,9 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> {
return terminationFuture;
}
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return deregisterApplicationAndClose(ApplicationStatus.CANCELED, "DispatcherResourceManagerComponent has been closed.");
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 2046daf..819dbf0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -24,9 +24,6 @@ import org.apache.flink.api.common.JobSubmissionResult;
import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ConfigurationUtils;
-import org.apache.flink.configuration.MetricOptions;
-import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobCacheService;
@@ -38,15 +35,12 @@ import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.clusterframework.FlinkResourceManager;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.FutureUtils;
-import org.apache.flink.runtime.dispatcher.DefaultJobManagerRunnerFactory;
-import org.apache.flink.runtime.dispatcher.Dispatcher;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.dispatcher.DispatcherId;
-import org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint;
-import org.apache.flink.runtime.dispatcher.HistoryServerArchivist;
import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
-import org.apache.flink.runtime.dispatcher.StandaloneDispatcher;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
+import org.apache.flink.runtime.entrypoint.component.SessionDispatcherResourceManagerComponentFactory;
import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -65,11 +59,7 @@ import org.apache.flink.runtime.metrics.util.MetricUtils;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.ResourceManagerRunner;
-import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
-import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
-import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher;
-import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcherImpl;
-import org.apache.flink.runtime.rest.handler.legacy.metrics.VoidMetricFetcher;
+import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcUtils;
@@ -77,12 +67,14 @@ import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceConfiguration;
import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
-import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
+import org.apache.flink.runtime.webmonitor.retriever.LeaderRetriever;
+import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.function.FunctionUtils;
import akka.actor.ActorSystem;
import com.typesafe.config.Config;
@@ -98,13 +90,13 @@ import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -135,13 +127,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
private RpcService commonRpcService;
@GuardedBy("lock")
- private RpcService jobManagerRpcService;
-
- @GuardedBy("lock")
- private RpcService[] taskManagerRpcServices;
-
- @GuardedBy("lock")
- private RpcService resourceManagerRpcService;
+ private final Collection<RpcService> rpcServices;
@GuardedBy("lock")
private ActorSystem metricQueryServiceActorSystem;
@@ -158,28 +144,19 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
@GuardedBy("lock")
private BlobCacheService blobCacheService;
- @GuardedBy("lock")
- private ResourceManagerRunner resourceManagerRunner;
-
private volatile TaskExecutor[] taskManagers;
@GuardedBy("lock")
- private DispatcherRestEndpoint dispatcherRestEndpoint;
-
- @GuardedBy("lock")
- private URI restAddressURI;
-
- @GuardedBy("lock")
private LeaderRetrievalService resourceManagerLeaderRetriever;
@GuardedBy("lock")
private LeaderRetrievalService dispatcherLeaderRetriever;
@GuardedBy("lock")
- private StandaloneDispatcher dispatcher;
+ private LeaderRetrievalService webMonitorLeaderRetrievalService;
@GuardedBy("lock")
- private JobManagerMetricGroup jobManagerMetricGroup;
+ private Collection<DispatcherResourceManagerComponent<?>> dispatcherResourceManagerComponents;
@GuardedBy("lock")
private RpcGatewayRetriever<DispatcherId, DispatcherGateway> dispatcherGatewayRetriever;
@@ -187,6 +164,9 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
@GuardedBy("lock")
private RpcGatewayRetriever<ResourceManagerId, ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @GuardedBy("lock")
+ private LeaderRetriever webMonitorLeaderRetriever;
+
/** Flag marking the mini cluster as started/running. */
private volatile boolean running;
@@ -199,16 +179,18 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
*/
public MiniCluster(MiniClusterConfiguration miniClusterConfiguration) {
this.miniClusterConfiguration = checkNotNull(miniClusterConfiguration, "config may not be null");
+ this.rpcServices = new ArrayList<>(1 + 2 + miniClusterConfiguration.getNumTaskManagers()); // common + JM + RM + TMs
+ this.dispatcherResourceManagerComponents = new ArrayList<>(1);
this.rpcTimeout = Time.seconds(10L);
this.terminationFuture = CompletableFuture.completedFuture(null);
running = false;
}
- public URI getRestAddress() {
+ public CompletableFuture<URI> getRestAddress() {
synchronized (lock) {
checkState(running, "MiniCluster is not yet running.");
- return restAddressURI;
+ return webMonitorLeaderRetriever.getLeaderFuture().thenApply(FunctionUtils.uncheckedFunction(addressLeaderIdTuple -> new URI(addressLeaderIdTuple.f0)));
}
}
@@ -251,7 +233,6 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
LOG.debug("Using configuration {}", miniClusterConfiguration);
final Configuration configuration = miniClusterConfiguration.getConfiguration();
- final Time rpcTimeout = miniClusterConfiguration.getRpcTimeout();
final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers();
final boolean useSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED;
@@ -260,59 +241,40 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
LOG.info("Starting Metrics Registry");
metricRegistry = createMetricRegistry(configuration);
- this.jobManagerMetricGroup = MetricUtils.instantiateJobManagerMetricGroup(
- metricRegistry,
- "localhost",
- ConfigurationUtils.getSystemResourceMetricsProbingInterval(configuration));
-
- final RpcService jobManagerRpcService;
- final RpcService resourceManagerRpcService;
- final RpcService[] taskManagerRpcServices = new RpcService[numTaskManagers];
// bring up all the RPC services
LOG.info("Starting RPC Service(s)");
AkkaRpcServiceConfiguration akkaRpcServiceConfig = AkkaRpcServiceConfiguration.fromConfiguration(configuration);
- // we always need the 'commonRpcService' for auxiliary calls
- commonRpcService = createRpcService(akkaRpcServiceConfig, false, null);
- // TODO: Temporary hack until the metric query service is ported to the RpcEndpoint
- metricQueryServiceActorSystem = MetricUtils.startMetricsActorSystem(
- configuration,
- commonRpcService.getAddress(),
- LOG);
- metricRegistry.startQueryService(metricQueryServiceActorSystem, null);
+ final RpcServiceFactory taskManagerRpcServiceFactory;
+ final RpcServiceFactory dispatcherResourceManagreComponentRpcServiceFactory;
if (useSingleRpcService) {
- for (int i = 0; i < numTaskManagers; i++) {
- taskManagerRpcServices[i] = commonRpcService;
- }
+ // we always need the 'commonRpcService' for auxiliary calls
+ commonRpcService = createRpcService(akkaRpcServiceConfig, false, null);
+ final CommonRpcServiceFactory commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
+ taskManagerRpcServiceFactory = commonRpcServiceFactory;
+ dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
+ } else {
+ // we always need the 'commonRpcService' for auxiliary calls
+ commonRpcService = createRpcService(akkaRpcServiceConfig, true, null);
- jobManagerRpcService = commonRpcService;
- resourceManagerRpcService = commonRpcService;
-
- this.resourceManagerRpcService = null;
- this.jobManagerRpcService = null;
- this.taskManagerRpcServices = null;
- }
- else {
// start a new service per component, possibly with custom bind addresses
final String jobManagerBindAddress = miniClusterConfiguration.getJobManagerBindAddress();
final String taskManagerBindAddress = miniClusterConfiguration.getTaskManagerBindAddress();
- final String resourceManagerBindAddress = miniClusterConfiguration.getResourceManagerBindAddress();
-
- jobManagerRpcService = createRpcService(akkaRpcServiceConfig, true, jobManagerBindAddress);
- resourceManagerRpcService = createRpcService(akkaRpcServiceConfig, true, resourceManagerBindAddress);
-
- for (int i = 0; i < numTaskManagers; i++) {
- taskManagerRpcServices[i] = createRpcService(akkaRpcServiceConfig, true, taskManagerBindAddress);
- }
- this.jobManagerRpcService = jobManagerRpcService;
- this.taskManagerRpcServices = taskManagerRpcServices;
- this.resourceManagerRpcService = resourceManagerRpcService;
+ dispatcherResourceManagreComponentRpcServiceFactory = new DedicatedRpcServiceFactory(akkaRpcServiceConfig, jobManagerBindAddress);
+ taskManagerRpcServiceFactory = new DedicatedRpcServiceFactory(akkaRpcServiceConfig, taskManagerBindAddress);
}
+ // TODO: Temporary hack until the metric query service is ported to the RpcEndpoint
+ metricQueryServiceActorSystem = MetricUtils.startMetricsActorSystem(
+ configuration,
+ commonRpcService.getAddress(),
+ LOG);
+ metricRegistry.startQueryService(metricQueryServiceActorSystem, null);
+
haServices = createHighAvailabilityServices(configuration, commonRpcService.getExecutor());
blobServer = new BlobServer(configuration, haServices.createBlobStore());
@@ -320,17 +282,6 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
- // bring up the ResourceManager(s)
- LOG.info("Starting ResourceManger");
- resourceManagerRunner = startResourceManager(
- configuration,
- haServices,
- heartbeatServices,
- metricRegistry,
- resourceManagerRpcService,
- new ClusterInformation("localhost", blobServer.getPort()),
- jobManagerMetricGroup);
-
blobCacheService = new BlobCacheService(
configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())
);
@@ -344,82 +295,44 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
metricRegistry,
blobCacheService,
numTaskManagers,
- taskManagerRpcServices);
+ taskManagerRpcServiceFactory);
- // starting the dispatcher rest endpoint
- LOG.info("Starting dispatcher rest endpoint.");
+ MetricQueryServiceRetriever metricQueryServiceRetriever = new AkkaQueryServiceRetriever(
+ metricQueryServiceActorSystem,
+ Time.milliseconds(configuration.getLong(WebOptions.TIMEOUT)));
+
+ dispatcherResourceManagerComponents.addAll(createDispatcherResourceManagerComponents(
+ configuration,
+ dispatcherResourceManagreComponentRpcServiceFactory,
+ haServices,
+ blobServer,
+ heartbeatServices,
+ metricRegistry,
+ metricQueryServiceRetriever,
+ new ShutDownFatalErrorHandler()
+ ));
+
+ resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();
+ dispatcherLeaderRetriever = haServices.getDispatcherLeaderRetriever();
+ webMonitorLeaderRetrievalService = haServices.getWebMonitorLeaderRetriever();
dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
- jobManagerRpcService,
+ commonRpcService,
DispatcherGateway.class,
DispatcherId::fromUuid,
20,
Time.milliseconds(20L));
resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
- jobManagerRpcService,
+ commonRpcService,
ResourceManagerGateway.class,
ResourceManagerId::fromUuid,
20,
Time.milliseconds(20L));
-
- final ExecutorService executor = WebMonitorEndpoint.createExecutorService(
- configuration.getInteger(RestOptions.SERVER_NUM_THREADS, 1),
- configuration.getInteger(RestOptions.SERVER_THREAD_PRIORITY),
- "DispatcherRestEndpoint");
-
- final long updateInterval = configuration.getLong(MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL);
- final MetricFetcher metricFetcher = updateInterval == 0
- ? VoidMetricFetcher.INSTANCE
- : MetricFetcherImpl.fromConfiguration(configuration,
- new AkkaQueryServiceRetriever(
- metricQueryServiceActorSystem,
- Time.milliseconds(
- configuration.getLong(WebOptions.TIMEOUT))),
- dispatcherGatewayRetriever, executor);
-
- this.dispatcherRestEndpoint = new DispatcherRestEndpoint(
- RestServerEndpointConfiguration.fromConfiguration(configuration),
- dispatcherGatewayRetriever,
- configuration,
- RestHandlerConfiguration.fromConfiguration(configuration),
- resourceManagerGatewayRetriever,
- blobServer.getTransientBlobService(),
- executor,
- metricFetcher,
- haServices.getWebMonitorLeaderElectionService(),
- new ShutDownFatalErrorHandler());
-
- this.dispatcherRestEndpoint.start();
-
- restAddressURI = new URI(this.dispatcherRestEndpoint.getRestBaseUrl());
-
- // bring up the dispatcher that launches JobManagers when jobs submitted
- LOG.info("Starting job dispatcher(s) for JobManger");
-
- final HistoryServerArchivist historyServerArchivist = HistoryServerArchivist.createHistoryServerArchivist(configuration, this.dispatcherRestEndpoint);
-
- dispatcher = new StandaloneDispatcher(
- jobManagerRpcService,
- Dispatcher.DISPATCHER_NAME + UUID.randomUUID(),
- configuration,
- haServices,
- resourceManagerRunner.getResourceManageGateway(),
- blobServer,
- heartbeatServices,
- jobManagerMetricGroup,
- metricRegistry.getMetricQueryServicePath(),
- new MemoryArchivedExecutionGraphStore(),
- DefaultJobManagerRunnerFactory.INSTANCE,
- new ShutDownFatalErrorHandler(),
- historyServerArchivist);
-
- dispatcher.start();
-
- resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();
- dispatcherLeaderRetriever = haServices.getDispatcherLeaderRetriever();
+ webMonitorLeaderRetriever = new LeaderRetriever();
resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);
dispatcherLeaderRetriever.start(dispatcherGatewayRetriever);
+ webMonitorLeaderRetrievalService.start(webMonitorLeaderRetriever);
}
catch (Exception e) {
// cleanup everything
@@ -441,6 +354,29 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
}
}
+ private Collection<? extends DispatcherResourceManagerComponent<?>> createDispatcherResourceManagerComponents(
+ Configuration configuration,
+ RpcServiceFactory rpcServiceFactory,
+ HighAvailabilityServices haServices,
+ BlobServer blobServer,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ MetricQueryServiceRetriever metricQueryServiceRetriever,
+ FatalErrorHandler fatalErrorHandler) throws Exception {
+ SessionDispatcherResourceManagerComponentFactory dispatcherResourceManagerComponentFactory = new SessionDispatcherResourceManagerComponentFactory(StandaloneResourceManagerFactory.INSTANCE);
+ return Collections.singleton(
+ dispatcherResourceManagerComponentFactory.create(
+ configuration,
+ rpcServiceFactory.createRpcService(),
+ haServices,
+ blobServer,
+ heartbeatServices,
+ metricRegistry,
+ new MemoryArchivedExecutionGraphStore(),
+ metricQueryServiceRetriever,
+ fatalErrorHandler));
+ }
+
protected HighAvailabilityServices createHighAvailabilityServices(Configuration configuration, Executor executor) throws Exception {
LOG.info("Starting high-availability services");
return HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
@@ -476,12 +412,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
taskManagers = null;
}
- componentTerminationFutures.add(shutDownDispatcher());
-
- if (resourceManagerRunner != null) {
- componentTerminationFutures.add(resourceManagerRunner.closeAsync());
- resourceManagerRunner = null;
- }
+ componentTerminationFutures.add(shutDownResourceManagerComponents());
final FutureUtils.ConjunctFuture<Void> componentsTerminationFuture = FutureUtils.completeAll(componentTerminationFutures);
@@ -516,11 +447,6 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
private CompletableFuture<Void> closeMetricSystem() {
synchronized (lock) {
- if (jobManagerMetricGroup != null) {
- jobManagerMetricGroup.close();
- jobManagerMetricGroup = null;
- }
-
final ArrayList<CompletableFuture<Void>> terminationFutures = new ArrayList<>(2);
// metrics shutdown
@@ -769,7 +695,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
MetricRegistry metricRegistry,
BlobCacheService blobCacheService,
int numTaskManagers,
- RpcService[] taskManagerRpcServices) throws Exception {
+ RpcServiceFactory rpcServiceFactory) throws Exception {
final TaskExecutor[] taskExecutors = new TaskExecutor[numTaskManagers];
final boolean localCommunication = numTaskManagers == 1;
@@ -778,7 +704,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
taskExecutors[i] = TaskManagerRunner.startTaskManager(
configuration,
new ResourceID(UUID.randomUUID().toString()),
- taskManagerRpcServices[i],
+ rpcServiceFactory.createRpcService(),
haServices,
heartbeatServices,
metricRegistry,
@@ -797,22 +723,12 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
// ------------------------------------------------------------------------
@GuardedBy("lock")
- private CompletableFuture<Void> shutDownDispatcher() {
-
- final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(2);
-
- // cancel all jobs and shut down the job dispatcher
- if (dispatcher != null) {
- dispatcher.shutDown();
- terminationFutures.add(dispatcher.getTerminationFuture());
-
- dispatcher = null;
- }
+ private CompletableFuture<Void> shutDownResourceManagerComponents() {
- if (dispatcherRestEndpoint != null) {
- terminationFutures.add(dispatcherRestEndpoint.closeAsync());
+ final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(dispatcherResourceManagerComponents.size());
- dispatcherRestEndpoint = null;
+ for (DispatcherResourceManagerComponent<?> dispatcherResourceManagerComponent : dispatcherResourceManagerComponents) {
+ terminationFutures.add(dispatcherResourceManagerComponent.closeAsync());
}
final FutureUtils.ConjunctFuture<Void> dispatcherTerminationFuture = FutureUtils.completeAll(terminationFutures);
@@ -842,6 +758,16 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
dispatcherLeaderRetriever = null;
}
+
+ if (webMonitorLeaderRetrievalService != null) {
+ try {
+ webMonitorLeaderRetrievalService.stop();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ webMonitorLeaderRetrievalService = null;
+ }
}
if (exception != null) {
@@ -893,74 +819,80 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
@Nonnull
private CompletionStage<Void> terminateRpcServices() {
- final int numRpcServices;
- if (miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED) {
- numRpcServices = 1;
- } else {
- numRpcServices = 1 + 2 + miniClusterConfiguration.getNumTaskManagers(); // common, JM, RM, TMs
- }
+ synchronized (lock) {
+ final int numRpcServices = 1 + rpcServices.size();
- final Collection<CompletableFuture<?>> rpcTerminationFutures = new ArrayList<>(numRpcServices);
+ final Collection<CompletableFuture<?>> rpcTerminationFutures = new ArrayList<>(numRpcServices);
- synchronized (lock) {
rpcTerminationFutures.add(commonRpcService.stopService());
- if (miniClusterConfiguration.getRpcServiceSharing() != RpcServiceSharing.SHARED) {
- rpcTerminationFutures.add(jobManagerRpcService.stopService());
- rpcTerminationFutures.add(resourceManagerRpcService.stopService());
-
- for (RpcService taskManagerRpcService : taskManagerRpcServices) {
- rpcTerminationFutures.add(taskManagerRpcService.stopService());
- }
+ for (RpcService rpcService : rpcServices) {
+ rpcTerminationFutures.add(rpcService.stopService());
}
commonRpcService = null;
- jobManagerRpcService = null;
- taskManagerRpcServices = null;
- resourceManagerRpcService = null;
+ rpcServices.clear();
+
+ return FutureUtils.completeAll(rpcTerminationFutures);
}
+ }
- return FutureUtils.completeAll(rpcTerminationFutures);
+ /**
+ * Internal factory for {@link RpcService}.
+ */
+ protected interface RpcServiceFactory {
+ RpcService createRpcService();
}
- // ------------------------------------------------------------------------
- // miscellaneous utilities
- // ------------------------------------------------------------------------
+ /**
+ * Factory which returns always the common {@link RpcService}.
+ */
+ protected static class CommonRpcServiceFactory implements RpcServiceFactory {
- private void initializeIOFormatClasses(Configuration configuration) {
- // TODO: That we still have to call something like this is a crime against humanity
- FileOutputFormat.initDefaultsFromConfiguration(configuration);
- }
+ private final RpcService commonRpcService;
- private static Throwable shutDownRpc(RpcService rpcService, Throwable priorException) {
- if (rpcService != null) {
- try {
- rpcService.stopService().get();
- }
- catch (Throwable t) {
- return ExceptionUtils.firstOrSuppressed(t, priorException);
- }
+ CommonRpcServiceFactory(RpcService commonRpcService) {
+ this.commonRpcService = commonRpcService;
}
- return priorException;
+ @Override
+ public RpcService createRpcService() {
+ return commonRpcService;
+ }
}
- private static Throwable shutDownRpcs(RpcService[] rpcServices, Throwable priorException) {
- if (rpcServices != null) {
- Throwable exception = priorException;
+ /**
+ * Factory which creates and registers new {@link RpcService}.
+ */
+ protected class DedicatedRpcServiceFactory implements RpcServiceFactory {
- for (RpcService service : rpcServices) {
- try {
- if (service != null) {
- service.stopService().get();
- }
- }
- catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
+ private final AkkaRpcServiceConfiguration akkaRpcServiceConfig;
+ private final String jobManagerBindAddress;
+
+ DedicatedRpcServiceFactory(AkkaRpcServiceConfiguration akkaRpcServiceConfig, String jobManagerBindAddress) {
+ this.akkaRpcServiceConfig = akkaRpcServiceConfig;
+ this.jobManagerBindAddress = jobManagerBindAddress;
+ }
+
+ @Override
+ public RpcService createRpcService() {
+ final RpcService rpcService = MiniCluster.this.createRpcService(akkaRpcServiceConfig, true, jobManagerBindAddress);
+
+ synchronized (lock) {
+ rpcServices.add(rpcService);
}
+
+ return rpcService;
}
- return priorException;
+ }
+
+ // ------------------------------------------------------------------------
+ // miscellaneous utilities
+ // ------------------------------------------------------------------------
+
+ private void initializeIOFormatClasses(Configuration configuration) {
+ // TODO: That we still have to call something like this is a crime against humanity
+ FileOutputFormat.initDefaultsFromConfiguration(configuration);
}
private class TerminatingFatalErrorHandler implements FatalErrorHandler {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
index ca5d01d..a39bfd4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
@@ -87,12 +87,6 @@ public class MiniClusterConfiguration {
configuration.getString(TaskManagerOptions.HOST, "localhost");
}
- public String getResourceManagerBindAddress() {
- return commonBindAddress != null ?
- commonBindAddress :
- configuration.getString(JobManagerOptions.ADDRESS, "localhost"); // TODO: Introduce proper configuration constant for the resource manager hostname
- }
-
public Time getRpcTimeout() {
FiniteDuration duration = AkkaUtils.getTimeout(configuration);
return Time.of(duration.length(), duration.unit());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/MiniClusterResource.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/MiniClusterResource.java
index 5cd1a50..1c4a32a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/MiniClusterResource.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/MiniClusterResource.java
@@ -74,7 +74,7 @@ public class MiniClusterResource extends ExternalResource {
}
public URI getRestAddres() {
- return miniCluster.getRestAddress();
+ return miniCluster.getRestAddress().join();
}
@Override
@@ -139,7 +139,7 @@ public class MiniClusterResource extends ExternalResource {
miniCluster.start();
- final URI restAddress = miniCluster.getRestAddress();
+ final URI restAddress = miniCluster.getRestAddress().get();
createClientConfiguration(restAddress);
}
diff --git a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
index 835c01c..bc2d91a 100644
--- a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
+++ b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
@@ -151,7 +151,7 @@ object FlinkShell {
.build()
val cluster = new MiniCluster(miniClusterConfig)
cluster.start()
- val port = cluster.getRestAddress.getPort
+ val port = cluster.getRestAddress.get.getPort
println(s"\nStarting local Flink cluster (host: localhost, port: $port).\n")
("localhost", port, Some(Left(cluster)))
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
index ec80973..bde8cb4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java
@@ -118,7 +118,7 @@ public class LocalStreamEnvironment extends StreamExecutionEnvironment {
try {
miniCluster.start();
- configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().getPort());
+ configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());
return miniCluster.executeJobBlocking(jobGraph);
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/operators/RemoteEnvironmentITCase.java b/flink-tests/src/test/java/org/apache/flink/test/operators/RemoteEnvironmentITCase.java
index b8b897d..4df9ae6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/operators/RemoteEnvironmentITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/operators/RemoteEnvironmentITCase.java
@@ -65,7 +65,7 @@ public class RemoteEnvironmentITCase extends TestLogger {
Configuration config = new Configuration();
config.setString(AkkaOptions.STARTUP_TIMEOUT, VALID_STARTUP_TIMEOUT);
- final URI restAddress = MINI_CLUSTER_RESOURCE.getMiniCluster().getRestAddress();
+ final URI restAddress = MINI_CLUSTER_RESOURCE.getRestAddres();
final String hostname = restAddress.getHost();
final int port = restAddress.getPort();