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 2018/02/06 16:53:17 UTC
[02/11] flink git commit: [FLINK-8501] [flip6] Use single
BlobCacheService per TaskExecutor
[FLINK-8501] [flip6] Use single BlobCacheService per TaskExecutor
Instead of creating for each new JobManagerConnection a dedicated BlobCacheService
the TaskExecutor uses a single BlobCacheService which it shares between the
different JobManagerConnections. The initial BlobServer address is passed by the
ResourceManager when the TaskExecutor registers at it. In order to avoid the re-
creation of BlobCacheServices, this commit changes the behaviour such that one can
update the BlobServer address.
This closes #5350.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/22bf3861
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/22bf3861
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/22bf3861
Branch: refs/heads/master
Commit: 22bf3861c95ed93b2eb5f8d0661d282ef6cdbbc8
Parents: 9d0eb22
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Feb 6 11:01:36 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Feb 6 12:15:30 2018 +0100
----------------------------------------------------------------------
.../entrypoint/MesosJobClusterEntrypoint.java | 3 +
.../MesosSessionClusterEntrypoint.java | 3 +
.../clusterframework/MesosResourceManager.java | 3 +
.../MesosResourceManagerTest.java | 22 ++++-
.../flink/runtime/blob/AbstractBlobCache.java | 56 +++++++++----
.../flink/runtime/blob/BlobCacheService.java | 25 ++++--
.../flink/runtime/blob/PermanentBlobCache.java | 15 ++--
.../flink/runtime/blob/TransientBlobCache.java | 29 ++++---
.../apache/flink/runtime/client/JobClient.java | 2 +-
.../runtime/entrypoint/ClusterInformation.java | 51 ++++++++++++
.../entrypoint/JobClusterEntrypoint.java | 2 +
.../entrypoint/SessionClusterEntrypoint.java | 2 +
.../StandaloneSessionClusterEntrypoint.java | 2 +
.../jobmaster/JMTMRegistrationSuccess.java | 14 ++--
.../flink/runtime/jobmaster/JobMaster.java | 5 +-
.../flink/runtime/minicluster/MiniCluster.java | 86 ++++++++++++++------
.../resourcemanager/ResourceManager.java | 8 +-
.../resourcemanager/ResourceManagerRunner.java | 5 +-
.../StandaloneResourceManager.java | 3 +
.../handler/legacy/TaskManagerLogHandler.java | 2 +-
.../taskexecutor/JobManagerConnection.java | 16 +---
.../runtime/taskexecutor/TaskExecutor.java | 61 ++++++--------
.../TaskExecutorRegistrationSuccess.java | 27 ++++--
.../runtime/taskexecutor/TaskManagerRunner.java | 18 +++-
.../flink/runtime/taskmanager/TaskManager.scala | 4 +-
.../runtime/blob/BlobCacheCleanupTest.java | 10 +--
.../runtime/blob/BlobCacheCorruptionTest.java | 4 +-
.../flink/runtime/blob/BlobCacheDeleteTest.java | 16 ++--
.../flink/runtime/blob/BlobCacheGetTest.java | 24 +++---
.../flink/runtime/blob/BlobCachePutTest.java | 34 ++++----
.../runtime/blob/BlobCacheRecoveryTest.java | 8 +-
.../runtime/blob/BlobCacheRetriesTest.java | 8 +-
.../runtime/blob/BlobCacheSuccessTest.java | 4 +-
.../runtime/blob/BlobServerRecoveryTest.java | 4 +-
.../clusterframework/ResourceManagerTest.java | 3 +
.../BlobLibraryCacheManagerTest.java | 8 +-
.../BlobLibraryCacheRecoveryITCase.java | 4 +-
...ecutionGraphDeploymentWithBlobCacheTest.java | 2 +-
.../resourcemanager/ResourceManagerHATest.java | 2 +
.../ResourceManagerJobMasterTest.java | 2 +
.../ResourceManagerTaskExecutorTest.java | 2 +
.../resourcemanager/TestingResourceManager.java | 14 +++-
.../utils/TestingResourceManagerGateway.java | 4 +-
.../taskexecutor/TaskExecutorITCase.java | 9 +-
.../runtime/taskexecutor/TaskExecutorTest.java | 85 ++++++++++---------
.../apache/flink/yarn/YarnResourceManager.java | 3 +
.../entrypoint/YarnJobClusterEntrypoint.java | 3 +
.../YarnSessionClusterEntrypoint.java | 3 +
.../flink/yarn/YarnResourceManagerTest.java | 4 +
49 files changed, 471 insertions(+), 253 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
index d38e734..21864dc 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -132,6 +133,7 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl) throws Exception {
final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
@@ -150,6 +152,7 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
rmRuntimeServices.getSlotManager(),
metricRegistry,
rmRuntimeServices.getJobLeaderIdService(),
+ clusterInformation,
fatalErrorHandler,
configuration,
mesosServices,
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
index f52532a..68cf7c6 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.entrypoint.SessionClusterEntrypoint;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -122,6 +123,7 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl) throws Exception {
final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
@@ -140,6 +142,7 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
rmRuntimeServices.getSlotManager(),
metricRegistry,
rmRuntimeServices.getJobLeaderIdService(),
+ clusterInformation,
fatalErrorHandler,
configuration,
mesosServices,
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
index bb60465..d42d8d1 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
@@ -47,6 +47,7 @@ import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameter
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.metrics.MetricRegistry;
@@ -149,6 +150,7 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
SlotManager slotManager,
MetricRegistry metricRegistry,
JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
FatalErrorHandler fatalErrorHandler,
// Mesos specifics
Configuration flinkConfig,
@@ -166,6 +168,7 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
slotManager,
metricRegistry,
jobLeaderIdService,
+ clusterInformation,
fatalErrorHandler);
this.mesosServices = Preconditions.checkNotNull(mesosServices);
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
index 57fb2df..e23e0cb 100644
--- a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
+++ b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
@@ -43,6 +43,7 @@ import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameter
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -172,10 +173,23 @@ public class MesosResourceManagerTest extends TestLogger {
MesosConfiguration mesosConfig,
MesosTaskManagerParameters taskManagerParameters,
ContainerSpecification taskManagerContainerSpec) {
- super(rpcService, resourceManagerEndpointId, resourceId, resourceManagerConfiguration,
- highAvailabilityServices, heartbeatServices, slotManager, metricRegistry,
- jobLeaderIdService, fatalErrorHandler, flinkConfig, mesosServices, mesosConfig,
- taskManagerParameters, taskManagerContainerSpec);
+ super(
+ rpcService,
+ resourceManagerEndpointId,
+ resourceId,
+ resourceManagerConfiguration,
+ highAvailabilityServices,
+ heartbeatServices,
+ slotManager,
+ metricRegistry,
+ jobLeaderIdService,
+ new ClusterInformation("localhost", 1234),
+ fatalErrorHandler,
+ flinkConfig,
+ mesosServices,
+ mesosConfig,
+ taskManagerParameters,
+ taskManagerContainerSpec);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java
index efb9ded..e0d63cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java
@@ -54,8 +54,6 @@ public abstract class AbstractBlobCache implements Closeable {
*/
protected final AtomicLong tempFileCounter = new AtomicLong(0);
- protected final InetSocketAddress serverAddress;
-
/**
* Root directory for local file storage.
*/
@@ -89,15 +87,16 @@ public abstract class AbstractBlobCache implements Closeable {
*/
protected final ReadWriteLock readWriteLock;
+ @Nullable
+ protected volatile InetSocketAddress serverAddress;
+
public AbstractBlobCache(
- final InetSocketAddress serverAddress,
final Configuration blobClientConfig,
final BlobView blobView,
- final Logger logger) throws IOException {
+ final Logger logger,
+ @Nullable final InetSocketAddress serverAddress) throws IOException {
this.log = checkNotNull(logger);
-
- this.serverAddress = checkNotNull(serverAddress);
this.blobClientConfig = checkNotNull(blobClientConfig);
this.blobView = checkNotNull(blobView);
this.readWriteLock = new ReentrantReadWriteLock();
@@ -118,6 +117,8 @@ public abstract class AbstractBlobCache implements Closeable {
// Add shutdown hook to delete storage directory
shutdownHook = BlobUtils.addShutdownHook(this, log);
+
+ this.serverAddress = serverAddress;
}
/**
@@ -172,16 +173,22 @@ public abstract class AbstractBlobCache implements Closeable {
log.info("Failed to copy from blob store. Downloading from BLOB server instead.", e);
}
- // fallback: download from the BlobServer
- BlobClient.downloadFromBlobServer(
- jobId, blobKey, incomingFile, serverAddress, blobClientConfig, numFetchRetries);
+ final InetSocketAddress currentServerAddress = serverAddress;
- readWriteLock.writeLock().lock();
- try {
- BlobUtils.moveTempFileToStore(
- incomingFile, jobId, blobKey, localFile, log, null);
- } finally {
- readWriteLock.writeLock().unlock();
+ if (currentServerAddress != null) {
+ // fallback: download from the BlobServer
+ BlobClient.downloadFromBlobServer(
+ jobId, blobKey, incomingFile, currentServerAddress, blobClientConfig, numFetchRetries);
+
+ readWriteLock.writeLock().lock();
+ try {
+ BlobUtils.moveTempFileToStore(
+ incomingFile, jobId, blobKey, localFile, log, null);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ } else {
+ throw new IOException("Cannot download from BlobServer, because the server address is unknown.");
}
return localFile;
@@ -197,10 +204,25 @@ public abstract class AbstractBlobCache implements Closeable {
/**
* Returns the port the BLOB server is listening on.
*
- * @return BLOB server port
+ * @return BLOB server port or {@code -1} if no server address
*/
public int getPort() {
- return serverAddress.getPort();
+ final InetSocketAddress currentServerAddress = serverAddress;
+
+ if (currentServerAddress != null) {
+ return serverAddress.getPort();
+ } else {
+ return -1;
+ }
+ }
+
+ /**
+ * Sets the address of the {@link BlobServer}.
+ *
+ * @param blobServerAddress address of the {@link BlobServer}.
+ */
+ public void setBlobServerAddress(InetSocketAddress blobServerAddress) {
+ serverAddress = checkNotNull(blobServerAddress);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCacheService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCacheService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCacheService.java
index 89ce2c4..b2ee100 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCacheService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCacheService.java
@@ -20,6 +20,8 @@ package org.apache.flink.runtime.blob;
import org.apache.flink.configuration.Configuration;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -39,23 +41,22 @@ public class BlobCacheService implements BlobService {
/**
* Instantiates a new BLOB cache.
*
- * @param serverAddress
- * address of the {@link BlobServer} to use for fetching files from
* @param blobClientConfig
* global configuration
* @param blobView
* (distributed) blob store file system to retrieve files from first
- *
+ * @param serverAddress
+ * address of the {@link BlobServer} to use for fetching files from or {@code null} if none yet
* @throws IOException
* thrown if the (local or distributed) file storage cannot be created or is not usable
*/
public BlobCacheService(
- final InetSocketAddress serverAddress,
final Configuration blobClientConfig,
- final BlobView blobView) throws IOException {
+ final BlobView blobView,
+ @Nullable final InetSocketAddress serverAddress) throws IOException {
- this(new PermanentBlobCache(serverAddress, blobClientConfig, blobView),
- new TransientBlobCache(serverAddress, blobClientConfig));
+ this(new PermanentBlobCache(blobClientConfig, blobView, serverAddress),
+ new TransientBlobCache(blobClientConfig, serverAddress));
}
/**
@@ -82,6 +83,16 @@ public class BlobCacheService implements BlobService {
return transientBlobCache;
}
+ /**
+ * Sets the address of the {@link BlobServer}.
+ *
+ * @param blobServerAddress address of the {@link BlobServer}.
+ */
+ public void setBlobServerAddress(InetSocketAddress blobServerAddress) {
+ permanentBlobCache.setBlobServerAddress(blobServerAddress);
+ transientBlobCache.setBlobServerAddress(blobServerAddress);
+ }
+
@Override
public void close() throws IOException {
permanentBlobCache.close();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobCache.java
index 83cb18e..88ea74a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobCache.java
@@ -26,6 +26,8 @@ import org.apache.flink.util.FileUtils;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -85,23 +87,22 @@ public class PermanentBlobCache extends AbstractBlobCache implements PermanentBl
/**
* Instantiates a new cache for permanent BLOBs which are also available in an HA store.
*
- * @param serverAddress
- * address of the {@link BlobServer} to use for fetching files from
* @param blobClientConfig
* global configuration
* @param blobView
* (distributed) HA blob store file system to retrieve files from first
- *
+ * @param serverAddress
+ * address of the {@link BlobServer} to use for fetching files from or {@code null} if none yet
* @throws IOException
* thrown if the (local or distributed) file storage cannot be created or is not usable
*/
public PermanentBlobCache(
- final InetSocketAddress serverAddress,
final Configuration blobClientConfig,
- final BlobView blobView) throws IOException {
+ final BlobView blobView,
+ @Nullable final InetSocketAddress serverAddress) throws IOException {
- super(serverAddress, blobClientConfig, blobView,
- LoggerFactory.getLogger(PermanentBlobCache.class));
+ super(blobClientConfig, blobView, LoggerFactory.getLogger(PermanentBlobCache.class), serverAddress
+ );
// Initializing the clean up task
this.cleanupTimer = new Timer(true);
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobCache.java
index bbd008a..aab2803 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobCache.java
@@ -67,20 +67,19 @@ public class TransientBlobCache extends AbstractBlobCache implements TransientBl
/**
* Instantiates a new BLOB cache.
*
- * @param serverAddress
- * address of the {@link BlobServer} to use for fetching files from
* @param blobClientConfig
* global configuration
- *
+ * @param serverAddress
+ * address of the {@link BlobServer} to use for fetching files from or {@code null} if none yet
* @throws IOException
* thrown if the (local or distributed) file storage cannot be created or is not usable
*/
public TransientBlobCache(
- final InetSocketAddress serverAddress,
- final Configuration blobClientConfig) throws IOException {
+ final Configuration blobClientConfig,
+ @Nullable final InetSocketAddress serverAddress) throws IOException {
- super(serverAddress, blobClientConfig, new VoidBlobStore(),
- LoggerFactory.getLogger(TransientBlobCache.class));
+ super(blobClientConfig, new VoidBlobStore(), LoggerFactory.getLogger(TransientBlobCache.class), serverAddress
+ );
// Initializing the clean up task
this.cleanupTimer = new Timer(true);
@@ -122,7 +121,7 @@ public class TransientBlobCache extends AbstractBlobCache implements TransientBl
@Override
public TransientBlobKey putTransient(byte[] value) throws IOException {
- try (BlobClient bc = new BlobClient(serverAddress, blobClientConfig)) {
+ try (BlobClient bc = createClient()) {
return (TransientBlobKey) bc.putBuffer(null, value, 0, value.length, TRANSIENT_BLOB);
}
}
@@ -130,14 +129,14 @@ public class TransientBlobCache extends AbstractBlobCache implements TransientBl
@Override
public TransientBlobKey putTransient(JobID jobId, byte[] value) throws IOException {
checkNotNull(jobId);
- try (BlobClient bc = new BlobClient(serverAddress, blobClientConfig)) {
+ try (BlobClient bc = createClient()) {
return (TransientBlobKey) bc.putBuffer(jobId, value, 0, value.length, TRANSIENT_BLOB);
}
}
@Override
public TransientBlobKey putTransient(InputStream inputStream) throws IOException {
- try (BlobClient bc = new BlobClient(serverAddress, blobClientConfig)) {
+ try (BlobClient bc = createClient()) {
return (TransientBlobKey) bc.putInputStream(null, inputStream, TRANSIENT_BLOB);
}
}
@@ -145,7 +144,7 @@ public class TransientBlobCache extends AbstractBlobCache implements TransientBl
@Override
public TransientBlobKey putTransient(JobID jobId, InputStream inputStream) throws IOException {
checkNotNull(jobId);
- try (BlobClient bc = new BlobClient(serverAddress, blobClientConfig)) {
+ try (BlobClient bc = createClient()) {
return (TransientBlobKey) bc.putInputStream(jobId, inputStream, TRANSIENT_BLOB);
}
}
@@ -222,7 +221,13 @@ public class TransientBlobCache extends AbstractBlobCache implements TransientBl
}
private BlobClient createClient() throws IOException {
- return new BlobClient(serverAddress, blobClientConfig);
+ final InetSocketAddress currentServerAddress = serverAddress;
+
+ if (currentServerAddress != null) {
+ return new BlobClient(serverAddress, blobClientConfig);
+ } else {
+ throw new IOException("Could not create BlobClient because the BlobServer address is unknown.");
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
index 82c40e0..4b84857 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
@@ -206,7 +206,7 @@ public class JobClient {
final PermanentBlobCache permanentBlobCache;
try {
// TODO: Fix lifecycle of PermanentBlobCache to properly close it upon usage
- permanentBlobCache = new PermanentBlobCache(serverAddress, config, highAvailabilityServices.createBlobStore());
+ permanentBlobCache = new PermanentBlobCache(config, highAvailabilityServices.createBlobStore(), serverAddress);
} catch (IOException e) {
throw new JobRetrievalException(
jobID,
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterInformation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterInformation.java
new file mode 100644
index 0000000..1d9f039
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterInformation.java
@@ -0,0 +1,51 @@
+/*
+ * 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.entrypoint;
+
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+
+/**
+ * Information about the cluster which is shared with the cluster components.
+ */
+public class ClusterInformation implements Serializable {
+
+ private static final long serialVersionUID = 316958921518479205L;
+
+ private final String blobServerHostname;
+
+ private final int blobServerPort;
+
+ public ClusterInformation(String blobServerHostname, int blobServerPort) {
+ this.blobServerHostname = Preconditions.checkNotNull(blobServerHostname);
+ Preconditions.checkArgument(
+ 0 < blobServerPort && blobServerPort < 65_536,
+ "The blob port must between 0 and 65_536. However, it was " + blobServerPort + '.');
+ this.blobServerPort = blobServerPort;
+ }
+
+ public String getBlobServerHostname() {
+ return blobServerHostname;
+ }
+
+ public int getBlobServerPort() {
+ return blobServerPort;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
index e52f113..1cf3de2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
@@ -134,6 +134,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
heartbeatServices,
metricRegistry,
this,
+ new ClusterInformation(rpcService.getAddress(), blobServer.getPort()),
jobMasterRestEndpoint.getRestAddress());
jobManagerRunner = createJobManagerRunner(
@@ -288,6 +289,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl) throws Exception;
protected abstract JobGraph retrieveJobGraph(Configuration configuration) throws FlinkException;
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
index fe9bd92..86f0fe3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
@@ -136,6 +136,7 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
heartbeatServices,
metricRegistry,
this,
+ new ClusterInformation(rpcService.getAddress(), blobServer.getPort()),
dispatcherRestEndpoint.getRestAddress());
dispatcher = createDispatcher(
@@ -285,5 +286,6 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java
index 8d01196..0f205f1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java
@@ -55,6 +55,7 @@ public class StandaloneSessionClusterEntrypoint extends SessionClusterEntrypoint
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl) throws Exception {
final ResourceManagerConfiguration resourceManagerConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
final ResourceManagerRuntimeServicesConfiguration resourceManagerRuntimeServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
@@ -73,6 +74,7 @@ public class StandaloneSessionClusterEntrypoint extends SessionClusterEntrypoint
resourceManagerRuntimeServices.getSlotManager(),
metricRegistry,
resourceManagerRuntimeServices.getJobLeaderIdService(),
+ clusterInformation,
fatalErrorHandler);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JMTMRegistrationSuccess.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JMTMRegistrationSuccess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JMTMRegistrationSuccess.java
index 7272cd4..34e38ac 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JMTMRegistrationSuccess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JMTMRegistrationSuccess.java
@@ -20,26 +20,22 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.util.Preconditions;
+/**
+ * Message indicating a successful {@link JobMaster} and {@link TaskExecutor} registration.
+ */
public class JMTMRegistrationSuccess extends RegistrationResponse.Success {
private static final long serialVersionUID = -3528383155961318929L;
private final ResourceID resourceID;
- private final int blobPort;
-
- public JMTMRegistrationSuccess(ResourceID resourceID, int blobPort) {
- Preconditions.checkArgument(0 < blobPort && 65536 > blobPort, "The blob port has to be 0 < blobPort < 65536.");
+ public JMTMRegistrationSuccess(ResourceID resourceID) {
this.resourceID = Preconditions.checkNotNull(resourceID);
- this.blobPort = blobPort;
}
public ResourceID getResourceID() {
return resourceID;
}
-
- public int getBlobPort() {
- return blobPort;
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 8d47655..f137f39 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -753,8 +753,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
final ResourceID taskManagerId = taskManagerLocation.getResourceID();
if (registeredTaskManagers.containsKey(taskManagerId)) {
- final RegistrationResponse response = new JMTMRegistrationSuccess(
- resourceId, blobServer.getPort());
+ final RegistrationResponse response = new JMTMRegistrationSuccess(resourceId);
return CompletableFuture.completedFuture(response);
} else {
return getRpcService()
@@ -781,7 +780,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
}
});
- return new JMTMRegistrationSuccess(resourceId, blobServer.getPort());
+ return new JMTMRegistrationSuccess(resourceId);
},
getMainThreadExecutor());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
----------------------------------------------------------------------
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 531c1a1..c54bf54 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
@@ -22,11 +22,13 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.client.JobExecutionException;
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.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
@@ -52,6 +54,8 @@ import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.GuardedBy;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -59,17 +63,20 @@ import static org.apache.flink.util.ExceptionUtils.firstOrSuppressed;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
+/**
+ * Flip-6 based MiniCluster.
+ */
public class MiniCluster implements JobExecutorService {
private static final Logger LOG = LoggerFactory.getLogger(MiniCluster.class);
- /** The lock to guard startup / shutdown / manipulation methods */
+ /** The lock to guard startup / shutdown / manipulation methods. */
private final Object lock = new Object();
- /** The configuration for this mini cluster */
+ /** The configuration for this mini cluster. */
private final MiniClusterConfiguration miniClusterConfiguration;
- @GuardedBy("lock")
+ @GuardedBy("lock")
private MetricRegistryImpl metricRegistry;
@GuardedBy("lock")
@@ -94,6 +101,9 @@ public class MiniCluster implements JobExecutorService {
private HeartbeatServices heartbeatServices;
@GuardedBy("lock")
+ private BlobCacheService blobCacheService;
+
+ @GuardedBy("lock")
private ResourceManagerRunner[] resourceManagerRunners;
private volatile TaskExecutor[] taskManagers;
@@ -101,14 +111,14 @@ public class MiniCluster implements JobExecutorService {
@GuardedBy("lock")
private MiniClusterJobDispatcher jobDispatcher;
- /** Flag marking the mini cluster as started/running */
+ /** Flag marking the mini cluster as started/running. */
private volatile boolean running;
// ------------------------------------------------------------------------
/**
* Creates a new Flink mini cluster based on the given configuration.
- *
+ *
* @param miniClusterConfiguration The configuration for the mini cluster
*/
public MiniCluster(MiniClusterConfiguration miniClusterConfiguration) {
@@ -130,7 +140,7 @@ public class MiniCluster implements JobExecutorService {
/**
* Starts the mini cluster, based on the configured properties.
- *
+ *
* @throws Exception This method passes on any exception that occurs during the startup of
* the mini cluster.
*/
@@ -227,12 +237,23 @@ public class MiniCluster implements JobExecutorService {
heartbeatServices,
metricRegistry,
numResourceManagers,
- resourceManagerRpcServices);
+ resourceManagerRpcServices,
+ new ClusterInformation("localhost", blobServer.getPort()));
+
+ blobCacheService = new BlobCacheService(
+ configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())
+ );
// bring up the TaskManager(s) for the mini cluster
LOG.info("Starting {} TaskManger(s)", numTaskManagers);
taskManagers = startTaskManagers(
- configuration, haServices, metricRegistry, numTaskManagers, taskManagerRpcServices);
+ configuration,
+ haServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ numTaskManagers,
+ taskManagerRpcServices);
// bring up the dispatcher that launches JobManagers when jobs submitted
LOG.info("Starting job dispatcher(s) for {} JobManger(s)", numJobManagers);
@@ -265,10 +286,10 @@ public class MiniCluster implements JobExecutorService {
/**
* Shuts down the mini cluster, failing all currently executing jobs.
* The mini cluster can be started again by calling the {@link #start()} method again.
- *
+ *
* <p>This method shuts down all started services and components,
- * even if an exception occurs in the process of shutting down some component.
- *
+ * even if an exception occurs in the process of shutting down some component.
+ *
* @throws Exception Thrown, if the shutdown did not complete cleanly.
*/
public void shutdown() throws Exception {
@@ -331,7 +352,6 @@ public class MiniCluster implements JobExecutorService {
}
taskManagers = null;
}
-
// metrics shutdown
if (metricRegistry != null) {
metricRegistry.shutdown();
@@ -348,6 +368,16 @@ public class MiniCluster implements JobExecutorService {
taskManagerRpcServices = null;
resourceManagerRpcServices = null;
+ if (blobCacheService != null) {
+ try {
+ blobCacheService.close();
+ } catch (Exception e) {
+ exception = firstOrSuppressed(e, exception);
+ }
+ blobCacheService = null;
+ }
+
+
// shut down the blob server
if (blobServer != null) {
try {
@@ -385,20 +415,23 @@ public class MiniCluster implements JobExecutorService {
OneTimeLeaderListenerFuture listenerFuture = new OneTimeLeaderListenerFuture();
rmMasterListener = haServices.getResourceManagerLeaderRetriever();
rmMasterListener.start(listenerFuture);
- addressAndIdFuture = listenerFuture.future();
+ addressAndIdFuture = listenerFuture.future();
}
final LeaderAddressAndId addressAndId = addressAndIdFuture.get();
- final ResourceManagerGateway resourceManager =
- commonRpcService.connect(addressAndId.leaderAddress(), new ResourceManagerId(addressAndId.leaderId()), ResourceManagerGateway.class).get();
+ final ResourceManagerGateway resourceManager = commonRpcService
+ .connect(
+ addressAndId.leaderAddress(),
+ new ResourceManagerId(addressAndId.leaderId()),
+ ResourceManagerGateway.class)
+ .get();
final int numTaskManagersToWaitFor = taskManagers.length;
// poll and wait until enough TaskManagers are available
while (true) {
- int numTaskManagersAvailable =
- resourceManager.getNumberOfRegisteredTaskManagers().get();
+ int numTaskManagersAvailable = resourceManager.getNumberOfRegisteredTaskManagers().get();
if (numTaskManagersAvailable >= numTaskManagersToWaitFor) {
break;
@@ -443,7 +476,7 @@ public class MiniCluster implements JobExecutorService {
* This method runs a job in blocking mode. The method returns only after the job
* completed successfully, or after it failed terminally.
*
- * @param job The Flink job to execute
+ * @param job The Flink job to execute
* @return The result of the job execution
*
* @throws JobExecutionException Thrown if anything went amiss during initial job launch,
@@ -471,8 +504,8 @@ public class MiniCluster implements JobExecutorService {
// ------------------------------------------------------------------------
/**
- * Factory method to create the metric registry for the mini cluster
- *
+ * Factory method to create the metric registry for the mini cluster.
+ *
* @param config The configuration of the mini cluster
*/
protected MetricRegistryImpl createMetricRegistry(Configuration config) {
@@ -481,7 +514,7 @@ public class MiniCluster implements JobExecutorService {
/**
* Factory method to instantiate the RPC service.
- *
+ *
* @param configuration
* The configuration of the mini cluster
* @param askTimeout
@@ -490,7 +523,7 @@ public class MiniCluster implements JobExecutorService {
* True, if the RPC service should be reachable from other (remote) RPC services.
* @param bindAddress
* The address to bind the RPC service to. Only relevant when "remoteEnabled" is true.
- *
+ *
* @return The instantiated RPC service
*/
protected RpcService createRpcService(
@@ -515,7 +548,8 @@ public class MiniCluster implements JobExecutorService {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
int numResourceManagers,
- RpcService[] resourceManagerRpcServices) throws Exception {
+ RpcService[] resourceManagerRpcServices,
+ ClusterInformation clusterInformation) throws Exception {
final ResourceManagerRunner[] resourceManagerRunners = new ResourceManagerRunner[numResourceManagers];
@@ -528,7 +562,8 @@ public class MiniCluster implements JobExecutorService {
resourceManagerRpcServices[i],
haServices,
heartbeatServices,
- metricRegistry);
+ metricRegistry,
+ clusterInformation);
resourceManagerRunners[i].start();
}
@@ -539,7 +574,9 @@ public class MiniCluster implements JobExecutorService {
protected TaskExecutor[] startTaskManagers(
Configuration configuration,
HighAvailabilityServices haServices,
+ HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
int numTaskManagers,
RpcService[] taskManagerRpcServices) throws Exception {
@@ -554,6 +591,7 @@ public class MiniCluster implements JobExecutorService {
haServices,
heartbeatServices,
metricRegistry,
+ blobCacheService,
localCommunication,
new TerminatingFatalErrorHandler(i));
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index e5fef14..7db127d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatListener;
import org.apache.flink.runtime.heartbeat.HeartbeatManager;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
@@ -130,6 +131,8 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
/** The slot manager maintains the available slots. */
private final SlotManager slotManager;
+ private final ClusterInformation clusterInformation;
+
/** The service to elect a ResourceManager leader. */
private LeaderElectionService leaderElectionService;
@@ -146,6 +149,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
SlotManager slotManager,
MetricRegistry metricRegistry,
JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
FatalErrorHandler fatalErrorHandler) {
super(rpcService, resourceManagerEndpointId);
@@ -156,6 +160,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
this.slotManager = checkNotNull(slotManager);
this.metricRegistry = checkNotNull(metricRegistry);
this.jobLeaderIdService = checkNotNull(jobLeaderIdService);
+ this.clusterInformation = checkNotNull(clusterInformation);
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender(
@@ -699,7 +704,8 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
return new TaskExecutorRegistrationSuccess(
registration.getInstanceID(),
resourceId,
- resourceManagerConfiguration.getHeartbeatInterval().toMilliseconds());
+ resourceManagerConfiguration.getHeartbeatInterval().toMilliseconds(),
+ clusterInformation);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
index caa3ba0..58a2789 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.resourcemanager;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.metrics.MetricRegistry;
@@ -55,7 +56,8 @@ public class ResourceManagerRunner implements FatalErrorHandler {
final RpcService rpcService,
final HighAvailabilityServices highAvailabilityServices,
final HeartbeatServices heartbeatServices,
- final MetricRegistry metricRegistry) throws Exception {
+ final MetricRegistry metricRegistry,
+ final ClusterInformation clusterInformation) throws Exception {
Preconditions.checkNotNull(resourceId);
Preconditions.checkNotNull(configuration);
@@ -83,6 +85,7 @@ public class ResourceManagerRunner implements FatalErrorHandler {
resourceManagerRuntimeServices.getSlotManager(),
metricRegistry,
resourceManagerRuntimeServices.getJobLeaderIdService(),
+ clusterInformation,
this);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
index 886a046..7226d29 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.resourcemanager;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.metrics.MetricRegistry;
@@ -49,6 +50,7 @@ public class StandaloneResourceManager extends ResourceManager<ResourceID> {
SlotManager slotManager,
MetricRegistry metricRegistry,
JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
FatalErrorHandler fatalErrorHandler) {
super(
rpcService,
@@ -60,6 +62,7 @@ public class StandaloneResourceManager extends ResourceManager<ResourceID> {
slotManager,
metricRegistry,
jobLeaderIdService,
+ clusterInformation,
fatalErrorHandler);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/TaskManagerLogHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/TaskManagerLogHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/TaskManagerLogHandler.java
index e4ad461..ed06d3d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/TaskManagerLogHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/TaskManagerLogHandler.java
@@ -154,7 +154,7 @@ public class TaskManagerLogHandler extends RedirectHandler<JobManagerGateway> im
cache = blobPortFuture.thenApplyAsync(
(Integer port) -> {
try {
- return new TransientBlobCache(new InetSocketAddress(jobManagerGateway.getHostname(), port), config);
+ return new TransientBlobCache(config, new InetSocketAddress(jobManagerGateway.getHostname(), port));
} catch (IOException e) {
throw new CompletionException(new FlinkException("Could not create TransientBlobCache.", e));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobManagerConnection.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobManagerConnection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobManagerConnection.java
index 7e92285..943901f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobManagerConnection.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobManagerConnection.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
import org.apache.flink.runtime.io.network.netty.PartitionProducerStateChecker;
@@ -50,9 +49,6 @@ public class JobManagerConnection {
// Checkpoint responder for the specific job manager
private final CheckpointResponder checkpointResponder;
- // BLOB cache connected to the BLOB server at the specific job manager
- private final BlobCacheService blobService;
-
// Library cache manager connected to the specific job manager
private final LibraryCacheManager libraryCacheManager;
@@ -68,7 +64,7 @@ public class JobManagerConnection {
JobMasterGateway jobMasterGateway,
TaskManagerActions taskManagerActions,
CheckpointResponder checkpointResponder,
- BlobCacheService blobService, LibraryCacheManager libraryCacheManager,
+ LibraryCacheManager libraryCacheManager,
ResultPartitionConsumableNotifier resultPartitionConsumableNotifier,
PartitionProducerStateChecker partitionStateChecker) {
this.jobID = Preconditions.checkNotNull(jobID);
@@ -76,7 +72,6 @@ public class JobManagerConnection {
this.jobMasterGateway = Preconditions.checkNotNull(jobMasterGateway);
this.taskManagerActions = Preconditions.checkNotNull(taskManagerActions);
this.checkpointResponder = Preconditions.checkNotNull(checkpointResponder);
- this.blobService = Preconditions.checkNotNull(blobService);
this.libraryCacheManager = Preconditions.checkNotNull(libraryCacheManager);
this.resultPartitionConsumableNotifier = Preconditions.checkNotNull(resultPartitionConsumableNotifier);
this.partitionStateChecker = Preconditions.checkNotNull(partitionStateChecker);
@@ -110,15 +105,6 @@ public class JobManagerConnection {
return libraryCacheManager;
}
- /**
- * Gets the BLOB cache connected to the respective BLOB server instance at the job manager.
- *
- * @return BLOB cache
- */
- public BlobCacheService getBlobService() {
- return blobService;
- }
-
public ResultPartitionConsumableNotifier getResultPartitionConsumableNotifier() {
return resultPartitionConsumableNotifier;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 6534c11..0e35f1e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -124,9 +125,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
/** The connection information of this task manager. */
private final TaskManagerLocation taskManagerLocation;
- /** Max blob port which is accepted. */
- public static final int MAX_BLOB_PORT = 65536;
-
/** The access to the leader election and retrieval services. */
private final HighAvailabilityServices haServices;
@@ -160,6 +158,8 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
private final FileCache fileCache;
+ private final BlobCacheService blobCacheService;
+
// --------- resource manager --------
private TaskExecutorToResourceManagerConnection resourceManagerConnection;
@@ -193,6 +193,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
TaskManagerMetricGroup taskManagerMetricGroup,
BroadcastVariableManager broadcastVariableManager,
FileCache fileCache,
+ BlobCacheService blobCacheService,
TaskSlotTable taskSlotTable,
JobManagerTable jobManagerTable,
JobLeaderService jobLeaderService,
@@ -214,6 +215,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
this.taskManagerMetricGroup = checkNotNull(taskManagerMetricGroup);
this.broadcastVariableManager = checkNotNull(broadcastVariableManager);
this.fileCache = checkNotNull(fileCache);
+ this.blobCacheService = checkNotNull(blobCacheService);
this.jobManagerTable = checkNotNull(jobManagerTable);
this.jobLeaderService = checkNotNull(jobLeaderService);
@@ -347,9 +349,8 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
// re-integrate offloaded data:
- BlobCacheService blobCache = jobManagerConnection.getBlobService();
try {
- tdd.loadBigData(blobCache.getPermanentBlobService());
+ tdd.loadBigData(blobCacheService.getPermanentBlobService());
} catch (IOException | ClassNotFoundException e) {
throw new TaskSubmissionException("Could not re-integrate offloaded TaskDeploymentDescriptor data.", e);
}
@@ -388,7 +389,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
TaskManagerActions taskManagerActions = jobManagerConnection.getTaskManagerActions();
CheckpointResponder checkpointResponder = jobManagerConnection.getCheckpointResponder();
- BlobCacheService blobService = jobManagerConnection.getBlobService();
LibraryCacheManager libraryCache = jobManagerConnection.getLibraryCacheManager();
ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = jobManagerConnection.getResultPartitionConsumableNotifier();
PartitionProducerStateChecker partitionStateChecker = jobManagerConnection.getPartitionStateChecker();
@@ -425,7 +425,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
taskManagerActions,
inputSplitProvider,
checkpointResponder,
- blobService,
+ blobCacheService,
libraryCache,
fileCache,
taskManagerConfiguration,
@@ -769,7 +769,9 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
}
- private void establishResourceManagerConnection(ResourceID resourceManagerResourceId) {
+ private void establishResourceManagerConnection(
+ ResourceID resourceManagerResourceId,
+ ClusterInformation clusterInformation) {
// monitor the resource manager as heartbeat target
resourceManagerHeartbeatManager.monitorTarget(resourceManagerResourceId, new HeartbeatTarget<SlotReport>() {
@Override
@@ -783,6 +785,13 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
// the TaskManager won't send heartbeat requests to the ResourceManager
}
});
+
+ // set the propagated blob server address
+ final InetSocketAddress blobServerAddress = new InetSocketAddress(
+ clusterInformation.getBlobServerHostname(),
+ clusterInformation.getBlobServerPort());
+
+ blobCacheService.setBlobServerAddress(blobServerAddress);
}
private void closeResourceManagerConnection(Exception cause) {
@@ -909,8 +918,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
JobManagerConnection newJobManagerConnection = associateWithJobManager(
jobId,
jobManagerResourceID,
- jobMasterGateway,
- registrationSuccess.getBlobPort());
+ jobMasterGateway);
jobManagerConnections.put(jobManagerResourceID, newJobManagerConnection);
jobManagerTable.put(jobId, newJobManagerConnection);
@@ -979,38 +987,19 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
private JobManagerConnection associateWithJobManager(
JobID jobID,
ResourceID resourceID,
- JobMasterGateway jobMasterGateway,
- int blobPort) {
+ JobMasterGateway jobMasterGateway) {
checkNotNull(jobID);
checkNotNull(resourceID);
checkNotNull(jobMasterGateway);
- checkArgument(blobPort > 0 && blobPort < MAX_BLOB_PORT, "Blob server port is out of range.");
TaskManagerActions taskManagerActions = new TaskManagerActionsImpl(jobMasterGateway);
CheckpointResponder checkpointResponder = new RpcCheckpointResponder(jobMasterGateway);
- InetSocketAddress blobServerAddress = new InetSocketAddress(jobMasterGateway.getHostname(), blobPort);
-
- final LibraryCacheManager libraryCacheManager;
- final BlobCacheService blobService;
- try {
- blobService = new BlobCacheService(
- blobServerAddress,
- taskManagerConfiguration.getConfiguration(),
- haServices.createBlobStore());
- libraryCacheManager = new BlobLibraryCacheManager(
- blobService.getPermanentBlobService(),
- taskManagerConfiguration.getClassLoaderResolveOrder(),
- taskManagerConfiguration.getAlwaysParentFirstLoaderPatterns());
- } catch (IOException e) {
- // Can't pass the IOException up - we need a RuntimeException anyway
- // two levels up where this is run asynchronously. Also, we don't
- // know whether this is caught in the thread running this method.
- final String message = "Could not create BLOB cache or library cache.";
- log.error(message, e);
- throw new RuntimeException(message, e);
- }
+ final LibraryCacheManager libraryCacheManager = new BlobLibraryCacheManager(
+ blobCacheService.getPermanentBlobService(),
+ taskManagerConfiguration.getClassLoaderResolveOrder(),
+ taskManagerConfiguration.getAlwaysParentFirstLoaderPatterns());
ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = new RpcResultPartitionConsumableNotifier(
jobMasterGateway,
@@ -1027,7 +1016,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
jobMasterGateway,
taskManagerActions,
checkpointResponder,
- blobService,
libraryCacheManager,
resultPartitionConsumableNotifier,
partitionStateChecker);
@@ -1051,7 +1039,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
JobMasterGateway jobManagerGateway = jobManagerConnection.getJobManagerGateway();
jobManagerGateway.disconnectTaskManager(getResourceID(), cause);
jobManagerConnection.getLibraryCacheManager().shutdown();
- jobManagerConnection.getBlobService().close();
}
private void registerQueryableState(JobID jobId, JobMasterGateway jobMasterGateway) {
@@ -1308,7 +1295,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
final ResourceID resourceManagerId = success.getResourceManagerId();
runAsync(
- () -> establishResourceManagerConnection(resourceManagerId)
+ () -> establishResourceManagerConnection(resourceManagerId, success.getClusterInformation())
);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
index 4b61f68..dc8d599 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
@@ -19,8 +19,10 @@
package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.util.Preconditions;
import java.io.Serializable;
@@ -38,17 +40,25 @@ public final class TaskExecutorRegistrationSuccess extends RegistrationResponse.
private final long heartbeatInterval;
+ private final ClusterInformation clusterInformation;
+
/**
* Create a new {@code TaskExecutorRegistrationSuccess} message.
*
- * @param registrationId The ID that the ResourceManager assigned the registration.
+ * @param registrationId The ID that the ResourceManager assigned the registration.
* @param resourceManagerResourceId The unique ID that identifies the ResourceManager.
- * @param heartbeatInterval The interval in which the ResourceManager will heartbeat the TaskExecutor.
+ * @param heartbeatInterval The interval in which the ResourceManager will heartbeat the TaskExecutor.
+ * @param clusterInformation information about the cluster
*/
- public TaskExecutorRegistrationSuccess(InstanceID registrationId, ResourceID resourceManagerResourceId, long heartbeatInterval) {
- this.registrationId = registrationId;
- this.resourceManagerResourceId = resourceManagerResourceId;
+ public TaskExecutorRegistrationSuccess(
+ InstanceID registrationId,
+ ResourceID resourceManagerResourceId,
+ long heartbeatInterval,
+ ClusterInformation clusterInformation) {
+ this.registrationId = Preconditions.checkNotNull(registrationId);
+ this.resourceManagerResourceId = Preconditions.checkNotNull(resourceManagerResourceId);
this.heartbeatInterval = heartbeatInterval;
+ this.clusterInformation = Preconditions.checkNotNull(clusterInformation);
}
/**
@@ -72,6 +82,13 @@ public final class TaskExecutorRegistrationSuccess extends RegistrationResponse.
return heartbeatInterval;
}
+ /**
+ * Gets the cluster information.
+ */
+ public ClusterInformation getClusterInformation() {
+ return clusterInformation;
+ }
+
@Override
public String toString() {
return "TaskExecutorRegistrationSuccess (" + registrationId + " / " + resourceManagerResourceId + " / " + heartbeatInterval + ')';
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index d089107..333ce12 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -88,6 +89,8 @@ public class TaskManagerRunner implements FatalErrorHandler {
private final MetricRegistryImpl metricRegistry;
+ private final BlobCacheService blobCacheService;
+
/** Executor used to run future callbacks. */
private final ExecutorService executor;
@@ -118,6 +121,10 @@ public class TaskManagerRunner implements FatalErrorHandler {
final ActorSystem actorSystem = ((AkkaRpcService) rpcService).getActorSystem();
metricRegistry.startQueryService(actorSystem, resourceId);
+ blobCacheService = new BlobCacheService(
+ configuration, highAvailabilityServices.createBlobStore(), null
+ );
+
taskManager = startTaskManager(
this.configuration,
this.resourceId,
@@ -125,6 +132,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
highAvailabilityServices,
heartbeatServices,
metricRegistry,
+ blobCacheService,
false,
this);
}
@@ -144,7 +152,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
protected void shutDownInternally() throws Exception {
Exception exception = null;
- synchronized(lock) {
+ synchronized (lock) {
try {
taskManager.shutDown();
} catch (Exception e) {
@@ -152,6 +160,12 @@ public class TaskManagerRunner implements FatalErrorHandler {
}
try {
+ blobCacheService.close();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ try {
metricRegistry.shutdown();
} catch (Exception e) {
exception = ExceptionUtils.firstOrSuppressed(e, exception);
@@ -252,6 +266,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
HighAvailabilityServices highAvailabilityServices,
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
boolean localCommunicationOnly,
FatalErrorHandler fatalErrorHandler) throws Exception {
@@ -292,6 +307,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
taskManagerMetricGroup,
taskManagerServices.getBroadcastVariableManager(),
taskManagerServices.getFileCache(),
+ blobCacheService,
taskManagerServices.getTaskSlotTable(),
taskManagerServices.getJobManagerTable(),
taskManagerServices.getJobLeaderService(),
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 93d432d..e8b043e 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -968,9 +968,9 @@ class TaskManager(
try {
val blobcache = new BlobCacheService(
- address,
config.getConfiguration(),
- highAvailabilityServices.createBlobStore())
+ highAvailabilityServices.createBlobStore(),
+ address)
blobCache = Option(blobcache)
libraryCacheManager = Some(
new BlobLibraryCacheManager(
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCleanupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCleanupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCleanupTest.java
index 5ec9b89..93371f7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCleanupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCleanupTest.java
@@ -91,7 +91,7 @@ public class BlobCacheCleanupTest extends TestLogger {
server = new BlobServer(config, new VoidBlobStore());
server.start();
InetSocketAddress serverAddress = new InetSocketAddress("localhost", server.getPort());
- cache = new PermanentBlobCache(serverAddress, config, new VoidBlobStore());
+ cache = new PermanentBlobCache(config, new VoidBlobStore(), serverAddress);
// upload blobs
keys.add(server.putPermanent(jobId, buf));
@@ -164,7 +164,7 @@ public class BlobCacheCleanupTest extends TestLogger {
// NOTE: use fake address - we will not connect to it here
InetSocketAddress serverAddress = new InetSocketAddress("localhost", 12345);
- try (PermanentBlobCache cache = new PermanentBlobCache(serverAddress, config, new VoidBlobStore())) {
+ try (PermanentBlobCache cache = new PermanentBlobCache(config, new VoidBlobStore(), serverAddress)) {
// register once
cache.registerJob(jobId);
@@ -231,7 +231,7 @@ public class BlobCacheCleanupTest extends TestLogger {
server = new BlobServer(config, new VoidBlobStore());
server.start();
InetSocketAddress serverAddress = new InetSocketAddress("localhost", server.getPort());
- cache = new PermanentBlobCache(serverAddress, config, new VoidBlobStore());
+ cache = new PermanentBlobCache(config, new VoidBlobStore(), serverAddress);
// upload blobs
keys.add(server.putPermanent(jobId, buf));
@@ -339,8 +339,8 @@ public class BlobCacheCleanupTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
final BlobCacheService cache = new BlobCacheService(
- new InetSocketAddress("localhost", server.getPort()), config,
- new VoidBlobStore())) {
+ config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
ConcurrentMap<Tuple2<JobID, TransientBlobKey>, Long> transientBlobExpiryTimes =
cache.getTransientBlobService().getBlobExpiryTimes();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCorruptionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCorruptionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCorruptionTest.java
index a0e46ba..4d8727c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCorruptionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheCorruptionTest.java
@@ -173,8 +173,8 @@ public class BlobCacheCorruptionTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, blobStore);
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, corruptOnHAStore ? blobStore : new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, corruptOnHAStore ? blobStore : new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
index 16ba020..42f88fb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
@@ -109,8 +109,8 @@ public class BlobCacheDeleteTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -187,8 +187,8 @@ public class BlobCacheDeleteTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -241,8 +241,8 @@ public class BlobCacheDeleteTest extends TestLogger {
File directory = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -319,8 +319,8 @@ public class BlobCacheDeleteTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
index 6e05aa0..cb493d7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
@@ -130,8 +130,8 @@ public class BlobCacheGetTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -221,8 +221,8 @@ public class BlobCacheGetTest extends TestLogger {
File tempFileDir = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -323,8 +323,8 @@ public class BlobCacheGetTest extends TestLogger {
File jobStoreDir = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -391,8 +391,8 @@ public class BlobCacheGetTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -457,8 +457,8 @@ public class BlobCacheGetTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -552,8 +552,8 @@ public class BlobCacheGetTest extends TestLogger {
try (
final BlobServer server = new BlobServer(config, blobStoreServer);
- final BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, cacheAccessesHAStore ? blobStoreServer : blobStoreCache)) {
+ final BlobCacheService cache = new BlobCacheService(config, cacheAccessesHAStore ? blobStoreServer : blobStoreCache, new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
index 264c7cb..c9bd442 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
@@ -156,7 +156,7 @@ public class BlobCachePutTest extends TestLogger {
try (BlobServer server = new BlobServer(config, new VoidBlobStore());
final TransientBlobCache cache = new TransientBlobCache(
- new InetSocketAddress("localhost", server.getPort()), config)) {
+ config, new InetSocketAddress("localhost", server.getPort()))) {
server.start();
@@ -181,8 +181,8 @@ public class BlobCachePutTest extends TestLogger {
try (BlobServer server = new BlobServer(config, new VoidBlobStore());
final PermanentBlobCache cache = new PermanentBlobCache(
- new InetSocketAddress("localhost", server.getPort()), config,
- new VoidBlobStore())) {
+ config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -263,8 +263,8 @@ public class BlobCachePutTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -379,8 +379,8 @@ public class BlobCachePutTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -489,8 +489,8 @@ public class BlobCachePutTest extends TestLogger {
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -590,8 +590,8 @@ public class BlobCachePutTest extends TestLogger {
File tempFileDir = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -653,8 +653,8 @@ public class BlobCachePutTest extends TestLogger {
File tempFileDir = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -721,8 +721,8 @@ public class BlobCachePutTest extends TestLogger {
File jobStoreDir = null;
try (
BlobServer server = new BlobServer(config, new VoidBlobStore());
- BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, new VoidBlobStore())) {
+ BlobCacheService cache = new BlobCacheService(config, new VoidBlobStore(), new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
@@ -815,8 +815,8 @@ public class BlobCachePutTest extends TestLogger {
try (
final BlobServer server = new BlobServer(config, blobStoreServer);
- final BlobCacheService cache = new BlobCacheService(new InetSocketAddress("localhost", server.getPort()),
- config, blobStoreCache)) {
+ final BlobCacheService cache = new BlobCacheService(config, blobStoreCache, new InetSocketAddress("localhost", server.getPort())
+ )) {
server.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/22bf3861/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
index e275949..13f94d8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
@@ -105,11 +105,11 @@ public class BlobCacheRecoveryTest extends TestLogger {
BlobServer server1 = new BlobServer(config, blobStore);
// use VoidBlobStore as the HA store to force download from each server's HA store
BlobCacheService cache0 = new BlobCacheService(
- new InetSocketAddress("localhost", server0.getPort()), config,
- new VoidBlobStore());
+ config, new VoidBlobStore(), new InetSocketAddress("localhost", server0.getPort())
+ );
BlobCacheService cache1 = new BlobCacheService(
- new InetSocketAddress("localhost", server1.getPort()), config,
- new VoidBlobStore())) {
+ config, new VoidBlobStore(), new InetSocketAddress("localhost", server1.getPort())
+ )) {
server0.start();
server1.start();