You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2026/03/26 02:03:53 UTC
(flink) branch master updated: [FLINK-39240][runtime] Jobs share application-level jar when possible
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new bbd045f324d [FLINK-39240][runtime] Jobs share application-level jar when possible
bbd045f324d is described below
commit bbd045f324dce0e68349c88ec1d3fc041f060efb
Author: Yi Zhang <zh...@163.com>
AuthorDate: Wed Mar 11 11:23:31 2026 +0800
[FLINK-39240][runtime] Jobs share application-level jar when possible
---
.../5b9eed8a-5fb6-4373-98ac-3be2a71941b8 | 3 +-
.../e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 | 8 +-
.../java/org/apache/flink/client/ClientUtils.java | 4 +
.../application/PackagedProgramApplication.java | 21 +-
.../PackagedProgramApplicationEntry.java | 12 +-
.../org/apache/flink/client/program/JarInfo.java | 32 ++-
.../client/program/StreamContextEnvironment.java | 11 +
.../PackagedProgramApplicationEntryTest.java | 8 +-
.../handlers/JarRunApplicationHandler.java | 18 +-
.../runtime/blob/JobPermanentBlobService.java | 7 +-
.../flink/runtime/blob/PermanentBlobCache.java | 229 +++++++++++++-----
.../JobMasterServiceLeadershipRunnerFactory.java | 4 +-
.../librarycache/BlobLibraryCacheManager.java | 27 ++-
.../librarycache/LibraryCacheManager.java | 4 +-
.../DefaultSlotPoolServiceSchedulerFactory.java | 4 +-
.../apache/flink/runtime/jobmaster/JobMaster.java | 9 +
.../jobmaster/SlotPoolServiceSchedulerFactory.java | 2 +
.../slotpool/DeclarativeSlotPoolBridge.java | 3 +
.../DeclarativeSlotPoolBridgeServiceFactory.java | 3 +
.../slotpool/DeclarativeSlotPoolService.java | 8 +-
.../DeclarativeSlotPoolServiceFactory.java | 3 +
.../jobmaster/slotpool/SlotPoolServiceFactory.java | 2 +
.../slotmanager/DefaultSlotStatusSyncer.java | 4 +
.../slotmanager/FineGrainedSlotManager.java | 12 +
.../slotmanager/SlotStatusSyncer.java | 3 +
.../flink/runtime/slots/ResourceRequirements.java | 32 ++-
.../flink/runtime/taskexecutor/TaskExecutor.java | 31 ++-
.../runtime/taskexecutor/TaskExecutorGateway.java | 3 +
.../TaskExecutorGatewayDecoratorBase.java | 3 +
.../taskexecutor/slot/SlotAllocationSnapshot.java | 14 +-
.../flink/streaming/api/graph/StreamGraph.java | 12 +
.../flink/runtime/blob/BlobCacheCleanupTest.java | 41 ++--
.../runtime/blob/NoOpJobPermanentBlobService.java | 5 +-
.../flink/runtime/blob/TestingBlobHelpers.java | 19 +-
.../flink/runtime/dispatcher/DispatcherTest.java | 11 +-
.../librarycache/BlobLibraryCacheManagerTest.java | 257 +++++++++++++++++++--
.../BlobLibraryCacheRecoveryITCase.java | 150 +++++++++++-
.../ContextClassLoaderLibraryCacheManager.java | 3 +-
.../librarycache/TestingLibraryCacheManager.java | 19 +-
.../flink/runtime/jobgraph/JobGraphTestUtils.java | 32 ++-
.../flink/runtime/jobmaster/JobMasterTest.java | 3 +
.../apache/flink/runtime/jobmaster/TestUtils.java | 15 +-
.../AbstractDeclarativeSlotPoolBridgeTest.java | 3 +
.../slotpool/DeclarativeSlotPoolBridgeBuilder.java | 4 +
...tiveSlotPoolBridgePreferredAllocationsTest.java | 2 +
.../slotpool/DeclarativeSlotPoolServiceTest.java | 3 +
.../slotpool/TestingSlotPoolServiceBuilder.java | 2 +
.../AbstractFineGrainedSlotManagerITCase.java | 43 ++--
.../slotmanager/DefaultSlotStatusSyncerTest.java | 12 +-
.../slotmanager/FineGrainedSlotManagerTest.java | 37 +--
...cutorExecutionDeploymentReconciliationTest.java | 3 +
.../TaskExecutorPartitionLifecycleTest.java | 3 +
.../taskexecutor/TaskExecutorRecoveryTest.java | 3 +
.../taskexecutor/TaskExecutorSlotLifetimeTest.java | 3 +
.../runtime/taskexecutor/TaskExecutorTest.java | 19 +-
.../taskexecutor/TestingTaskExecutorGateway.java | 19 +-
.../TestingTaskExecutorGatewayBuilder.java | 24 +-
...otAllocationSnapshotPersistenceServiceTest.java | 2 +
.../flink/streaming/api/graph/StreamGraphTest.java | 29 +++
.../service/application/ScriptExecutorITCase.java | 1 +
60 files changed, 1056 insertions(+), 247 deletions(-)
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
index ccf1f8ba490..914108019f3 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8
@@ -21,7 +21,8 @@ org.apache.flink.cep.pattern.conditions.IterativeCondition.filter(java.lang.Obje
org.apache.flink.cep.pattern.conditions.SimpleCondition.filter(java.lang.Object, org.apache.flink.cep.pattern.conditions.IterativeCondition$Context): Argument leaf type org.apache.flink.cep.pattern.conditions.IterativeCondition$Context does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.client.program.StreamContextEnvironment.execute(org.apache.flink.streaming.api.graph.StreamGraph): Argument leaf type org.apache.flink.streaming.api.graph.StreamGraph does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.client.program.StreamContextEnvironment.executeAsync(org.apache.flink.streaming.api.graph.StreamGraph): Argument leaf type org.apache.flink.streaming.api.graph.StreamGraph does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.client.program.StreamContextEnvironment.setAsContext(org.apache.flink.core.execution.PipelineExecutorServiceLoader, org.apache.flink.configuration.Configuration, java.lang.ClassLoader, boolean, boolean, org.apache.flink.api.common.ApplicationID, java.util.Collection): Argument leaf type org.apache.flink.core.execution.PipelineExecutorServiceLoader does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @P [...]
+org.apache.flink.client.program.StreamContextEnvironment.setAsContext(org.apache.flink.core.execution.PipelineExecutorServiceLoader, org.apache.flink.configuration.Configuration, java.lang.ClassLoader, boolean, boolean, org.apache.flink.api.common.ApplicationID, org.apache.flink.client.program.JarInfo, java.util.Collection): Argument leaf type org.apache.flink.client.program.JarInfo does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] o [...]
+org.apache.flink.client.program.StreamContextEnvironment.setAsContext(org.apache.flink.core.execution.PipelineExecutorServiceLoader, org.apache.flink.configuration.Configuration, java.lang.ClassLoader, boolean, boolean, org.apache.flink.api.common.ApplicationID, org.apache.flink.client.program.JarInfo, java.util.Collection): Argument leaf type org.apache.flink.core.execution.PipelineExecutorServiceLoader does not satisfy: reside outside of package 'org.apache.flink..' or reside in any pa [...]
org.apache.flink.client.program.StreamPlanEnvironment.executeAsync(org.apache.flink.streaming.api.graph.StreamGraph): Argument leaf type org.apache.flink.streaming.api.graph.StreamGraph does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.client.program.StreamPlanEnvironment.getPipeline(): Returned leaf type org.apache.flink.api.dag.Pipeline does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.configuration.ClusterOptions.getSchedulerType(org.apache.flink.configuration.Configuration): Returned leaf type org.apache.flink.configuration.JobManagerOptions$SchedulerType does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
index 7e0ecf4979f..f5af66381c0 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5
@@ -3,10 +3,10 @@ Constructor <org.apache.flink.streaming.runtime.io.StreamTaskExternallyInducedSo
Method <org.apache.flink.cep.nfa.sharedbuffer.LockableTypeSerializerSnapshot.getNestedSerializers(org.apache.flink.cep.nfa.sharedbuffer.Lockable$LockableTypeSerializer)> calls method <org.apache.flink.cep.nfa.sharedbuffer.Lockable$LockableTypeSerializer.getElementSerializer()> in (LockableTypeSerializerSnapshot.java:58)
Method <org.apache.flink.runtime.blob.BlobInputStream.read()> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobInputStream.java:127)
Method <org.apache.flink.runtime.blob.BlobInputStream.read([B, int, int)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobInputStream.java:163)
-Method <org.apache.flink.runtime.blob.BlobOutputStream.receiveAndCheckPutResponse(java.io.InputStream, java.security.MessageDigest, org.apache.flink.runtime.blob.BlobKey$BlobType)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobOutputStream.java:155)
-Method <org.apache.flink.runtime.blob.BlobUtils.checkAndDeleteCorruptedBlobs(java.nio.file.Path, org.slf4j.Logger)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobUtils.java:514)
-Method <org.apache.flink.runtime.blob.FileSystemBlobStore.get(java.lang.String, java.io.File, org.apache.flink.runtime.blob.BlobKey)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (FileSystemBlobStore.java:129)
-Method <org.apache.flink.runtime.executiongraph.Execution.finishPartitionsAndUpdateConsumers()> calls method <org.apache.flink.runtime.executiongraph.ExecutionVertex.finishPartitionsIfNeeded()> in (Execution.java:1031)
+Method <org.apache.flink.runtime.blob.BlobOutputStream.receiveAndCheckPutResponse(java.io.InputStream, java.security.MessageDigest, org.apache.flink.runtime.blob.BlobKey$BlobType)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobOutputStream.java:186)
+Method <org.apache.flink.runtime.blob.BlobUtils.checkAndDeleteCorruptedBlobs(java.nio.file.Path, org.slf4j.Logger)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (BlobUtils.java:665)
+Method <org.apache.flink.runtime.blob.FileSystemBlobStore.get(java.lang.String, java.io.File, org.apache.flink.runtime.blob.BlobKey)> calls method <org.apache.flink.runtime.blob.BlobKey.getHash()> in (FileSystemBlobStore.java:148)
+Method <org.apache.flink.runtime.executiongraph.Execution.finishPartitionsAndUpdateConsumers()> calls method <org.apache.flink.runtime.executiongraph.ExecutionVertex.finishPartitionsIfNeeded()> in (Execution.java:1202)
Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantDispatcherLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:83)
Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantJobMasterLeadership(org.apache.flink.api.common.JobID)> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:95)
Method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl.grantResourceManagerLeadership()> calls method <org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService.grantLeadership()> in (EmbeddedHaServicesWithLeadershipControl.java:109)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 4604b109d28..07b75b89d45 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.cli.ClientOptions;
+import org.apache.flink.client.program.JarInfo;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.StreamContextEnvironment;
@@ -92,6 +93,7 @@ public enum ClientUtils {
enforceSingleJobExecution,
suppressSysout,
null,
+ null,
Collections.emptyList());
}
@@ -102,6 +104,7 @@ public enum ClientUtils {
boolean enforceSingleJobExecution,
boolean suppressSysout,
@Nullable ApplicationID applicationId,
+ @Nullable JarInfo userJarInfo,
Collection<JobInfo> allRecoveredJobInfos)
throws ProgramInvocationException {
checkNotNull(executorServiceLoader);
@@ -121,6 +124,7 @@ public enum ClientUtils {
enforceSingleJobExecution,
suppressSysout,
applicationId,
+ userJarInfo,
allRecoveredJobInfos);
// For DataStream v2.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplication.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplication.java
index 6f42ea719b1..71f9df60101 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplication.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplication.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.cli.ClientOptions;
import org.apache.flink.client.deployment.application.executors.EmbeddedExecutorServiceLoader;
+import org.apache.flink.client.program.JarInfo;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramDescriptor;
import org.apache.flink.configuration.Configuration;
@@ -34,7 +35,6 @@ import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.PipelineOptionsInternal;
import org.apache.flink.core.execution.PipelineExecutorServiceLoader;
import org.apache.flink.runtime.application.AbstractApplication;
-import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.client.DuplicateJobSubmissionException;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
@@ -99,7 +99,7 @@ public class PackagedProgramApplication extends AbstractApplication {
private final boolean shutDownOnFinish;
- @Nullable private final PermanentBlobKey userJarBlobKey;
+ @Nullable private final JarInfo userJarInfo;
private transient PackagedProgram program;
@@ -139,7 +139,7 @@ public class PackagedProgramApplication extends AbstractApplication {
final boolean enforceSingleJobExecution,
final boolean submitFailedJobOnApplicationError,
final boolean shutDownOnFinish,
- final @Nullable PermanentBlobKey userJarBlobKey) {
+ final @Nullable JarInfo userJarInfo) {
this(
applicationId,
program,
@@ -150,7 +150,7 @@ public class PackagedProgramApplication extends AbstractApplication {
enforceSingleJobExecution,
submitFailedJobOnApplicationError,
shutDownOnFinish,
- userJarBlobKey);
+ userJarInfo);
}
public PackagedProgramApplication(
@@ -186,7 +186,7 @@ public class PackagedProgramApplication extends AbstractApplication {
final boolean enforceSingleJobExecution,
final boolean submitFailedJobOnApplicationError,
final boolean shutDownOnFinish,
- final @Nullable PermanentBlobKey userJarBlobKey) {
+ final @Nullable JarInfo userJarInfo) {
super(applicationId);
this.program = checkNotNull(program);
this.recoveredJobInfos = checkNotNull(recoveredJobInfos);
@@ -196,7 +196,7 @@ public class PackagedProgramApplication extends AbstractApplication {
this.enforceSingleJobExecution = enforceSingleJobExecution;
this.submitFailedJobOnApplicationError = submitFailedJobOnApplicationError;
this.shutDownOnFinish = shutDownOnFinish;
- this.userJarBlobKey = userJarBlobKey;
+ this.userJarInfo = userJarInfo;
this.programDescriptor = program.getDescriptor();
}
@@ -375,13 +375,13 @@ public class PackagedProgramApplication extends AbstractApplication {
@Override
public Optional<ApplicationStoreEntry> getApplicationStoreEntry() {
- if (userJarBlobKey == null) {
+ if (userJarInfo == null) {
return Optional.empty();
}
return Optional.of(
new PackagedProgramApplicationEntry(
configuration,
- userJarBlobKey,
+ userJarInfo,
programDescriptor.getMainClassName(),
programDescriptor.getProgramArgs(),
getApplicationId(),
@@ -393,8 +393,8 @@ public class PackagedProgramApplication extends AbstractApplication {
}
@VisibleForTesting
- PermanentBlobKey getUserJarBlobKey() {
- return userJarBlobKey;
+ JarInfo getUserJarInfo() {
+ return userJarInfo;
}
@VisibleForTesting
@@ -694,6 +694,7 @@ public class PackagedProgramApplication extends AbstractApplication {
enforceSingleJobExecution,
true /* suppress sysout */,
getApplicationId(),
+ userJarInfo,
getAllRecoveredJobInfos());
if (applicationJobIds.isEmpty()) {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java
index c435323b0c9..931bd8b9d75 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java
@@ -20,12 +20,12 @@ package org.apache.flink.client.deployment.application;
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.client.program.JarInfo;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.runtime.application.AbstractApplication;
-import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.blob.PermanentBlobService;
import org.apache.flink.runtime.jobmanager.ApplicationStoreEntry;
@@ -40,7 +40,7 @@ public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
private final Configuration configuration;
- private final PermanentBlobKey userJarBlobKey;
+ private final JarInfo userJarInfo;
private final String entryClass;
@@ -60,7 +60,7 @@ public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
public PackagedProgramApplicationEntry(
Configuration configuration,
- PermanentBlobKey userJarBlobKey,
+ JarInfo userJarInfo,
String entryClass,
String[] programArgs,
ApplicationID applicationId,
@@ -70,7 +70,7 @@ public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
boolean submitFailedJobOnApplicationError,
boolean shutDownOnFinish) {
this.configuration = configuration;
- this.userJarBlobKey = userJarBlobKey;
+ this.userJarInfo = userJarInfo;
this.entryClass = entryClass;
this.programArgs = programArgs;
this.applicationId = applicationId;
@@ -88,7 +88,7 @@ public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
Collection<JobInfo> recoveredTerminalJobInfos) {
File jarFile;
try {
- jarFile = blobService.getFile(applicationId, userJarBlobKey);
+ jarFile = blobService.getFile(applicationId, userJarInfo.getJarBlobKey());
} catch (Exception e) {
throw new RuntimeException("Failed to get user jar file from blob", e);
}
@@ -124,7 +124,7 @@ public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
enforceSingleJobExecution,
submitFailedJobOnApplicationError,
shutDownOnFinish,
- userJarBlobKey);
+ userJarInfo);
}
private List<URL> getClasspaths() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java b/flink-clients/src/main/java/org/apache/flink/client/program/JarInfo.java
similarity index 53%
copy from flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java
copy to flink-clients/src/main/java/org/apache/flink/client/program/JarInfo.java
index f0b848b700a..d868b6af8c9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/JarInfo.java
@@ -16,19 +16,31 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.execution.librarycache;
+package org.apache.flink.client.program;
-import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
-/** {@link LibraryCacheManager} implementation which returns the context class loader. */
-public enum ContextClassLoaderLibraryCacheManager implements LibraryCacheManager {
- INSTANCE;
+import java.io.Serializable;
- @Override
- public ClassLoaderLease registerClassLoaderLease(JobID jobId) {
- return TestingClassLoaderLease.newBuilder().build();
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Info about a jar. */
+public class JarInfo implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private final String jarName;
+ private final PermanentBlobKey jarBlobKey;
+
+ public JarInfo(String jarName, PermanentBlobKey jarBlobKey) {
+ this.jarName = checkNotNull(jarName);
+ this.jarBlobKey = checkNotNull(jarBlobKey);
}
- @Override
- public void shutdown() {}
+ public String getJarName() {
+ return jarName;
+ }
+
+ public PermanentBlobKey getJarBlobKey() {
+ return jarBlobKey;
+ }
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
index d45a49d52c3..aeffa48738e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
@@ -81,6 +81,8 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
@Nullable private final ApplicationID applicationId;
+ @Nullable private final JarInfo userJarInfo;
+
private final JobIdManager jobIdManager;
public StreamContextEnvironment(
@@ -119,6 +121,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
programConfigEnabled,
programConfigWildcards,
null,
+ null,
Collections.emptyList());
}
@@ -133,6 +136,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
final boolean programConfigEnabled,
final Collection<String> programConfigWildcards,
@Nullable final ApplicationID applicationId,
+ @Nullable final JarInfo userJarInfo,
final Collection<JobInfo> allRecoveredJobInfos) {
super(executorServiceLoader, configuration, userCodeClassLoader);
this.suppressSysout = suppressSysout;
@@ -142,6 +146,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
this.programConfigEnabled = programConfigEnabled;
this.programConfigWildcards = programConfigWildcards;
this.applicationId = applicationId;
+ this.userJarInfo = userJarInfo;
this.jobIdManager = createJobIdManager(configuration, allRecoveredJobInfos);
}
@@ -241,6 +246,10 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
if (applicationId != null) {
streamGraph.setApplicationId(applicationId);
}
+ if (userJarInfo != null) {
+ streamGraph.addUserJarBlobKey(userJarInfo.getJarBlobKey());
+ streamGraph.addUserJarToSkip(userJarInfo.getJarName());
+ }
jobIdManager.updateJobId(streamGraph);
final JobClient jobClient = super.executeAsync(streamGraph);
@@ -268,6 +277,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
final boolean enforceSingleJobExecution,
final boolean suppressSysout,
@Nullable final ApplicationID applicationId,
+ @Nullable final JarInfo userJarInfo,
Collection<JobInfo> allRecoveredJobInfos) {
final StreamExecutionEnvironmentFactory factory =
envInitConfig -> {
@@ -288,6 +298,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
programConfigEnabled,
programConfigWildcards,
applicationId,
+ userJarInfo,
allRecoveredJobInfos);
};
initializeContextEnvironment(factory);
diff --git a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntryTest.java b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntryTest.java
index d6e559d6ad7..169c31cee5c 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntryTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntryTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.client.cli.CliFrontendTestUtils;
+import org.apache.flink.client.program.JarInfo;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.testjar.MultiExecuteJob;
import org.apache.flink.configuration.Configuration;
@@ -103,9 +104,10 @@ class PackagedProgramApplicationEntryTest {
blobKey = blobServer.putPermanent(applicationId, is);
}
+ JarInfo userJarInfo = new JarInfo(jarFile.getName(), blobKey);
PackagedProgramApplication application =
new PackagedProgramApplication(
- applicationId, program, config, true, false, false, true, blobKey);
+ applicationId, program, config, true, false, false, true, userJarInfo);
ApplicationStoreEntry entry = application.getApplicationStoreEntry().orElse(null);
assertInstanceOf(PackagedProgramApplicationEntry.class, entry);
@@ -125,7 +127,7 @@ class PackagedProgramApplicationEntryTest {
PackagedProgramApplication packagedProgramApplication =
(PackagedProgramApplication) reconstructed;
- assertEquals(blobKey, packagedProgramApplication.getUserJarBlobKey());
+ assertEquals(blobKey, packagedProgramApplication.getUserJarInfo().getJarBlobKey());
assertThat(
packagedProgramApplication.getRecoveredJobInfos().stream()
.map(JobInfo::getJobId)
@@ -171,7 +173,7 @@ class PackagedProgramApplicationEntryTest {
private PackagedProgramApplicationEntry createEntryWithJarBlob(PermanentBlobKey blobKey) {
return new PackagedProgramApplicationEntry(
config,
- blobKey,
+ new JarInfo(jarFile.getName(), blobKey),
entryClass,
programArgs,
applicationId,
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunApplicationHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunApplicationHandler.java
index 33f684174f5..0ca5b679712 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunApplicationHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunApplicationHandler.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.client.deployment.application.PackagedProgramApplication;
import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor;
+import org.apache.flink.client.program.JarInfo;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
@@ -110,7 +111,7 @@ public class JarRunApplicationHandler
final boolean isHaEnabled =
HighAvailabilityMode.isHighAvailabilityModeActivated(configuration);
- final CompletableFuture<PermanentBlobKey> jarUploadFuture;
+ final CompletableFuture<JarInfo> jarUploadFuture;
if (isHaEnabled) {
// In HA mode, a fixed job id is required to ensure consistency across failovers.
// The job id is derived from the application id if not configured.
@@ -124,7 +125,7 @@ public class JarRunApplicationHandler
return jarUploadFuture
.thenCompose(
- blobKey -> {
+ jarInfo -> {
PackagedProgramApplication application =
new PackagedProgramApplication(
applicationId,
@@ -134,7 +135,7 @@ public class JarRunApplicationHandler
true,
false,
false,
- blobKey);
+ jarInfo);
return gateway.submitApplication(application, timeout);
})
.handle(
@@ -160,7 +161,7 @@ public class JarRunApplicationHandler
}
}
- private CompletableFuture<PermanentBlobKey> uploadJarFile(
+ private CompletableFuture<JarInfo> uploadJarFile(
final DispatcherGateway gateway,
final JarHandlerContext context,
final ApplicationID applicationId) {
@@ -177,12 +178,11 @@ public class JarRunApplicationHandler
.thenApply(
blobSocketAddress -> {
try {
- org.apache.flink.core.fs.Path jarPath =
- new org.apache.flink.core.fs.Path(
- context.getJarFile().toString());
+ Path jarPath = context.getJarFile();
PermanentBlobKey blobKey =
ClientUtils.uploadUserJarForApplication(
- jarPath,
+ new org.apache.flink.core.fs.Path(
+ jarPath.toString()),
applicationId,
() ->
new BlobClient(
@@ -191,7 +191,7 @@ public class JarRunApplicationHandler
"Uploaded user jar for application {} to blob server with blob key {}.",
applicationId,
blobKey);
- return blobKey;
+ return new JarInfo(jarPath.getFileName().toString(), blobKey);
} catch (Exception e) {
throw new CompletionException(
new RestHandlerException(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/JobPermanentBlobService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/JobPermanentBlobService.java
index 3c937d007e8..0cf0f42f044 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/JobPermanentBlobService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/JobPermanentBlobService.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.blob;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
/**
@@ -28,13 +29,15 @@ public interface JobPermanentBlobService extends PermanentBlobService {
* Register the given job.
*
* @param jobId job id identifying the job to register
+ * @param applicationId application id identifying the application the job belongs to
*/
- void registerJob(JobID jobId);
+ void registerJob(JobID jobId, ApplicationID applicationId);
/**
* Release the given job. This makes the blobs stored for this job up for cleanup.
*
* @param jobId job id identifying the job to register
+ * @param applicationId application id identifying the application the job belongs to
*/
- void releaseJob(JobID jobId);
+ void releaseJob(JobID jobId, ApplicationID applicationId);
}
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 2a24dfab62b..c279ded5e4e 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
@@ -43,6 +43,8 @@ import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
+import java.util.function.Consumer;
+import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -59,14 +61,14 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public class PermanentBlobCache extends AbstractBlobCache implements JobPermanentBlobService {
- /** Job reference counters with a time-to-live (TTL). */
+ /** Object reference counters with a time-to-live (TTL). */
@VisibleForTesting
static class RefCount {
- /** Number of references to a job. */
+ /** Number of references to an object. */
public int references = 0;
/**
- * Timestamp in milliseconds when any job data should be cleaned up (no cleanup for
+ * Timestamp in milliseconds when any object data should be cleaned up (no cleanup for
* non-positive values).
*/
public long keepUntil = -1;
@@ -77,6 +79,12 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
/** Map to store the number of references to a specific job. */
private final Map<JobID, RefCount> jobRefCounters = new HashMap<>();
+ /** Map to store the number of references to a specific application. */
+ private final Map<ApplicationID, RefCount> applicationRefCounters = new HashMap<>();
+
+ /** Lock object for protecting both jobRefCounters and applicationRefCounters. */
+ private final Object refCountersLock = new Object();
+
/** Time interval (ms) to run the cleanup task; also used as the default TTL. */
private final long cleanupInterval;
@@ -161,6 +169,7 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
this.blobCacheSizeTracker = blobCacheSizeTracker;
registerDetectedJobs();
+ registerDetectedApplications();
}
private void registerDetectedJobs() throws IOException {
@@ -177,7 +186,7 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
private void registerJobWithExpiry(JobID jobId, long expiryTimeout) {
checkNotNull(jobId);
- synchronized (jobRefCounters) {
+ synchronized (refCountersLock) {
final RefCount refCount =
jobRefCounters.computeIfAbsent(jobId, ignored -> new RefCount());
@@ -185,20 +194,43 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
}
}
+ private void registerDetectedApplications() throws IOException {
+ if (storageDir.deref().exists()) {
+ final Collection<ApplicationID> applicationIds =
+ BlobUtils.listExistingApplications(storageDir.deref().toPath());
+
+ final long expiryTimeout = System.currentTimeMillis() + cleanupInterval;
+ for (ApplicationID applicationId : applicationIds) {
+ registerApplicationWithExpiry(applicationId, expiryTimeout);
+ }
+ }
+ }
+
+ private void registerApplicationWithExpiry(ApplicationID applicationId, long expiryTimeout) {
+ checkNotNull(applicationId);
+ synchronized (refCountersLock) {
+ final RefCount refCount =
+ applicationRefCounters.computeIfAbsent(
+ applicationId, ignored -> new RefCount());
+
+ refCount.keepUntil = expiryTimeout;
+ }
+ }
+
/**
* Registers use of job-related BLOBs.
*
* <p>Using any other method to access BLOBs, e.g. {@link #getFile}, is only valid within calls
- * to <tt>registerJob(JobID)</tt> and {@link #releaseJob(JobID)}.
+ * to <tt>registerJob(JobID)</tt> and {@link #releaseJob(JobID, ApplicationID)}.
*
* @param jobId ID of the job this blob belongs to
- * @see #releaseJob(JobID)
+ * @see #releaseJob(JobID, ApplicationID)
*/
@Override
- public void registerJob(JobID jobId) {
+ public void registerJob(JobID jobId, ApplicationID applicationId) {
checkNotNull(jobId);
- synchronized (jobRefCounters) {
+ synchronized (refCountersLock) {
RefCount ref = jobRefCounters.get(jobId);
if (ref == null) {
ref = new RefCount();
@@ -208,6 +240,17 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
ref.keepUntil = -1;
}
++ref.references;
+
+ // also register under application because the job may use application-level blobs
+ RefCount applicationRef = applicationRefCounters.get(applicationId);
+ if (applicationRef == null) {
+ applicationRef = new RefCount();
+ applicationRefCounters.put(applicationId, applicationRef);
+ } else {
+ // reset cleanup timeout
+ applicationRef.keepUntil = -1;
+ }
+ ++applicationRef.references;
}
}
@@ -215,19 +258,21 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
* Unregisters use of job-related BLOBs and allow them to be released.
*
* @param jobId ID of the job this blob belongs to
- * @see #registerJob(JobID)
+ * @see #registerJob(JobID, ApplicationID)
*/
@Override
- public void releaseJob(JobID jobId) {
+ public void releaseJob(JobID jobId, ApplicationID applicationId) {
checkNotNull(jobId);
- synchronized (jobRefCounters) {
+ synchronized (refCountersLock) {
+ String warning =
+ "improper use of releaseJob() without a matching number of registerJob() calls for jobId "
+ + jobId;
+
RefCount ref = jobRefCounters.get(jobId);
if (ref == null || ref.references == 0) {
- log.warn(
- "improper use of releaseJob() without a matching number of registerJob() calls for jobId "
- + jobId);
+ log.warn(warning);
return;
}
@@ -235,13 +280,29 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
if (ref.references == 0) {
ref.keepUntil = System.currentTimeMillis() + cleanupInterval;
}
+
+ // make sure application related data can be cleaned up
+ RefCount applicationRef = applicationRefCounters.get(applicationId);
+
+ if (applicationRef == null || applicationRef.references == 0) {
+ log.warn(
+ "Cache reference count mismatch for application {}, which indicates {}.",
+ applicationId,
+ warning);
+ return;
+ }
+
+ --applicationRef.references;
+ if (applicationRef.references == 0) {
+ applicationRef.keepUntil = System.currentTimeMillis() + cleanupInterval;
+ }
}
}
public int getNumberOfReferenceHolders(JobID jobId) {
checkNotNull(jobId);
- synchronized (jobRefCounters) {
+ synchronized (refCountersLock) {
RefCount ref = jobRefCounters.get(jobId);
if (ref == null) {
return 0;
@@ -439,6 +500,21 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
return BlobUtils.getStorageLocation(storageDir.deref(), jobId, key);
}
+ /**
+ * Returns a file handle to the file associated with the given blob key on the blob server.
+ *
+ * @param applicationId ID of the application this blob belongs to (or <tt>null</tt> if
+ * job-unrelated)
+ * @param key identifying the file
+ * @return file handle to the file
+ * @throws IOException if creating the directory fails
+ */
+ @VisibleForTesting
+ public File getStorageLocation(ApplicationID applicationId, BlobKey key) throws IOException {
+ checkNotNull(applicationId);
+ return BlobUtils.getStorageLocation(storageDir.deref(), applicationId, key);
+ }
+
/**
* Returns the job reference counters - for testing purposes only!
*
@@ -457,53 +533,88 @@ public class PermanentBlobCache extends AbstractBlobCache implements JobPermanen
/** Cleans up BLOBs which are not referenced anymore. */
@Override
public void run() {
- synchronized (jobRefCounters) {
- Iterator<Map.Entry<JobID, RefCount>> entryIter =
- jobRefCounters.entrySet().iterator();
- final long currentTimeMillis = System.currentTimeMillis();
+ synchronized (refCountersLock) {
+ // Clean up job blobs
+ cleanupRefCounters(
+ jobRefCounters.entrySet().iterator(),
+ jobId ->
+ new File(
+ BlobUtils.getStorageLocationPath(
+ storageDir.deref().getAbsolutePath(), jobId)),
+ blobCacheSizeTracker::untrackAll,
+ "job");
+
+ // Clean up application blobs
+ /*
+ * NOTE: We do not need to call blobCacheSizeTracker.untrackAll here because
+ * blobCacheSizeTracker currently only tracks and manages a subset of job-level
+ * blobs (e.g., ShuffleDescriptors, see FLINK-23354). Application-level blobs are
+ * not tracked by the size tracker.
+ */
+ cleanupRefCounters(
+ applicationRefCounters.entrySet().iterator(),
+ applicationId ->
+ new File(
+ BlobUtils.getStorageLocationPath(
+ storageDir.deref().getAbsolutePath(),
+ applicationId)),
+ applicationId -> {},
+ "application");
+ }
+ }
+ }
- while (entryIter.hasNext()) {
- Map.Entry<JobID, RefCount> entry = entryIter.next();
- RefCount ref = entry.getValue();
+ /**
+ * Generic method to clean up BLOBs for both job and application ref-counters.
+ *
+ * @param entryIter iterator over the ref-count entries to process
+ * @param getStorageLocation function to get the storage location for an entity ID
+ * @param beforeDelete action to perform before deleting the directory (e.g., untracking)
+ * @param entityType type description for logging (e.g., "job", "application")
+ * @param <ID> type of the entity ID (JobID or ApplicationID)
+ */
+ private <ID> void cleanupRefCounters(
+ Iterator<Map.Entry<ID, RefCount>> entryIter,
+ Function<ID, File> getStorageLocation,
+ Consumer<ID> beforeDelete,
+ String entityType) {
+
+ final long currentTimeMillis = System.currentTimeMillis();
+
+ while (entryIter.hasNext()) {
+ Map.Entry<ID, RefCount> entry = entryIter.next();
+ RefCount ref = entry.getValue();
+
+ if (ref.references <= 0 && ref.keepUntil > 0 && currentTimeMillis >= ref.keepUntil) {
+ ID id = entry.getKey();
+ final File localFile = getStorageLocation.apply(id);
+
+ /*
+ * NOTE: normally it is not required to acquire the write lock to delete the storage
+ * directory since there should be no one accessing it with the ref counter
+ * being 0 - acquire it just in case, to always be on the safe side
+ */
+ readWriteLock.writeLock().lock();
- if (ref.references <= 0
- && ref.keepUntil > 0
- && currentTimeMillis >= ref.keepUntil) {
- JobID jobId = entry.getKey();
+ boolean success = false;
+ try {
+ beforeDelete.accept(id);
+ FileUtils.deleteDirectory(localFile);
+ success = true;
+ } catch (Throwable t) {
+ log.warn(
+ "Failed to locally delete {} directory {}",
+ entityType,
+ localFile.getAbsolutePath(),
+ t);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
- final File localFile =
- new File(
- BlobUtils.getStorageLocationPath(
- storageDir.deref().getAbsolutePath(), jobId));
-
- /*
- * NOTE: normally it is not required to acquire the write lock to delete the job's
- * storage directory since there should be no one accessing it with the ref
- * counter being 0 - acquire it just in case, to always be on the safe side
- */
- readWriteLock.writeLock().lock();
-
- boolean success = false;
- try {
- blobCacheSizeTracker.untrackAll(jobId);
- FileUtils.deleteDirectory(localFile);
- success = true;
- } catch (Throwable t) {
- log.warn(
- "Failed to locally delete job directory "
- + localFile.getAbsolutePath(),
- t);
- } finally {
- readWriteLock.writeLock().unlock();
- }
-
- // let's only remove this directory from cleanup if the cleanup was
- // successful
- // (does not need the write lock)
- if (success) {
- entryIter.remove();
- }
- }
+ // let's only remove this directory from cleanup if the cleanup was successful
+ // (does not need the write lock)
+ if (success) {
+ entryIter.remove();
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
index ed564eed5d8..4b6ba4a44f9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
@@ -92,7 +92,9 @@ public enum JobMasterServiceLeadershipRunnerFactory implements JobManagerRunnerF
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
jobManagerServices
.getLibraryCacheManager()
- .registerClassLoaderLease(executionPlan.getJobID());
+ .registerClassLoaderLease(
+ executionPlan.getJobID(),
+ executionPlan.getApplicationId().orElseThrow());
final ClassLoader userCodeClassLoader =
classLoaderLease
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
index eea9650fbba..c031e049aff 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.blob.PermanentBlobService;
@@ -92,10 +93,10 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
}
@Override
- public ClassLoaderLease registerClassLoaderLease(JobID jobId) {
+ public ClassLoaderLease registerClassLoaderLease(JobID jobId, ApplicationID applicationId) {
synchronized (lockObject) {
return cacheEntries
- .computeIfAbsent(jobId, jobID -> new LibraryCacheEntry(jobId))
+ .computeIfAbsent(jobId, jobID -> new LibraryCacheEntry(jobId, applicationId))
.obtainLease();
}
}
@@ -210,6 +211,8 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
private final class LibraryCacheEntry {
private final JobID jobId;
+ private final ApplicationID applicationId;
+
@GuardedBy("lockObject")
private int referenceCount;
@@ -220,8 +223,9 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
@GuardedBy("lockObject")
private boolean isReleased;
- private LibraryCacheEntry(JobID jobId) {
+ private LibraryCacheEntry(JobID jobId, ApplicationID applicationId) {
this.jobId = jobId;
+ this.applicationId = applicationId;
referenceCount = 0;
this.resolvedClassLoader = null;
this.isReleased = false;
@@ -241,7 +245,7 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
systemClassLoader
? ClassLoader.getSystemClassLoader()
: createUserCodeClassLoader(
- jobId, libraries, classPaths),
+ jobId, applicationId, libraries, classPaths),
libraries,
classPaths,
systemClassLoader);
@@ -256,6 +260,7 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
@GuardedBy("lockObject")
private URLClassLoader createUserCodeClassLoader(
JobID jobId,
+ ApplicationID applicationId,
Collection<PermanentBlobKey> requiredJarFiles,
Collection<URL> requiredClasspaths)
throws IOException {
@@ -265,7 +270,19 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
int count = 0;
// add URLs to locally cached JAR files
for (PermanentBlobKey key : requiredJarFiles) {
- libraryURLs[count] = blobService.getFile(jobId, key).toURI().toURL();
+ try {
+ // first try job specific libraries
+ libraryURLs[count] = blobService.getFile(jobId, key).toURI().toURL();
+ } catch (Exception e) {
+ // then try application specific libraries
+ LOG.info(
+ "Falling back to application blob {} with application id {} for job {}.",
+ key,
+ applicationId,
+ jobId);
+ libraryURLs[count] =
+ blobService.getFile(applicationId, key).toURI().toURL();
+ }
++count;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
index d38301ea32a..e7f9e82f1a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.util.UserCodeClassLoader;
@@ -43,9 +44,10 @@ public interface LibraryCacheManager {
* job will be valid as long as there exists a valid lease for this job.
*
* @param jobId jobId for which to register a new class loader lease
+ * @param applicationId applicationId to which the job belongs
* @return a new class loader lease for the given job
*/
- ClassLoaderLease registerClassLoaderLease(JobID jobId);
+ ClassLoaderLease registerClassLoaderLease(JobID jobId, ApplicationID applicationId);
/**
* Shuts the library cache manager down. Thereby it will close all open {@link ClassLoaderLease}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
index d95190351ed..a49942b52c1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.jobmaster;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
@@ -93,10 +94,11 @@ public final class DefaultSlotPoolServiceSchedulerFactory
@Override
public SlotPoolService createSlotPoolService(
JobID jid,
+ ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
return slotPoolServiceFactory.createSlotPoolService(
- jid, declarativeSlotPoolFactory, componentMainThreadExecutor);
+ jid, applicationId, declarativeSlotPoolFactory, componentMainThreadExecutor);
}
@Override
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 00b88453533..6f67ecc43c3 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
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.jobmaster;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.functions.AggregateFunction;
@@ -321,6 +322,13 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
};
final String jobName = executionPlan.getName();
final JobID jid = executionPlan.getJobID();
+ final ApplicationID applicationId =
+ executionPlan
+ .getApplicationId()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "The job does not have an associated application."));
this.executionPlan = checkNotNull(executionPlan);
@@ -362,6 +370,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
checkNotNull(slotPoolServiceSchedulerFactory)
.createSlotPoolService(
jid,
+ applicationId,
createDeclarativeSlotPoolFactory(
jobMasterConfiguration.getConfiguration()),
getMainThreadExecutor());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
index 96cdc841435..d85f9b96552 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
@@ -58,6 +59,7 @@ public interface SlotPoolServiceSchedulerFactory {
*/
SlotPoolService createSlotPoolService(
JobID jid,
+ ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
index 2fdd45b5018..3cfeac17226 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -102,6 +103,7 @@ public class DeclarativeSlotPoolBridge extends DeclarativeSlotPoolService implem
public DeclarativeSlotPoolBridge(
JobID jobId,
+ ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
Clock clock,
Duration rpcTimeout,
@@ -113,6 +115,7 @@ public class DeclarativeSlotPoolBridge extends DeclarativeSlotPoolService implem
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
super(
jobId,
+ applicationId,
declarativeSlotPoolFactory,
clock,
idleSlotTimeout,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeServiceFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeServiceFactory.java
index 33391da3e15..955a9906373 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeServiceFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeServiceFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.util.clock.Clock;
@@ -53,10 +54,12 @@ public class DeclarativeSlotPoolBridgeServiceFactory extends AbstractSlotPoolSer
@Override
public SlotPoolService createSlotPoolService(
@Nonnull JobID jobId,
+ @Nonnull ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
return new DeclarativeSlotPoolBridge(
jobId,
+ applicationId,
declarativeSlotPoolFactory,
clock,
rpcTimeout,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolService.java
index 7ced0b70189..5869a4a2f3c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolService.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -57,6 +58,8 @@ public class DeclarativeSlotPoolService implements SlotPoolService {
private final JobID jobId;
+ private final ApplicationID applicationId;
+
private final Duration rpcTimeout;
private final DeclarativeSlotPool declarativeSlotPool;
@@ -80,6 +83,7 @@ public class DeclarativeSlotPoolService implements SlotPoolService {
public DeclarativeSlotPoolService(
JobID jobId,
+ ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
Clock clock,
Duration idleSlotTimeout,
@@ -87,6 +91,7 @@ public class DeclarativeSlotPoolService implements SlotPoolService {
Duration slotRequestMaxInterval,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
this.jobId = jobId;
+ this.applicationId = applicationId;
this.clock = clock;
this.rpcTimeout = rpcTimeout;
this.registeredTaskManagers = new HashSet<>();
@@ -307,7 +312,8 @@ public class DeclarativeSlotPoolService implements SlotPoolService {
assertHasBeenStarted();
resourceRequirementServiceConnectionManager.declareResourceRequirements(
- ResourceRequirements.create(jobId, jobManagerAddress, resourceRequirements));
+ ResourceRequirements.create(
+ jobId, applicationId, jobManagerAddress, resourceRequirements));
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceFactory.java
index be26dd496e8..fd8241eeae3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.util.clock.Clock;
@@ -49,10 +50,12 @@ public class DeclarativeSlotPoolServiceFactory implements SlotPoolServiceFactory
@Override
public SlotPoolService createSlotPoolService(
@Nonnull JobID jobId,
+ @Nonnull ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
return new DeclarativeSlotPoolService(
jobId,
+ applicationId,
declarativeSlotPoolFactory,
clock,
idleSlotTimeout,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolServiceFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolServiceFactory.java
index f987f28a67f..0d25f1a41a7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolServiceFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolServiceFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -29,6 +30,7 @@ public interface SlotPoolServiceFactory {
@Nonnull
SlotPoolService createSlotPoolService(
@Nonnull JobID jobId,
+ @Nonnull ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
index e957eb3b644..15354c89ed6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -96,10 +97,12 @@ public class DefaultSlotStatusSyncer implements SlotStatusSyncer {
public CompletableFuture<Void> allocateSlot(
InstanceID instanceId,
JobID jobId,
+ ApplicationID applicationId,
String targetAddress,
ResourceProfile resourceProfile) {
Preconditions.checkNotNull(instanceId);
Preconditions.checkNotNull(jobId);
+ Preconditions.checkNotNull(applicationId);
Preconditions.checkNotNull(targetAddress);
Preconditions.checkNotNull(resourceProfile);
checkStarted();
@@ -132,6 +135,7 @@ public class DefaultSlotStatusSyncer implements SlotStatusSyncer {
gateway.requestSlot(
SlotID.getDynamicSlotID(resourceId),
jobId,
+ applicationId,
allocationId,
resourceProfile,
targetAddress,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
index 2a03539a22e..676838496dc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.resources.CPUResource;
import org.apache.flink.configuration.MemorySize;
@@ -96,6 +97,8 @@ public class FineGrainedSlotManager implements SlotManager {
private final Map<JobID, String> jobMasterTargetAddresses = new HashMap<>();
+ private final Map<JobID, ApplicationID> jobIdsToApplicationIds = new HashMap<>();
+
private final CPUResource maxTotalCpu;
private final MemorySize maxTotalMem;
@@ -343,6 +346,7 @@ public class FineGrainedSlotManager implements SlotManager {
if (resourceRequirements.getResourceRequirements().isEmpty()) {
LOG.info("Clearing resource requirements of job {}", resourceRequirements.getJobId());
jobMasterTargetAddresses.remove(resourceRequirements.getJobId());
+ jobIdsToApplicationIds.remove(resourceRequirements.getJobId());
if (resourceAllocator.isSupported()) {
taskManagerTracker.clearPendingAllocationsOfJob(resourceRequirements.getJobId());
}
@@ -353,6 +357,8 @@ public class FineGrainedSlotManager implements SlotManager {
resourceRequirements.getResourceRequirements());
jobMasterTargetAddresses.put(
resourceRequirements.getJobId(), resourceRequirements.getTargetAddress());
+ jobIdsToApplicationIds.put(
+ resourceRequirements.getJobId(), resourceRequirements.getApplicationId());
}
resourceTracker.notifyResourceRequirements(
@@ -751,6 +757,7 @@ public class FineGrainedSlotManager implements SlotManager {
slotStatusSyncer.allocateSlot(
instanceID,
jobID,
+ jobIdsToApplicationIds.get(jobID),
jobMasterTargetAddresses.get(jobID),
slotEntry.getKey()));
}
@@ -923,6 +930,11 @@ public class FineGrainedSlotManager implements SlotManager {
.orElse(0L);
}
+ @VisibleForTesting
+ public Optional<ApplicationID> getApplicationId(JobID jobId) {
+ return Optional.ofNullable(jobIdsToApplicationIds.get(jobId));
+ }
+
// ---------------------------------------------------------------------------------------------
// Internal utility methods
// ---------------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotStatusSyncer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotStatusSyncer.java
index ca25e3402b6..c2906f52674 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotStatusSyncer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotStatusSyncer.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -56,6 +57,7 @@ public interface SlotStatusSyncer {
*
* @param instanceId of the task manager
* @param jobId of the slot
+ * @param applicationId of the application
* @param targetAddress of the job
* @param resourceProfile of the slot
* @return a {@link CompletableFuture} of the slot allocation, which will be completed
@@ -64,6 +66,7 @@ public interface SlotStatusSyncer {
CompletableFuture<Void> allocateSlot(
InstanceID instanceId,
JobID jobId,
+ ApplicationID applicationId,
String targetAddress,
ResourceProfile resourceProfile);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/slots/ResourceRequirements.java b/flink-runtime/src/main/java/org/apache/flink/runtime/slots/ResourceRequirements.java
index 0d8235ece2f..7112de3ed33 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/slots/ResourceRequirements.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/slots/ResourceRequirements.java
@@ -17,6 +17,8 @@
package org.apache.flink.runtime.slots;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.util.Preconditions;
@@ -35,15 +37,19 @@ public class ResourceRequirements implements Serializable {
private final JobID jobId;
+ private final ApplicationID applicationId;
+
private final String targetAddress;
private final Collection<ResourceRequirement> resourceRequirements;
private ResourceRequirements(
JobID jobId,
+ ApplicationID applicationId,
String targetAddress,
Collection<ResourceRequirement> resourceRequirements) {
this.jobId = Preconditions.checkNotNull(jobId);
+ this.applicationId = Preconditions.checkNotNull(applicationId);
this.targetAddress = Preconditions.checkNotNull(targetAddress);
this.resourceRequirements = Preconditions.checkNotNull(resourceRequirements);
}
@@ -52,6 +58,10 @@ public class ResourceRequirements implements Serializable {
return jobId;
}
+ public ApplicationID getApplicationId() {
+ return applicationId;
+ }
+
public String getTargetAddress() {
return targetAddress;
}
@@ -60,15 +70,26 @@ public class ResourceRequirements implements Serializable {
return resourceRequirements;
}
+ @VisibleForTesting
+ public static ResourceRequirements create(
+ JobID jobId,
+ String targetAddress,
+ Collection<ResourceRequirement> resourceRequirements) {
+ return create(jobId, new ApplicationID(), targetAddress, resourceRequirements);
+ }
+
public static ResourceRequirements create(
JobID jobId,
+ ApplicationID applicationId,
String targetAddress,
Collection<ResourceRequirement> resourceRequirements) {
- return new ResourceRequirements(jobId, targetAddress, resourceRequirements);
+ return new ResourceRequirements(jobId, applicationId, targetAddress, resourceRequirements);
}
- public static ResourceRequirements empty(JobID jobId, String targetAddress) {
- return new ResourceRequirements(jobId, targetAddress, Collections.emptyList());
+ public static ResourceRequirements empty(
+ JobID jobId, ApplicationID applicationId, String targetAddress) {
+ return new ResourceRequirements(
+ jobId, applicationId, targetAddress, Collections.emptyList());
}
@Override
@@ -81,13 +102,14 @@ public class ResourceRequirements implements Serializable {
}
ResourceRequirements that = (ResourceRequirements) o;
return Objects.equals(jobId, that.jobId)
+ && Objects.equals(applicationId, that.applicationId)
&& Objects.equals(targetAddress, that.targetAddress)
&& Objects.equals(resourceRequirements, that.resourceRequirements);
}
@Override
public int hashCode() {
- return Objects.hash(jobId, targetAddress, resourceRequirements);
+ return Objects.hash(jobId, applicationId, targetAddress, resourceRequirements);
}
@Override
@@ -95,6 +117,8 @@ public class ResourceRequirements implements Serializable {
return "ResourceRequirements{"
+ "jobId="
+ jobId
+ + ", applicationId="
+ + applicationId
+ ", targetAddress='"
+ targetAddress
+ '\''
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 88f24d7df0d..e5b9fc9b79d 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
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.BatchExecutionOptions;
import org.apache.flink.configuration.ClusterOptions;
@@ -1188,6 +1189,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
public CompletableFuture<Acknowledge> requestSlot(
final SlotID slotId,
final JobID jobId,
+ final ApplicationID applicationId,
final AllocationID allocationId,
final ResourceProfile resourceProfile,
final String targetAddress,
@@ -1214,12 +1216,22 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
tryPersistAllocationSnapshot(
new SlotAllocationSnapshot(
- slotId, jobId, targetAddress, allocationId, resourceProfile));
+ slotId,
+ jobId,
+ applicationId,
+ targetAddress,
+ allocationId,
+ resourceProfile));
try {
final boolean isConnected =
allocateSlotForJob(
- jobId, slotId, allocationId, resourceProfile, targetAddress);
+ jobId,
+ applicationId,
+ slotId,
+ allocationId,
+ resourceProfile,
+ targetAddress);
if (isConnected) {
offerSlotsToJobManager(jobId);
@@ -1235,6 +1247,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
private boolean allocateSlotForJob(
JobID jobId,
+ ApplicationID applicationId,
SlotID slotId,
AllocationID allocationId,
ResourceProfile resourceProfile,
@@ -1247,7 +1260,10 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
try {
job =
jobTable.getOrCreateJob(
- jobId, () -> registerNewJobAndCreateServices(jobId, targetAddress));
+ jobId,
+ () ->
+ registerNewJobAndCreateServices(
+ jobId, applicationId, targetAddress));
} catch (Exception e) {
// free the allocated slot
try {
@@ -1273,15 +1289,15 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
private TaskExecutorJobServices registerNewJobAndCreateServices(
- JobID jobId, String targetAddress) throws Exception {
+ JobID jobId, ApplicationID applicationId, String targetAddress) throws Exception {
jobLeaderService.addJob(jobId, targetAddress);
final JobPermanentBlobService permanentBlobService =
taskExecutorBlobService.getPermanentBlobService();
- permanentBlobService.registerJob(jobId);
+ permanentBlobService.registerJob(jobId, applicationId);
return TaskExecutorJobServices.create(
- libraryCacheManager.registerClassLoaderLease(jobId),
- () -> permanentBlobService.releaseJob(jobId));
+ libraryCacheManager.registerClassLoaderLease(jobId, applicationId),
+ () -> permanentBlobService.releaseJob(jobId, applicationId));
}
private void allocateSlot(
@@ -2341,6 +2357,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
try {
allocateSlotForJob(
slotAllocationSnapshot.getJobId(),
+ slotAllocationSnapshot.getApplicationId(),
slotAllocationSnapshot.getSlotID(),
slotAllocationSnapshot.getAllocationId(),
slotAllocationSnapshot.getResourceProfile(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
index 5bde205afc3..29607c2c2f8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.TransientBlobKey;
@@ -61,6 +62,7 @@ public interface TaskExecutorGateway
*
* @param slotId slot id for the request
* @param jobId for which to request a slot
+ * @param applicationId to which the job belongs
* @param allocationId id for the request
* @param resourceProfile of requested slot, used only for dynamic slot allocation and will be
* ignored otherwise
@@ -72,6 +74,7 @@ public interface TaskExecutorGateway
CompletableFuture<Acknowledge> requestSlot(
SlotID slotId,
JobID jobId,
+ ApplicationID applicationId,
AllocationID allocationId,
ResourceProfile resourceProfile,
String targetAddress,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
index 78217283049..5d33ddd65a0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -77,6 +78,7 @@ public class TaskExecutorGatewayDecoratorBase implements TaskExecutorGateway {
public CompletableFuture<Acknowledge> requestSlot(
SlotID slotId,
JobID jobId,
+ ApplicationID applicationId,
AllocationID allocationId,
ResourceProfile resourceProfile,
String targetAddress,
@@ -85,6 +87,7 @@ public class TaskExecutorGatewayDecoratorBase implements TaskExecutorGateway {
return originalGateway.requestSlot(
slotId,
jobId,
+ applicationId,
allocationId,
resourceProfile,
targetAddress,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/SlotAllocationSnapshot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/SlotAllocationSnapshot.java
index d58deb96989..b8ab093d6ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/SlotAllocationSnapshot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/SlotAllocationSnapshot.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor.slot;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -32,6 +33,7 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
private final SlotID slotID;
private final JobID jobId;
+ private final ApplicationID applicationId;
private final String jobTargetAddress;
private final AllocationID allocationId;
private final ResourceProfile resourceProfile;
@@ -39,11 +41,13 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
public SlotAllocationSnapshot(
SlotID slotID,
JobID jobId,
+ ApplicationID applicationId,
String jobTargetAddress,
AllocationID allocationId,
ResourceProfile resourceProfile) {
this.slotID = slotID;
this.jobId = jobId;
+ this.applicationId = applicationId;
this.jobTargetAddress = jobTargetAddress;
this.allocationId = allocationId;
this.resourceProfile = resourceProfile;
@@ -57,6 +61,10 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
return jobId;
}
+ public ApplicationID getApplicationId() {
+ return applicationId;
+ }
+
public String getJobTargetAddress() {
return jobTargetAddress;
}
@@ -80,6 +88,8 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
+ slotID
+ ", jobId="
+ jobId
+ + ", applicationId="
+ + applicationId
+ ", jobTargetAddress='"
+ jobTargetAddress
+ '\''
@@ -101,6 +111,7 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
SlotAllocationSnapshot that = (SlotAllocationSnapshot) o;
return slotID.equals(that.slotID)
&& jobId.equals(that.jobId)
+ && applicationId.equals(that.applicationId)
&& jobTargetAddress.equals(that.jobTargetAddress)
&& allocationId.equals(that.allocationId)
&& resourceProfile.equals(that.resourceProfile);
@@ -108,6 +119,7 @@ public class SlotAllocationSnapshot implements java.io.Serializable {
@Override
public int hashCode() {
- return Objects.hash(slotID, jobId, jobTargetAddress, allocationId, resourceProfile);
+ return Objects.hash(
+ slotID, jobId, applicationId, jobTargetAddress, allocationId, resourceProfile);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index f03912bc857..6d2e112d6c6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -204,6 +204,8 @@ public class StreamGraph implements Pipeline, ExecutionPlan {
// declared
private byte[] serializedWatermarkDeclarations;
+ private final Set<String> userJarsToSkip = new HashSet<>();
+
public StreamGraph(
Configuration jobConfiguration,
ExecutionConfig executionConfig,
@@ -266,6 +268,10 @@ public class StreamGraph implements Pipeline, ExecutionPlan {
throw new IllegalArgumentException();
}
+ if (userJarsToSkip.contains(jar.getName())) {
+ return;
+ }
+
if (!userJars.contains(jar)) {
userJars.add(jar);
}
@@ -1302,6 +1308,12 @@ public class StreamGraph implements Pipeline, ExecutionPlan {
return Optional.ofNullable(applicationId);
}
+ public void addUserJarToSkip(String userJarName) {
+ LOG.info("Add user jar to skip uploading: {}", userJarName);
+ userJarsToSkip.add(userJarName);
+ userJars.removeIf(jar -> userJarName.equals(jar.getName()));
+ }
+
/**
* Sets the classpath required to run the job on a task manager.
*
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 40dcb223900..5d80bc7ea98 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.blob;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.BlobServerOptions;
@@ -63,9 +64,11 @@ class BlobCacheCleanupTest {
private final Random rnd = new Random();
+ private final ApplicationID applicationId = new ApplicationID();
+
/**
* Tests that {@link PermanentBlobCache} cleans up after calling {@link
- * PermanentBlobCache#releaseJob(JobID)}.
+ * PermanentBlobCache#releaseJob(JobID, ApplicationID)}.
*/
@Test
void testPermanentBlobCleanup() throws IOException, InterruptedException {
@@ -95,7 +98,7 @@ class BlobCacheCleanupTest {
checkFileCountForJob(0, jobId, cache);
// register once
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
checkFileCountForJob(2, jobId, server);
checkFileCountForJob(0, jobId, cache);
@@ -105,7 +108,7 @@ class BlobCacheCleanupTest {
}
// register again (let's say, from another thread or so)
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
for (PermanentBlobKey key : keys) {
cache.getFile(jobId, key);
}
@@ -115,14 +118,14 @@ class BlobCacheCleanupTest {
checkFileCountForJob(2, jobId, cache);
// after releasing once, nothing should change
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
assertThat(checkFilesExist(jobId, keys, cache, true)).isEqualTo(2);
checkFileCountForJob(2, jobId, server);
checkFileCountForJob(2, jobId, cache);
// after releasing the second time, the job is up for deferred cleanup
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
verifyJobCleanup(cache, jobId, keys);
// server should be unaffected
checkFileCountForJob(2, jobId, server);
@@ -160,37 +163,37 @@ class BlobCacheCleanupTest {
TestingBlobUtils.createPermanentCache(tempDir, config, serverAddress)) {
// register once
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isOne();
assertThat(cache.getJobRefCounters().get(jobId).keepUntil).isEqualTo(-1);
// register a second time
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isEqualTo(2);
assertThat(cache.getJobRefCounters().get(jobId).keepUntil).isEqualTo(-1);
// release once
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isOne();
assertThat(cache.getJobRefCounters().get(jobId).keepUntil).isEqualTo(-1);
// release a second time
long cleanupLowerBound =
System.currentTimeMillis() + config.get(BlobServerOptions.CLEANUP_INTERVAL);
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isZero();
assertThat(cache.getJobRefCounters().get(jobId).keepUntil)
.isGreaterThanOrEqualTo(cleanupLowerBound);
// register again
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isOne();
assertThat(cache.getJobRefCounters().get(jobId).keepUntil).isEqualTo(-1);
// finally release the job
cleanupLowerBound =
System.currentTimeMillis() + config.get(BlobServerOptions.CLEANUP_INTERVAL);
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
assertThat(cache.getJobRefCounters().get(jobId).references).isZero();
assertThat(cache.getJobRefCounters().get(jobId).keepUntil)
.isGreaterThanOrEqualTo(cleanupLowerBound);
@@ -199,8 +202,8 @@ class BlobCacheCleanupTest {
/**
* Tests the deferred cleanup of {@link PermanentBlobCache}, i.e. after calling {@link
- * PermanentBlobCache#releaseJob(JobID)} the file should be preserved a bit longer and then
- * cleaned up.
+ * PermanentBlobCache#releaseJob(JobID, ApplicationID)} the file should be preserved a bit
+ * longer and then cleaned up.
*/
@Test
@Disabled(
@@ -238,7 +241,7 @@ class BlobCacheCleanupTest {
checkBlobCacheSizeTracker(tracker, jobId, 0);
// register once
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
checkFileCountForJob(2, jobId, server);
checkFileCountForJob(0, jobId, cache);
@@ -249,7 +252,7 @@ class BlobCacheCleanupTest {
}
// register again (let's say, from another thread or so)
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
for (PermanentBlobKey key : keys) {
cache.readFile(jobId, key);
}
@@ -260,7 +263,7 @@ class BlobCacheCleanupTest {
checkBlobCacheSizeTracker(tracker, jobId, 2);
// after releasing once, nothing should change
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
assertThat(checkFilesExist(jobId, keys, cache, true)).isEqualTo(2);
checkFileCountForJob(2, jobId, server);
@@ -268,7 +271,7 @@ class BlobCacheCleanupTest {
checkBlobCacheSizeTracker(tracker, jobId, 2);
// after releasing the second time, the job is up for deferred cleanup
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
// files should still be accessible for now
assertThat(checkFilesExist(jobId, keys, cache, true)).isEqualTo(2);
@@ -414,8 +417,8 @@ class BlobCacheCleanupTest {
/**
* Checks that BLOBs for the given <tt>jobId</tt> are cleaned up eventually (after calling
- * {@link PermanentBlobCache#releaseJob(JobID)}, which is not done by this method!) (waits at
- * most 30s).
+ * {@link PermanentBlobCache#releaseJob(JobID, ApplicationID)}, which is not done by this
+ * method!) (waits at most 30s).
*
* @param cache BLOB server
* @param jobId job ID or <tt>null</tt> if job-unrelated
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/NoOpJobPermanentBlobService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/NoOpJobPermanentBlobService.java
index 7251642ded2..c85240b4f0e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/NoOpJobPermanentBlobService.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/NoOpJobPermanentBlobService.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.blob;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import java.io.File;
@@ -30,12 +31,12 @@ public class NoOpJobPermanentBlobService implements JobPermanentBlobService {
private NoOpJobPermanentBlobService() {}
@Override
- public void registerJob(JobID jobId) {
+ public void registerJob(JobID jobId, ApplicationID applicationId) {
// no-op
}
@Override
- public void releaseJob(JobID jobId) {
+ public void releaseJob(JobID jobId, ApplicationID applicationId) {
// no-op
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/TestingBlobHelpers.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/TestingBlobHelpers.java
index 696adba4b77..df4be740a71 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/TestingBlobHelpers.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/TestingBlobHelpers.java
@@ -190,16 +190,23 @@ public final class TestingBlobHelpers {
*
* @param expectedCount number of expected files in the blob service for the given application
* @param applicationId ID of an application
- * @param blobServer BLOB server to use
+ * @param blobService BLOB store to use
*/
public static void checkFileCountForApplication(
- int expectedCount, ApplicationID applicationId, BlobServer blobServer)
+ int expectedCount, ApplicationID applicationId, PermanentBlobService blobService)
throws IOException {
- final File applicationDir =
- blobServer
- .getStorageLocation(applicationId, new PermanentBlobKey())
- .getParentFile();
+ final File applicationDir;
+ if (blobService instanceof BlobServer) {
+ BlobServer server = (BlobServer) blobService;
+ applicationDir =
+ server.getStorageLocation(applicationId, new PermanentBlobKey())
+ .getParentFile();
+ } else {
+ PermanentBlobCache cache = (PermanentBlobCache) blobService;
+ applicationDir =
+ cache.getStorageLocation(applicationId, new PermanentBlobKey()).getParentFile();
+ }
File[] blobsForApplication = applicationDir.listFiles();
if (blobsForApplication == null) {
if (expectedCount != 0) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index 0048ab5d86a..ad4f10be633 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -1903,7 +1903,8 @@ public class DispatcherTest extends AbstractDispatcherTest {
highAvailabilityServices.getJobResultStore(),
jobManagerServices
.getLibraryCacheManager()
- .registerClassLoaderLease(graph.getJobID()),
+ .registerClassLoaderLease(
+ graph.getJobID(), graph.getApplicationId().orElseThrow()),
fatalErrorHandler);
}
@@ -1978,7 +1979,9 @@ public class DispatcherTest extends AbstractDispatcherTest {
highAvailabilityServices.getJobResultStore(),
jobManagerServices
.getLibraryCacheManager()
- .registerClassLoaderLease(executionPlan.getJobID()),
+ .registerClassLoaderLease(
+ executionPlan.getJobID(),
+ executionPlan.getApplicationId().orElseThrow()),
fatalErrorHandler);
}
@@ -2032,7 +2035,9 @@ public class DispatcherTest extends AbstractDispatcherTest {
highAvailabilityServices.getJobResultStore(),
jobManagerServices
.getLibraryCacheManager()
- .registerClassLoaderLease(executionPlan.getJobID()),
+ .registerClassLoaderLease(
+ executionPlan.getJobID(),
+ executionPlan.getApplicationId().orElseThrow()),
fatalErrorHandler);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
index 9fe89f97cf5..d401c409723 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
@@ -18,10 +18,12 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.PermanentBlobCache;
import org.apache.flink.runtime.blob.PermanentBlobKey;
@@ -51,6 +53,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import static org.apache.flink.runtime.blob.TestingBlobHelpers.checkFileCountForApplication;
import static org.apache.flink.runtime.blob.TestingBlobHelpers.checkFileCountForJob;
import static org.apache.flink.runtime.blob.TestingBlobHelpers.checkFilesExist;
import static org.hamcrest.Matchers.not;
@@ -75,6 +78,8 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
@Parameterized.Parameter public boolean wrapsSystemClassLoader;
+ private final ApplicationID applicationId = new ApplicationID();
+
/**
* Tests that the {@link BlobLibraryCacheManager} cleans up after the class loader leases for
* different jobs are closed.
@@ -112,8 +117,8 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
keys2.add(server.putPermanent(jobId2, buf));
libCache = createBlobLibraryCacheManager(cache);
- cache.registerJob(jobId1);
- cache.registerJob(jobId2);
+ cache.registerJob(jobId1, applicationId);
+ cache.registerJob(jobId2, applicationId);
assertEquals(0, libCache.getNumberOfManagedJobs());
assertEquals(0, libCache.getNumberOfReferenceHolders(jobId1));
@@ -123,7 +128,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(0, jobId2, cache);
final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob1 =
- libCache.registerClassLoaderLease(jobId1);
+ libCache.registerClassLoaderLease(jobId1, applicationId);
final UserCodeClassLoader classLoader1 =
classLoaderLeaseJob1.getOrResolveClassLoader(keys1, Collections.emptyList());
@@ -138,7 +143,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(0, jobId2, cache);
final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob2 =
- libCache.registerClassLoaderLease(jobId2);
+ libCache.registerClassLoaderLease(jobId2, applicationId);
final UserCodeClassLoader classLoader2 =
classLoaderLeaseJob2.getOrResolveClassLoader(keys2, Collections.emptyList());
assertThat(classLoader1, not(sameInstance(classLoader2)));
@@ -244,7 +249,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
keys.add(server.putPermanent(jobId, buf));
libCache = createBlobLibraryCacheManager(cache);
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
assertEquals(0, libCache.getNumberOfManagedJobs());
assertEquals(0, libCache.getNumberOfReferenceHolders(jobId));
@@ -252,7 +257,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(0, jobId, cache);
final LibraryCacheManager.ClassLoaderLease classLoaderLease1 =
- libCache.registerClassLoaderLease(jobId);
+ libCache.registerClassLoaderLease(jobId, applicationId);
UserCodeClassLoader classLoader1 =
classLoaderLease1.getOrResolveClassLoader(keys, Collections.emptyList());
@@ -263,7 +268,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(2, jobId, cache);
final LibraryCacheManager.ClassLoaderLease classLoaderLease2 =
- libCache.registerClassLoaderLease(jobId);
+ libCache.registerClassLoaderLease(jobId, applicationId);
final UserCodeClassLoader classLoader2 =
classLoaderLease2.getOrResolveClassLoader(keys, Collections.emptyList());
assertThat(classLoader1, sameInstance(classLoader2));
@@ -367,9 +372,9 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
{
Collection<PermanentBlobKey> keys = Collections.singleton(dataKey1);
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
final LibraryCacheManager.ClassLoaderLease classLoaderLease1 =
- libCache.registerClassLoaderLease(jobId);
+ libCache.registerClassLoaderLease(jobId, applicationId);
final UserCodeClassLoader classLoader1 =
classLoaderLease1.getOrResolveClassLoader(keys, Collections.emptyList());
assertEquals(1, libCache.getNumberOfManagedJobs());
@@ -379,7 +384,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(1, jobId, cache);
final LibraryCacheManager.ClassLoaderLease classLoaderLease2 =
- libCache.registerClassLoaderLease(jobId);
+ libCache.registerClassLoaderLease(jobId, applicationId);
final UserCodeClassLoader classLoader2 =
classLoaderLease2.getOrResolveClassLoader(keys, Collections.emptyList());
assertThat(classLoader1, sameInstance(classLoader2));
@@ -406,7 +411,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
checkFileCountForJob(2, jobId, server);
checkFileCountForJob(1, jobId, cache);
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
// library is still cached (but not associated with job any more)
checkFileCountForJob(2, jobId, server);
@@ -425,15 +430,15 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
// since we cannot download this library any more, this call should fail
try {
- cache.registerJob(jobId);
+ cache.registerJob(jobId, applicationId);
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libCache.registerClassLoaderLease(jobId);
+ libCache.registerClassLoaderLease(jobId, applicationId);
classLoaderLease.getOrResolveClassLoader(
Collections.singleton(dataKey2), Collections.emptyList());
fail("This should fail with an IOException");
} catch (IOException e) {
// splendid!
- cache.releaseJob(jobId);
+ cache.releaseJob(jobId, applicationId);
}
} finally {
if (cacheDir != null) {
@@ -460,7 +465,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
final BlobLibraryCacheManager libraryCacheManager = createSimpleBlobLibraryCacheManager();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
classLoaderLease.getOrResolveClassLoader(
Collections.singletonList(missingKey), Collections.emptyList());
@@ -471,7 +476,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
final BlobLibraryCacheManager libraryCacheManager = createSimpleBlobLibraryCacheManager();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
classLoaderLease.release();
@@ -488,9 +493,9 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
final JobID jobId = new JobID();
final LibraryCacheManager.ClassLoaderLease classLoaderLease1 =
- libraryCacheManager.registerClassLoaderLease(jobId);
+ libraryCacheManager.registerClassLoaderLease(jobId, applicationId);
final LibraryCacheManager.ClassLoaderLease classLoaderLease2 =
- libraryCacheManager.registerClassLoaderLease(jobId);
+ libraryCacheManager.registerClassLoaderLease(jobId, applicationId);
UserCodeClassLoader userCodeClassLoader =
classLoaderLease1.getOrResolveClassLoader(
@@ -516,9 +521,9 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
final JobID jobId = new JobID();
final LibraryCacheManager.ClassLoaderLease classLoaderLease1 =
- libraryCacheManager.registerClassLoaderLease(jobId);
+ libraryCacheManager.registerClassLoaderLease(jobId, applicationId);
final LibraryCacheManager.ClassLoaderLease classLoaderLease2 =
- libraryCacheManager.registerClassLoaderLease(jobId);
+ libraryCacheManager.registerClassLoaderLease(jobId, applicationId);
final UserCodeClassLoader classLoader1 =
classLoaderLease1.getOrResolveClassLoader(
@@ -535,7 +540,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
final BlobLibraryCacheManager libraryCacheManager = createSimpleBlobLibraryCacheManager();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
libraryCacheManager.shutdown();
@@ -551,7 +556,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
.build();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
UserCodeClassLoader userCodeClassLoader =
classLoaderLease.getOrResolveClassLoader(
Collections.emptyList(), Collections.emptyList());
@@ -573,7 +578,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
new TestingBlobLibraryCacheManagerBuilder().build();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
final UserCodeClassLoader userCodeClassLoader =
classLoaderLease.getOrResolveClassLoader(
Collections.emptyList(), Collections.emptyList());
@@ -594,7 +599,7 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
new TestingBlobLibraryCacheManagerBuilder().build();
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libraryCacheManager.registerClassLoaderLease(new JobID());
+ libraryCacheManager.registerClassLoaderLease(new JobID(), applicationId);
final UserCodeClassLoader userCodeClassLoader =
classLoaderLease.getOrResolveClassLoader(
Collections.emptyList(), Collections.emptyList());
@@ -661,6 +666,210 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
}
}
+ /** Tests that jobs with the same ApplicationID share the same jar. */
+ @Test
+ public void testApplicationLevelJarSharing() throws Exception {
+ JobID jobId1 = new JobID();
+ JobID jobId2 = new JobID();
+ List<PermanentBlobKey> keys = new ArrayList<>();
+ BlobServer server = null;
+ PermanentBlobCache cache = null;
+ BlobLibraryCacheManager libCache = null;
+
+ final byte[] buf = new byte[128];
+
+ try {
+ Configuration config = new Configuration();
+ config.set(BlobServerOptions.CLEANUP_INTERVAL, 1L);
+
+ server = new BlobServer(config, temporaryFolder.newFolder(), new VoidBlobStore());
+ server.start();
+
+ cache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ new VoidBlobStore(),
+ new InetSocketAddress("localhost", server.getPort()));
+
+ keys.add(server.putPermanent(applicationId, buf));
+
+ libCache = createBlobLibraryCacheManager(cache);
+ cache.registerJob(jobId1, applicationId);
+ cache.registerJob(jobId2, applicationId);
+
+ assertEquals(0, libCache.getNumberOfManagedJobs());
+ assertEquals(0, libCache.getNumberOfReferenceHolders(jobId1));
+ assertEquals(0, libCache.getNumberOfReferenceHolders(jobId2));
+ checkFileCountForJob(0, jobId1, server);
+ checkFileCountForJob(0, jobId1, cache);
+ checkFileCountForJob(0, jobId2, server);
+ checkFileCountForJob(0, jobId2, cache);
+ checkFileCountForApplication(1, applicationId, server);
+ checkFileCountForApplication(0, applicationId, cache);
+
+ final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob1 =
+ libCache.registerClassLoaderLease(jobId1, applicationId);
+ final UserCodeClassLoader classLoader1 =
+ classLoaderLeaseJob1.getOrResolveClassLoader(keys, Collections.emptyList());
+
+ assertEquals(1, libCache.getNumberOfManagedJobs());
+ assertEquals(1, libCache.getNumberOfReferenceHolders(jobId1));
+ assertEquals(0, libCache.getNumberOfReferenceHolders(jobId2));
+ checkFileCountForJob(0, jobId1, server);
+ checkFileCountForJob(0, jobId1, cache);
+ checkFileCountForJob(0, jobId2, server);
+ checkFileCountForJob(0, jobId2, cache);
+ checkFileCountForApplication(1, applicationId, server);
+ checkFileCountForApplication(1, applicationId, cache);
+
+ final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob2 =
+ libCache.registerClassLoaderLease(jobId2, applicationId);
+ final UserCodeClassLoader classLoader2 =
+ classLoaderLeaseJob2.getOrResolveClassLoader(keys, Collections.emptyList());
+
+ // Classloaders are not the same
+ assertThat(classLoader1, not(sameInstance(classLoader2)));
+
+ classLoaderLeaseJob1.release();
+ classLoaderLeaseJob2.release();
+ cache.releaseJob(jobId1, applicationId);
+ cache.releaseJob(jobId2, applicationId);
+
+ verifyApplicationCleanup(cache, applicationId, keys);
+ } finally {
+ if (libCache != null) {
+ libCache.shutdown();
+ }
+ if (cache != null) {
+ cache.close();
+ }
+ if (server != null) {
+ server.close();
+ }
+ }
+ }
+
+ /** Tests that jobs with different ApplicationIDs do not share the same classloader. */
+ @Test
+ public void testDifferentApplicationsNotSharing() throws Exception {
+ JobID jobId1 = new JobID();
+ JobID jobId2 = new JobID();
+ ApplicationID applicationId1 = new ApplicationID();
+ ApplicationID applicationId2 = new ApplicationID();
+ List<PermanentBlobKey> keys1 = new ArrayList<>();
+ List<PermanentBlobKey> keys2 = new ArrayList<>();
+ BlobServer server = null;
+ PermanentBlobCache cache = null;
+ BlobLibraryCacheManager libCache = null;
+
+ final byte[] buf = new byte[128];
+
+ try {
+ Configuration config = new Configuration();
+ config.set(BlobServerOptions.CLEANUP_INTERVAL, 1L);
+
+ server = new BlobServer(config, temporaryFolder.newFolder(), new VoidBlobStore());
+ server.start();
+
+ cache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ new VoidBlobStore(),
+ new InetSocketAddress("localhost", server.getPort()));
+
+ keys1.add(server.putPermanent(applicationId1, buf));
+ buf[0] += 1;
+ keys1.add(server.putPermanent(applicationId1, buf));
+ keys2.add(server.putPermanent(applicationId2, buf));
+
+ libCache = createBlobLibraryCacheManager(cache);
+ cache.registerJob(jobId1, applicationId1);
+ cache.registerJob(jobId2, applicationId2);
+
+ checkFileCountForApplication(2, applicationId1, server);
+ checkFileCountForApplication(0, applicationId1, cache);
+ checkFileCountForApplication(1, applicationId2, server);
+ checkFileCountForApplication(0, applicationId2, cache);
+
+ final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob1 =
+ libCache.registerClassLoaderLease(jobId1, applicationId1);
+ final UserCodeClassLoader classLoader1 =
+ classLoaderLeaseJob1.getOrResolveClassLoader(keys1, Collections.emptyList());
+
+ assertEquals(1, libCache.getNumberOfManagedJobs());
+ assertEquals(1, libCache.getNumberOfReferenceHolders(jobId1));
+ assertEquals(0, libCache.getNumberOfReferenceHolders(jobId2));
+ checkFileCountForApplication(2, applicationId1, server);
+ checkFileCountForApplication(2, applicationId1, cache);
+ checkFileCountForApplication(1, applicationId2, server);
+ checkFileCountForApplication(0, applicationId2, cache);
+
+ final LibraryCacheManager.ClassLoaderLease classLoaderLeaseJob2 =
+ libCache.registerClassLoaderLease(jobId2, applicationId2);
+ final UserCodeClassLoader classLoader2 =
+ classLoaderLeaseJob2.getOrResolveClassLoader(keys2, Collections.emptyList());
+ assertThat(classLoader1, not(sameInstance(classLoader2)));
+
+ checkFileCountForApplication(1, applicationId2, server);
+ checkFileCountForApplication(1, applicationId2, cache);
+
+ try {
+ classLoaderLeaseJob2.getOrResolveClassLoader(keys1, Collections.<URL>emptyList());
+ fail("Should fail with an IllegalStateException");
+ } catch (IllegalStateException e) {
+ // that's what we want
+ }
+
+ classLoaderLeaseJob1.release();
+ classLoaderLeaseJob2.release();
+ cache.releaseJob(jobId1, applicationId1);
+ cache.releaseJob(jobId2, applicationId2);
+
+ verifyApplicationCleanup(cache, applicationId1, keys1);
+ verifyApplicationCleanup(cache, applicationId2, keys2);
+ } finally {
+ if (libCache != null) {
+ libCache.shutdown();
+ }
+ if (cache != null) {
+ cache.close();
+ }
+ if (server != null) {
+ server.close();
+ }
+ }
+ }
+
+ /**
+ * Checks that BLOBs for the given <tt>applicationId</tt> are cleaned up eventually (after
+ * calling {@link PermanentBlobCache#releaseJob(JobID, ApplicationID)}, which is not done by
+ * this method!) (waits at most 30s).
+ *
+ * @param cache BLOB server
+ * @param applicationId application ID
+ * @param keys keys identifying BLOBs which were previously registered for the
+ * <tt>applicationId</tt>
+ */
+ private static void verifyApplicationCleanup(
+ PermanentBlobCache cache, ApplicationID applicationId, List<? extends BlobKey> keys)
+ throws InterruptedException, IOException {
+ // because we cannot guarantee that there are not thread races in the build system, we
+ // loop for a certain while until the references disappear
+ {
+ long deadline = System.currentTimeMillis() + 30_000L;
+ do {
+ Thread.sleep(100);
+ } while (checkFilesExist(applicationId, keys, cache, false) != 0
+ && System.currentTimeMillis() < deadline);
+ }
+
+ // the blob cache should no longer contain the files
+ // this fails if we exited via a timeout
+ checkFileCountForApplication(0, applicationId, cache);
+ }
+
private static final class TestingClassLoader extends URLClassLoader {
private boolean isClosed;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
index 0060675884a..44ca4a74620 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
@@ -114,6 +115,7 @@ public class BlobLibraryCacheRecoveryITCase extends TestLogger {
ArrayList<PermanentBlobKey> keys = new ArrayList<>(2);
JobID jobId = new JobID();
+ ApplicationID applicationId = new ApplicationID();
// Upload some data (libraries)
keys.add(server[0].putPermanent(jobId, expected)); // Request 1
@@ -130,7 +132,7 @@ public class BlobLibraryCacheRecoveryITCase extends TestLogger {
// Register uploaded libraries
final LibraryCacheManager.ClassLoaderLease classLoaderLease =
- libServer[0].registerClassLoaderLease(jobId);
+ libServer[0].registerClassLoaderLease(jobId, applicationId);
classLoaderLease.getOrResolveClassLoader(keys, Collections.emptyList());
// Verify key 1
@@ -216,4 +218,150 @@ public class BlobLibraryCacheRecoveryITCase extends TestLogger {
}
}
}
+
+ /**
+ * Integration test for application-level jar sharing across multiple jobs with the same
+ * ApplicationID.
+ */
+ @Test
+ public void testRecoveryWithApplicationLevelBlob() throws Exception {
+ Random rand = new Random();
+
+ BlobServer[] server = new BlobServer[2];
+ InetSocketAddress[] serverAddress = new InetSocketAddress[2];
+ BlobLibraryCacheManager[] libServer = new BlobLibraryCacheManager[2];
+ PermanentBlobCache cache = null;
+ BlobStoreService blobStoreService = null;
+
+ Configuration config = new Configuration();
+ config.set(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
+ config.set(
+ HighAvailabilityOptions.HA_STORAGE_PATH,
+ temporaryFolder.newFolder().getAbsolutePath());
+ config.set(BlobServerOptions.CLEANUP_INTERVAL, 3_600L);
+
+ final ExecutorService executorService = Executors.newSingleThreadExecutor();
+ try {
+ blobStoreService = BlobUtils.createBlobStoreFromConfig(config);
+
+ final BlobLibraryCacheManager.ClassLoaderFactory classLoaderFactory =
+ BlobLibraryCacheManager.defaultClassLoaderFactory(
+ FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST,
+ new String[0],
+ null,
+ true);
+
+ for (int i = 0; i < 2; i++) {
+ server[i] = new BlobServer(config, temporaryFolder.newFolder(), blobStoreService);
+ server[i].start();
+ serverAddress[i] = new InetSocketAddress("localhost", server[i].getPort());
+
+ PermanentBlobCache blobCache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ blobStoreService,
+ serverAddress[i]);
+
+ libServer[i] =
+ new BlobLibraryCacheManager(
+ blobCache, classLoaderFactory, wrapsSystemClassLoader);
+ }
+
+ // Random data
+ byte[] expected = new byte[1024];
+ rand.nextBytes(expected);
+
+ ArrayList<PermanentBlobKey> keys = new ArrayList<>(2);
+
+ JobID jobId = new JobID();
+ ApplicationID applicationId = new ApplicationID();
+
+ // Upload some data (libraries)
+ keys.add(server[0].putPermanent(applicationId, expected));
+
+ // The cache
+ cache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ blobStoreService,
+ serverAddress[0]);
+
+ // Register uploaded libraries
+ final LibraryCacheManager.ClassLoaderLease classLoaderLease =
+ libServer[0].registerClassLoaderLease(jobId, applicationId);
+ classLoaderLease.getOrResolveClassLoader(keys, Collections.emptyList());
+
+ // Verify file
+ File f = cache.getFile(applicationId, keys.get(0));
+ assertEquals(expected.length, f.length());
+
+ try (FileInputStream fis = new FileInputStream(f)) {
+ for (int i = 0; i < expected.length && fis.available() > 0; i++) {
+ assertEquals(expected[i], (byte) fis.read());
+ }
+
+ assertEquals(0, fis.available());
+ }
+
+ // Shutdown cache and start with other server
+ cache.close();
+
+ cache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ blobStoreService,
+ serverAddress[1]);
+
+ // Verify file
+ f = cache.getFile(applicationId, keys.get(0));
+ assertEquals(expected.length, f.length());
+
+ try (FileInputStream fis = new FileInputStream(f)) {
+ for (int i = 0; i < expected.length && fis.available() > 0; i++) {
+ assertEquals(expected[i], (byte) fis.read());
+ }
+
+ assertEquals(0, fis.available());
+ }
+
+ // Remove blobs again
+ server[1].globalCleanupAsync(applicationId, executorService).join();
+
+ // Verify everything is clean below recoveryDir/<cluster_id>
+ final String clusterId = config.get(HighAvailabilityOptions.HA_CLUSTER_ID);
+ String haBlobStorePath = config.get(HighAvailabilityOptions.HA_STORAGE_PATH);
+ File haBlobStoreDir = new File(haBlobStorePath, clusterId);
+ File[] recoveryFiles = haBlobStoreDir.listFiles();
+ assertNotNull("HA storage directory does not exist", recoveryFiles);
+ assertEquals(
+ "Unclean state backend: " + Arrays.toString(recoveryFiles),
+ 0,
+ recoveryFiles.length);
+ } finally {
+ assertThat(executorService.shutdownNow(), IsEmptyCollection.empty());
+
+ for (BlobLibraryCacheManager s : libServer) {
+ if (s != null) {
+ s.shutdown();
+ }
+ }
+ for (BlobServer s : server) {
+ if (s != null) {
+ s.close();
+ }
+ }
+
+ if (cache != null) {
+ cache.close();
+ }
+
+ if (blobStoreService != null) {
+ blobStoreService.cleanupAllData();
+ blobStoreService.close();
+ }
+ }
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java
index f0b848b700a..8bf10242e46 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/ContextClassLoaderLibraryCacheManager.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
/** {@link LibraryCacheManager} implementation which returns the context class loader. */
@@ -25,7 +26,7 @@ public enum ContextClassLoaderLibraryCacheManager implements LibraryCacheManager
INSTANCE;
@Override
- public ClassLoaderLease registerClassLoaderLease(JobID jobId) {
+ public ClassLoaderLease registerClassLoaderLease(JobID jobId, ApplicationID applicationId) {
return TestingClassLoaderLease.newBuilder().build();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/TestingLibraryCacheManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/TestingLibraryCacheManager.java
index 8e1257e46de..971412bad91 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/TestingLibraryCacheManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/TestingLibraryCacheManager.java
@@ -18,18 +18,19 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
-import java.util.function.Function;
+import java.util.function.BiFunction;
/** Testing {@link LibraryCacheManager} implementation. */
public class TestingLibraryCacheManager implements LibraryCacheManager {
- private final Function<JobID, LibraryCacheManager.ClassLoaderLease>
+ private final BiFunction<JobID, ApplicationID, ClassLoaderLease>
registerOrRetainClassLoaderFunction;
private final Runnable shutdownRunnable;
private TestingLibraryCacheManager(
- Function<JobID, LibraryCacheManager.ClassLoaderLease>
+ BiFunction<JobID, ApplicationID, LibraryCacheManager.ClassLoaderLease>
registerOrRetainClassLoaderFunction,
Runnable shutdownRunnable) {
this.registerOrRetainClassLoaderFunction = registerOrRetainClassLoaderFunction;
@@ -37,8 +38,9 @@ public class TestingLibraryCacheManager implements LibraryCacheManager {
}
@Override
- public LibraryCacheManager.ClassLoaderLease registerClassLoaderLease(JobID jobId) {
- return registerOrRetainClassLoaderFunction.apply(jobId);
+ public LibraryCacheManager.ClassLoaderLease registerClassLoaderLease(
+ JobID jobId, ApplicationID applicationId) {
+ return registerOrRetainClassLoaderFunction.apply(jobId, applicationId);
}
@Override
@@ -53,9 +55,9 @@ public class TestingLibraryCacheManager implements LibraryCacheManager {
public static final class Builder {
private Runnable shutdownRunnable = () -> {};
- private Function<JobID, LibraryCacheManager.ClassLoaderLease>
+ private BiFunction<JobID, ApplicationID, LibraryCacheManager.ClassLoaderLease>
registerOrRetainClassLoaderFunction =
- ignored -> TestingClassLoaderLease.newBuilder().build();
+ (ignored1, ignored2) -> TestingClassLoaderLease.newBuilder().build();
private Builder() {}
@@ -65,7 +67,8 @@ public class TestingLibraryCacheManager implements LibraryCacheManager {
}
public Builder setRegisterOrRetainClassLoaderFunction(
- Function<JobID, ClassLoaderLease> registerOrRetainClassLoaderFunction) {
+ BiFunction<JobID, ApplicationID, ClassLoaderLease>
+ registerOrRetainClassLoaderFunction) {
this.registerOrRetainClassLoaderFunction = registerOrRetainClassLoaderFunction;
return this;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTestUtils.java
index 6e5688327ad..68d98b71141 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTestUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobgraph;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import java.util.Arrays;
@@ -27,7 +28,9 @@ import java.util.function.Function;
public class JobGraphTestUtils {
public static JobGraph emptyJobGraph() {
- return JobGraphBuilder.newStreamingJobGraphBuilder().build();
+ final JobGraph jobGraph = JobGraphBuilder.newStreamingJobGraphBuilder().build();
+ jobGraph.setApplicationId(new ApplicationID());
+ return jobGraph;
}
public static JobGraph singleNoOpJobGraph() {
@@ -35,7 +38,10 @@ public class JobGraphTestUtils {
jobVertex.setInvokableClass(NoOpInvokable.class);
jobVertex.setParallelism(1);
- return JobGraphBuilder.newStreamingJobGraphBuilder().addJobVertex(jobVertex).build();
+ final JobGraph jobGraph =
+ JobGraphBuilder.newStreamingJobGraphBuilder().addJobVertex(jobVertex).build();
+ jobGraph.setApplicationId(new ApplicationID());
+ return jobGraph;
}
public static JobGraph streamingJobGraph(JobVertex... jobVertices) {
@@ -44,17 +50,23 @@ public class JobGraphTestUtils {
public static JobGraph streamingJobGraph(
Function<JobGraphBuilder, JobGraphBuilder> transformJgb, JobVertex... jobVertices) {
- return transformJgb
- .apply(
- JobGraphBuilder.newStreamingJobGraphBuilder()
- .addJobVertices(Arrays.asList(jobVertices)))
- .build();
+ final JobGraph jobGraph =
+ transformJgb
+ .apply(
+ JobGraphBuilder.newStreamingJobGraphBuilder()
+ .addJobVertices(Arrays.asList(jobVertices)))
+ .build();
+ jobGraph.setApplicationId(new ApplicationID());
+ return jobGraph;
}
public static JobGraph batchJobGraph(JobVertex... jobVertices) {
- return JobGraphBuilder.newBatchJobGraphBuilder()
- .addJobVertices(Arrays.asList(jobVertices))
- .build();
+ final JobGraph jobGraph =
+ JobGraphBuilder.newBatchJobGraphBuilder()
+ .addJobVertices(Arrays.asList(jobVertices))
+ .build();
+ jobGraph.setApplicationId(new ApplicationID());
+ return jobGraph;
}
private JobGraphTestUtils() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index ebacc4847d9..24a7be65256 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
@@ -499,6 +500,7 @@ class JobMasterTest {
@Override
public SlotPoolService createSlotPoolService(
@Nonnull JobID jobId,
+ @Nonnull ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
return new TestingSlotPool(jobId, hasReceivedSlotOffers);
@@ -1083,6 +1085,7 @@ class JobMasterTest {
final JobGraph inputSplitJobGraph =
JobGraphBuilder.newStreamingJobGraphBuilder().addJobVertex(source).build();
+ inputSplitJobGraph.setApplicationId(new ApplicationID());
RestartStrategyUtils.configureFixedDelayRestartStrategy(inputSplitJobGraph, 100, 0L);
try (final JobMaster jobMaster =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java
index ef486068c93..0b51a13734c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy;
import org.apache.flink.runtime.checkpoint.Checkpoints;
import org.apache.flink.runtime.checkpoint.OperatorState;
@@ -108,11 +109,15 @@ public class TestUtils {
final JobCheckpointingSettings checkpointingSettings =
new JobCheckpointingSettings(checkpointCoordinatorConfiguration, null);
- return JobGraphBuilder.newStreamingJobGraphBuilder()
- .addJobVertices(Arrays.asList(jobVertices))
- .setJobCheckpointingSettings(checkpointingSettings)
- .setSavepointRestoreSettings(savepointRestoreSettings)
- .build();
+ final JobGraph jobGraph =
+ JobGraphBuilder.newStreamingJobGraphBuilder()
+ .addJobVertices(Arrays.asList(jobVertices))
+ .setJobCheckpointingSettings(checkpointingSettings)
+ .setSavepointRestoreSettings(savepointRestoreSettings)
+ .build();
+ jobGraph.setApplicationId(new ApplicationID());
+
+ return jobGraph;
}
private TestUtils() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractDeclarativeSlotPoolBridgeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractDeclarativeSlotPoolBridgeTest.java
index d2cb95ebe2e..13c3a1f3342 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractDeclarativeSlotPoolBridgeTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractDeclarativeSlotPoolBridgeTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -43,6 +44,7 @@ abstract class AbstractDeclarativeSlotPoolBridgeTest {
protected static final Duration RPC_TIMEOUT = Duration.ofSeconds(20);
protected static final JobID JOB_ID = new JobID();
+ protected static final ApplicationID APPLICATION_ID = new ApplicationID();
protected static final JobMasterId JOB_MASTER_ID = JobMasterId.generate();
protected final ComponentMainThreadExecutor componentMainThreadExecutor = forMainThread();
@@ -106,6 +108,7 @@ abstract class AbstractDeclarativeSlotPoolBridgeTest {
ComponentMainThreadExecutor componentMainThreadExecutor) {
return new DeclarativeSlotPoolBridge(
JOB_ID,
+ APPLICATION_ID,
declarativeSlotPoolFactory,
SystemClock.getInstance(),
RPC_TIMEOUT,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java
index 7241ffc1075..4ca47b3173a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -40,6 +41,7 @@ import static org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorSer
public class DeclarativeSlotPoolBridgeBuilder {
private JobID jobId = new JobID();
+ private ApplicationID applicationId = new ApplicationID();
private Duration batchSlotTimeout = JobManagerOptions.SLOT_IDLE_TIMEOUT.defaultValue();
private Duration idleSlotTimeout = TestingUtils.infiniteDuration();
private Clock clock = SystemClock.getInstance();
@@ -105,6 +107,7 @@ public class DeclarativeSlotPoolBridgeBuilder {
public DeclarativeSlotPoolBridge build() {
return new DeclarativeSlotPoolBridge(
jobId,
+ applicationId,
new DefaultDeclarativeSlotPoolFactory(),
clock,
TestingUtils.infiniteDuration(),
@@ -120,6 +123,7 @@ public class DeclarativeSlotPoolBridgeBuilder {
final DeclarativeSlotPoolBridge slotPool =
new DeclarativeSlotPoolBridge(
jobId,
+ applicationId,
new DefaultDeclarativeSlotPoolFactory(),
clock,
TestingUtils.infiniteDuration(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java
index a97ac818963..60cfc15f46e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -52,6 +53,7 @@ class DeclarativeSlotPoolBridgePreferredAllocationsTest {
final DeclarativeSlotPoolBridge declarativeSlotPoolBridge =
new DeclarativeSlotPoolBridge(
new JobID(),
+ new ApplicationID(),
new DefaultDeclarativeSlotPoolFactory(),
SystemClock.getInstance(),
TestingUtils.infiniteDuration(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
index f74d5f0410f..362e1c1425f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -64,6 +65,7 @@ import static org.assertj.core.api.Assertions.assertThat;
class DeclarativeSlotPoolServiceTest {
private static final JobID jobId = new JobID();
+ private static final ApplicationID applicationId = new ApplicationID();
private static final JobMasterId jobMasterId = JobMasterId.generate();
private final ComponentMainThreadExecutor mainThreadExecutor = forMainThread();
private static final String address = "localhost";
@@ -359,6 +361,7 @@ class DeclarativeSlotPoolServiceTest {
final DeclarativeSlotPoolService declarativeSlotPoolService =
new DeclarativeSlotPoolService(
jobId,
+ applicationId,
declarativeSlotPoolFactory,
SystemClock.getInstance(),
Duration.ofSeconds(20L),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
index c863727246e..16ae12dac09 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolServiceBuilder.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -71,6 +72,7 @@ public class TestingSlotPoolServiceBuilder implements SlotPoolServiceFactory {
@Override
public SlotPoolService createSlotPoolService(
@Nonnull JobID jobId,
+ @Nonnull ApplicationID applicationId,
DeclarativeSlotPoolFactory declarativeSlotPoolFactory,
@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor) {
return new TestingSlotPoolService(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
index a5a3fc5c396..c7f37427aaf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
@@ -17,8 +17,9 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -154,19 +155,22 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
throws Exception {
final ResourceID resourceID = ResourceID.generate();
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final SlotID slotId = SlotID.getDynamicSlotID(resourceID);
final String targetAddress = "localhost";
final ResourceRequirements requirements =
ResourceRequirements.create(
jobId,
+ applicationId,
targetAddress,
Collections.singleton(
ResourceRequirement.create(DEFAULT_SLOT_RESOURCE_PROFILE, 1)));
final CompletableFuture<
- Tuple6<
+ Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
@@ -176,8 +180,8 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- requestFuture.complete(tuple6);
+ tuple7 -> {
+ requestFuture.complete(tuple7);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
@@ -222,10 +226,11 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
assertThat(assertFutureCompleteAndReturn(requestFuture))
.isEqualTo(
- Tuple6.of(
+ Tuple7.of(
slotId,
jobId,
- assertFutureCompleteAndReturn(requestFuture).f2,
+ applicationId,
+ assertFutureCompleteAndReturn(requestFuture).f3,
DEFAULT_SLOT_RESOURCE_PROFILE,
targetAddress,
getResourceManagerId()));
@@ -233,10 +238,10 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
final TaskManagerSlotInformation slot =
getTaskManagerTracker()
.getAllocatedOrPendingSlot(
- assertFutureCompleteAndReturn(requestFuture).f2)
+ assertFutureCompleteAndReturn(requestFuture).f3)
.get();
- assertThat(assertFutureCompleteAndReturn(requestFuture).f2)
+ assertThat(assertFutureCompleteAndReturn(requestFuture).f3)
.as(
"The slot has not been allocated to the expected allocation id.")
.isEqualTo(slot.getAllocationId());
@@ -368,11 +373,11 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
+ tuple7 -> {
if (!allocationIdFuture1.isDone()) {
- allocationIdFuture1.complete(tuple6.f2);
+ allocationIdFuture1.complete(tuple7.f3);
} else {
- allocationIdFuture2.complete(tuple6.f2);
+ allocationIdFuture2.complete(tuple7.f3);
}
return CompletableFuture.completedFuture(Acknowledge.get());
})
@@ -467,8 +472,8 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- allocationIdFuture.complete(tuple6.f2);
+ tuple7 -> {
+ allocationIdFuture.complete(tuple7.f3);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
@@ -524,8 +529,8 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- allocationIdFuture.complete(tuple6.f2);
+ tuple7 -> {
+ allocationIdFuture.complete(tuple7.f3);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
@@ -551,6 +556,8 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
.processResourceRequirements(
ResourceRequirements.empty(
resourceRequirements.getJobId(),
+ resourceRequirements
+ .getApplicationId(),
resourceRequirements
.getTargetAddress()));
getSlotManager()
@@ -655,7 +662,7 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
.setRequestSlotFunction(
FunctionUtils.uncheckedFunction(
requestSlotParameters -> {
- allocationIds.put(requestSlotParameters.f2);
+ allocationIds.put(requestSlotParameters.f3);
return slotRequestFutureIterator.next();
}))
.createTestingTaskExecutorGateway();
@@ -783,8 +790,8 @@ abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSlotManag
// it is important that the returned future is already completed
// otherwise it will be cancelled when the task executor is unregistered
.setRequestSlotFunction(
- tuple6 -> {
- allocationIdFuture.complete(tuple6.f2);
+ tuple7 -> {
+ allocationIdFuture.complete(tuple7.f3);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
index cea99fe6537..11290f841ff 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -94,6 +95,7 @@ class DefaultSlotStatusSyncerTest {
taskExecutorConnection, ResourceProfile.ANY, ResourceProfile.ANY);
final ResourceTracker resourceTracker = new DefaultResourceTracker();
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final SlotStatusSyncer slotStatusSyncer =
new DefaultSlotStatusSyncer(TASK_MANAGER_REQUEST_TIMEOUT);
slotStatusSyncer.initialize(
@@ -106,6 +108,7 @@ class DefaultSlotStatusSyncerTest {
slotStatusSyncer.allocateSlot(
taskExecutorConnection.getInstanceID(),
jobId,
+ applicationId,
"address",
ResourceProfile.ANY);
@@ -188,6 +191,7 @@ class DefaultSlotStatusSyncerTest {
final TaskExecutorConnection taskExecutorConnection =
new TaskExecutorConnection(ResourceID.generate(), taskExecutorGateway);
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final AllocationID allocationId1 = new AllocationID();
final AllocationID allocationId2 = new AllocationID();
final SlotID slotId1 = new SlotID(taskExecutorConnection.getResourceID(), 0);
@@ -222,7 +226,7 @@ class DefaultSlotStatusSyncerTest {
assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2)).isPresent();
slotStatusSyncer.allocateSlot(
- taskExecutorConnection.getInstanceID(), jobId, "address", resource);
+ taskExecutorConnection.getInstanceID(), jobId, applicationId, "address", resource);
assertThat(resourceTracker.getAcquiredResources(jobId))
.contains(ResourceRequirement.create(resource, 3));
assertThat(
@@ -279,8 +283,8 @@ class DefaultSlotStatusSyncerTest {
final TestingTaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- requestFuture.complete(tuple6.f2);
+ tuple7 -> {
+ requestFuture.complete(tuple7.f3);
return responseFuture;
})
.createTestingTaskExecutorGateway();
@@ -290,6 +294,7 @@ class DefaultSlotStatusSyncerTest {
taskExecutorConnection, ResourceProfile.ANY, ResourceProfile.ANY);
final ResourceTracker resourceTracker = new DefaultResourceTracker();
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final SlotStatusSyncer slotStatusSyncer =
new DefaultSlotStatusSyncer(TASK_MANAGER_REQUEST_TIMEOUT);
slotStatusSyncer.initialize(
@@ -302,6 +307,7 @@ class DefaultSlotStatusSyncerTest {
slotStatusSyncer.allocateSlot(
taskExecutorConnection.getInstanceID(),
jobId,
+ applicationId,
"address",
ResourceProfile.ANY);
final AllocationID allocationId = requestFuture.get();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
index 5f374f2f9d1..2f9cc943121 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java
@@ -17,9 +17,10 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -354,9 +355,10 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
@Test
void testSlotAllocationAccordingToStrategyResult() throws Exception {
final CompletableFuture<
- Tuple6<
+ Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
@@ -365,8 +367,8 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- requestSlotFuture.complete(tuple6);
+ tuple7 -> {
+ requestSlotFuture.complete(tuple7);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
@@ -398,16 +400,17 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
.processResourceRequirements(
createResourceRequirements(jobId, 1));
});
- final Tuple6<
+ final Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
ResourceManagerId>
requestSlot = assertFutureCompleteAndReturn(requestSlotFuture);
assertThat(requestSlot.f1).isEqualTo(jobId);
- assertThat(requestSlot.f3).isEqualTo(DEFAULT_SLOT_RESOURCE_PROFILE);
+ assertThat(requestSlot.f4).isEqualTo(DEFAULT_SLOT_RESOURCE_PROFILE);
});
}
};
@@ -466,9 +469,10 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
new PendingTaskManager(
DEFAULT_TOTAL_RESOURCE_PROFILE, DEFAULT_NUM_SLOTS_PER_WORKER);
final CompletableFuture<
- Tuple6<
+ Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
@@ -477,8 +481,8 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
final TaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setRequestSlotFunction(
- tuple6 -> {
- requestSlotFuture.complete(tuple6);
+ tuple7 -> {
+ requestSlotFuture.complete(tuple7);
return CompletableFuture.completedFuture(Acknowledge.get());
})
.createTestingTaskExecutorGateway();
@@ -517,16 +521,17 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
new SlotReport(),
DEFAULT_TOTAL_RESOURCE_PROFILE,
DEFAULT_SLOT_RESOURCE_PROFILE));
- final Tuple6<
+ final Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
ResourceManagerId>
requestSlot = assertFutureCompleteAndReturn(requestSlotFuture);
assertThat(requestSlot.f1).isEqualTo(jobId);
- assertThat(requestSlot.f3).isEqualTo(DEFAULT_SLOT_RESOURCE_PROFILE);
+ assertThat(requestSlot.f4).isEqualTo(DEFAULT_SLOT_RESOURCE_PROFILE);
});
}
};
@@ -940,6 +945,7 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
new CompletableFuture<>();
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final TestingTaskExecutorGateway taskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
.setFreeInactiveSlotsConsumer(
@@ -981,7 +987,9 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
getSlotManager()
.processResourceRequirements(
ResourceRequirements.empty(
- jobId, "foobar")));
+ jobId,
+ applicationId,
+ "foobar")));
assertFutureNotComplete(freeInactiveSlotsJobIdFuture);
// clear requirements, which should trigger slots being reclaimed
@@ -1000,6 +1008,7 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
new Context() {
{
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final CompletableFuture<Void> allocateResourceFuture = new CompletableFuture<>();
resourceAllocatorBuilder.setDeclareResourceNeededConsumer(
@@ -1042,7 +1051,9 @@ class FineGrainedSlotManagerTest extends FineGrainedSlotManagerTestBase {
getSlotManager()
.processResourceRequirements(
ResourceRequirements.empty(
- jobId, "foobar")));
+ jobId,
+ applicationId,
+ "foobar")));
// disconnect to job master,will trigger
// PendingTaskManager#clearPendingAllocationsOfJob again
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorExecutionDeploymentReconciliationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorExecutionDeploymentReconciliationTest.java
index e9e73da5e28..ae0aca2b667 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorExecutionDeploymentReconciliationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorExecutionDeploymentReconciliationTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.AllCallbackWrapper;
@@ -99,6 +100,7 @@ class TaskExecutorExecutionDeploymentReconciliationTest {
private final SettableLeaderRetrievalService resourceManagerLeaderRetriever =
new SettableLeaderRetrievalService();
private final JobID jobId = new JobID();
+ private final ApplicationID applicationId = new ApplicationID();
@RegisterExtension
static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
@@ -346,6 +348,7 @@ class TaskExecutorExecutionDeploymentReconciliationTest {
.requestSlot(
slotStatusOptional.get().getSlotID(),
jobId,
+ applicationId,
allocationId,
ResourceProfile.ZERO,
jobMasterAddress,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
index 9164eef6214..a06b85edb0e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.BatchExecutionOptions;
@@ -119,6 +120,7 @@ class TaskExecutorPartitionLifecycleTest {
private final SettableLeaderRetrievalService resourceManagerLeaderRetriever =
new SettableLeaderRetrievalService();
private final JobID jobId = new JobID();
+ private final ApplicationID applicationId = new ApplicationID();
private Configuration configuration;
@@ -611,6 +613,7 @@ class TaskExecutorPartitionLifecycleTest {
.requestSlot(
slotStatus.getSlotID(),
jobId,
+ applicationId,
taskDeploymentDescriptor.getAllocationId(),
ResourceProfile.ZERO,
jobMasterAddress,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
index 25f2970c365..caa16b1ab28 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateRecoveryOptions;
@@ -115,6 +116,7 @@ class TaskExecutorRecoveryTest {
testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final TestingHighAvailabilityServices highAvailabilityServices =
new TestingHighAvailabilityServices();
@@ -160,6 +162,7 @@ class TaskExecutorRecoveryTest {
.requestSlot(
allocatedSlotID,
jobId,
+ applicationId,
allocationId,
slotStatus.getResourceProfile(),
"localhost",
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSlotLifetimeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSlotLifetimeTest.java
index 7573520e148..46dd50f2335 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSlotLifetimeTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSlotLifetimeTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.AllCallbackWrapper;
@@ -167,11 +168,13 @@ class TaskExecutorSlotLifetimeTest {
taskExecutor.getSelfGateway(TaskExecutorGateway.class);
final JobID jobId = new JobID();
+ final ApplicationID applicationId = new ApplicationID();
final AllocationID allocationId = new AllocationID();
taskExecutorGateway
.requestSlot(
firstSlotId,
jobId,
+ applicationId,
allocationId,
ResourceProfile.ZERO,
jobMasterGateway.getAddress(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index 276353987f5..11dc2e2c532 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.resources.CPUResource;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -210,6 +211,8 @@ class TaskExecutorTest {
private JobID jobId;
private JobID jobId2;
+ private ApplicationID applicationId;
+
private TestingFatalErrorHandler testingFatalErrorHandler;
private TestingHighAvailabilityServices haServices;
@@ -231,6 +234,7 @@ class TaskExecutorTest {
unresolvedTaskManagerLocation = new LocalUnresolvedTaskManagerLocation();
jobId = new JobID();
jobId2 = new JobID();
+ applicationId = new ApplicationID();
testingFatalErrorHandler = new TestingFatalErrorHandler();
@@ -1846,6 +1850,7 @@ class TaskExecutorTest {
taskExecutorGateway.requestSlot(
new SlotID(resourceId, 0),
jobId,
+ applicationId,
new AllocationID(),
ResourceProfile.ZERO,
"foobar",
@@ -2336,6 +2341,7 @@ class TaskExecutorTest {
taskExecutorGateway.requestSlot(
new SlotID(taskExecutor.getResourceID(), 0),
jobId,
+ applicationId,
allocationIdInBoth,
ResourceProfile.ZERO,
"foobar",
@@ -2344,6 +2350,7 @@ class TaskExecutorTest {
taskExecutorGateway.requestSlot(
new SlotID(taskExecutor.getResourceID(), 1),
jobId,
+ applicationId,
allocationIdOnlyInTM,
ResourceProfile.ZERO,
"foobar",
@@ -2462,6 +2469,7 @@ class TaskExecutorTest {
.requestSlot(
new SlotID(taskExecutorResourceId, 0),
jobId,
+ applicationId,
new AllocationID(),
ResourceProfile.ZERO,
"foobar",
@@ -3003,7 +3011,16 @@ class TaskExecutorTest {
String address,
ResourceManagerId token)
throws InterruptedException, ExecutionException {
- gateway.requestSlot(slotId, jobId, allocationId, profile, address, token, timeout).get();
+ gateway.requestSlot(
+ slotId,
+ jobId,
+ applicationId,
+ allocationId,
+ profile,
+ address,
+ token,
+ timeout)
+ .get();
}
private SlotID buildSlotID(int slotIndex) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
index 4c7618dc894..cf1b79c68e6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -18,8 +18,9 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -76,7 +77,14 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
submitTaskConsumer;
private final Function<
- Tuple6<SlotID, JobID, AllocationID, ResourceProfile, String, ResourceManagerId>,
+ Tuple7<
+ SlotID,
+ JobID,
+ ApplicationID,
+ AllocationID,
+ ResourceProfile,
+ String,
+ ResourceManagerId>,
CompletableFuture<Acknowledge>>
requestSlotFunction;
@@ -134,9 +142,10 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
BiFunction<TaskDeploymentDescriptor, JobMasterId, CompletableFuture<Acknowledge>>
submitTaskConsumer,
Function<
- Tuple6<
+ Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
@@ -203,15 +212,17 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
public CompletableFuture<Acknowledge> requestSlot(
SlotID slotId,
JobID jobId,
+ ApplicationID applicationId,
AllocationID allocationId,
ResourceProfile resourceProfile,
String targetAddress,
ResourceManagerId resourceManagerId,
Duration timeout) {
return requestSlotFunction.apply(
- Tuple6.of(
+ Tuple7.of(
slotId,
jobId,
+ applicationId,
allocationId,
resourceProfile,
targetAddress,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
index 6f21873af39..988dcdf1f3c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
@@ -18,8 +18,9 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -68,7 +69,14 @@ public class TestingTaskExecutorGatewayBuilder {
NOOP_SUBMIT_TASK_CONSUMER =
(ignoredA, ignoredB) -> CompletableFuture.completedFuture(Acknowledge.get());
private static final Function<
- Tuple6<SlotID, JobID, AllocationID, ResourceProfile, String, ResourceManagerId>,
+ Tuple7<
+ SlotID,
+ JobID,
+ ApplicationID,
+ AllocationID,
+ ResourceProfile,
+ String,
+ ResourceManagerId>,
CompletableFuture<Acknowledge>>
NOOP_REQUEST_SLOT_FUNCTION =
ignored -> CompletableFuture.completedFuture(Acknowledge.get());
@@ -128,7 +136,14 @@ public class TestingTaskExecutorGatewayBuilder {
private BiFunction<TaskDeploymentDescriptor, JobMasterId, CompletableFuture<Acknowledge>>
submitTaskConsumer = NOOP_SUBMIT_TASK_CONSUMER;
private Function<
- Tuple6<SlotID, JobID, AllocationID, ResourceProfile, String, ResourceManagerId>,
+ Tuple7<
+ SlotID,
+ JobID,
+ ApplicationID,
+ AllocationID,
+ ResourceProfile,
+ String,
+ ResourceManagerId>,
CompletableFuture<Acknowledge>>
requestSlotFunction = NOOP_REQUEST_SLOT_FUNCTION;
private Function<JobID, CompletableFuture<Collection<PartitionWithMetrics>>>
@@ -209,9 +224,10 @@ public class TestingTaskExecutorGatewayBuilder {
public TestingTaskExecutorGatewayBuilder setRequestSlotFunction(
Function<
- Tuple6<
+ Tuple7<
SlotID,
JobID,
+ ApplicationID,
AllocationID,
ResourceProfile,
String,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/FileSlotAllocationSnapshotPersistenceServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/FileSlotAllocationSnapshotPersistenceServiceTest.java
index 62edccd2088..b79eef33efe 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/FileSlotAllocationSnapshotPersistenceServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/FileSlotAllocationSnapshotPersistenceServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.taskexecutor.slot;
+import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -144,6 +145,7 @@ class FileSlotAllocationSnapshotPersistenceServiceTest {
new SlotAllocationSnapshot(
new SlotID(resourceId, slotIndex),
new JobID(),
+ new ApplicationID(),
"foobar",
new AllocationID(),
ResourceProfile.UNKNOWN));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphTest.java
index 00ab48566e5..5e1387848ee 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphTest.java
@@ -26,6 +26,9 @@ import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.File;
+import java.net.URL;
+import java.util.Collections;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -85,4 +88,30 @@ class StreamGraphTest {
}
return null;
}
+
+ @Test
+ void testSkippedUserJar() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.fromSequence(1, 100).sinkTo(new DiscardingSink<>());
+ final StreamGraph streamGraph = env.getStreamGraph();
+
+ // Create a temporary jar file
+ final File tempJar = File.createTempFile("stream_graph_test", ".jar");
+ tempJar.deleteOnExit();
+
+ final URL jarUrl = tempJar.toURI().toURL();
+ final String jarName = tempJar.getName();
+
+ // Add the jar - it should be removed later
+ streamGraph.addJars(Collections.singletonList(jarUrl));
+
+ // Add jar to skip list
+ streamGraph.addUserJarToSkip(jarName);
+
+ // Try to add the jar again - it should be skipped
+ streamGraph.addJars(Collections.singletonList(jarUrl));
+
+ // Verify that the jar was not added
+ Assertions.assertTrue(streamGraph.getUserJars().isEmpty());
+ }
}
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/application/ScriptExecutorITCase.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/application/ScriptExecutorITCase.java
index 066831127e0..2771911abc5 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/application/ScriptExecutorITCase.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/application/ScriptExecutorITCase.java
@@ -86,6 +86,7 @@ public class ScriptExecutorITCase extends AbstractSqlGatewayStatementITCaseBase
false,
false,
null,
+ null,
Collections.emptyList());
executor =