This is an automated email from the ASF dual-hosted git repository.
rmetzger 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 2b978526a39 [FLINK-32212][runtime] Match library BLOBs by content in
BlobLibraryCacheManager (#27579)
2b978526a39 is described below
commit 2b978526a3921c7a16f804d81b7a452ccb5e2a57
Author: Natea Eshetu Beshada <[email protected]>
AuthorDate: Fri Aug 14 00:11:43 2026 -0700
[FLINK-32212][runtime] Match library BLOBs by content in
BlobLibraryCacheManager (#27579)
A PermanentBlobKey carries a random component on top of the content hash
(FLINK-7140), so re-uploading an unchanged JAR yields a key that is unequal
to the one a cached class loader was resolved with. When a JobManager fails
over and resubmits the job, TaskExecutors that survived the failover still
hold a class loader resolved with the old keys, so verifyClassLoader()
throws
The library registration references a different set of library BLOBs than
previous registrations for this job:
old:[p-<hash>-<random1>] new:[p-<hash>-<random2>]
even though the JAR content is identical. Every subsequent deployment
attempt
hits the same mismatch, so the job restarts indefinitely.
verifyClassLoader() now compares the content hashes of the required BLOBs
instead of whole keys. Identical content therefore resolves to the cached
class loader regardless of the random component, while a genuinely different
set of JARs or class paths still fails as before.
The comparison is over the set of distinct contents, so adding or dropping a
duplicate of a JAR already in the required library set is also accepted -
the
resolved class loader can serve it either way. This makes the check
symmetric.
The previous size-then-containsAll comparison was not: it accepted a
registration of [A, A] followed by [A], but rejected [A, A] followed by
[A, A], even though both describe the same libraries.
Comparing by content rather than re-creating the class loader keeps the
existing contract check intact. A single ResolvedClassLoader is shared by
all
leases of a job, so replacing it on a mismatch would invalidate the class
loader other leases already resolved, let old and new class loaders for the
same job coexist, and orphan any release hooks registered against the
replaced instance.
---
.../e5126cae-f3fe-48aa-b6fb-60ae6cc3fcd5 | 5 --
.../org/apache/flink/runtime/blob/BlobKey.java | 6 +-
.../librarycache/BlobLibraryCacheManager.java | 55 +++++++++++----
.../librarycache/LibraryCacheManager.java | 3 +
.../librarycache/BlobLibraryCacheManagerTest.java | 81 ++++++++++++++++++++++
5 files changed, 130 insertions(+), 20 deletions(-)
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 f5af66381c0..3a6ae92e8f7 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
@@ -1,11 +1,6 @@
Constructor
<org.apache.flink.runtime.state.heap.CopyOnWriteStateMapSnapshot.<init>(org.apache.flink.runtime.state.heap.CopyOnWriteStateMap)>
calls method
<org.apache.flink.runtime.state.heap.CopyOnWriteStateMap.snapshotMapArrays()>
in (CopyOnWriteStateMapSnapshot.java:86)
Constructor
<org.apache.flink.streaming.runtime.io.StreamTaskExternallyInducedSourceInput.<init>(org.apache.flink.streaming.api.operators.SourceOperator,
java.util.function.Consumer, int, int)> calls method
<org.apache.flink.streaming.api.operators.SourceOperator.getSourceReader()> in
(StreamTaskExternallyInducedSourceInput.java:41)
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: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)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
index bf89636df1c..1f802bf2a7b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
@@ -159,9 +159,11 @@ public abstract class BlobKey implements Serializable,
Comparable<BlobKey> {
/**
* Returns the hash component of this key.
*
- * @return a 20 bit hash of the contents the key refers to
+ * <p>Unlike {@link #equals(Object)}, this excludes the random component
of the key, so two keys
+ * referring to the same content share the same hash.
+ *
+ * @return a 20 byte hash of the contents the key refers to
*/
- @VisibleForTesting
public byte[] getHash() {
return key;
}
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 c031e049aff..395815074e8 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
@@ -28,6 +28,7 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkUserCodeClassLoader;
import org.apache.flink.util.FlinkUserCodeClassLoaders;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
import org.apache.flink.util.UserCodeClassLoader;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.type.TypeFactory;
@@ -391,12 +392,27 @@ public class BlobLibraryCacheManager implements
LibraryCacheManager {
private final ClassLoader classLoader;
/**
- * Set of BLOB keys used for a previous job/task registration.
+ * Set of BLOB keys used for a previous job/task registration, kept
for error messages and
+ * logging only. Later registrations are checked against {@link
#libraryHashes}.
+ */
+ private final Set<PermanentBlobKey> libraries;
+
+ /**
+ * Content hashes of {@link #libraries}.
*
* <p>The purpose of this is to make sure, future registrations do not
differ in content as
* this is a contract of the {@link BlobLibraryCacheManager}.
+ *
+ * <p>A {@link PermanentBlobKey} contains a random component on top of
the content hash (see
+ * FLINK-7140), so the same JAR uploaded twice yields two unequal
keys. Comparing content
+ * hashes instead of whole keys keeps the contract check meaningful
while tolerating the
+ * re-upload of unchanged JARs that happens when a JobManager fails
over (FLINK-32212).
+ *
+ * <p>Note that this compares the set of distinct contents, so a
registration that only adds
+ * or drops a duplicate of a JAR already in the required library set
is accepted: the
+ * resolved class loader can serve it either way.
*/
- private final Set<PermanentBlobKey> libraries;
+ private final Set<String> libraryHashes;
/**
* Set of class path URLs used for a previous job/task registration.
@@ -425,6 +441,7 @@ public class BlobLibraryCacheManager implements
LibraryCacheManager {
classPaths.add(url.toString());
}
this.libraries = new HashSet<>(requiredLibraries);
+ this.libraryHashes = hashesOf(requiredLibraries);
this.wrapsSystemClassLoader = wrapsSystemClassLoader;
this.releaseHooks = new HashMap<>();
@@ -444,21 +461,25 @@ public class BlobLibraryCacheManager implements
LibraryCacheManager {
Collection<PermanentBlobKey> requiredLibraries,
Collection<URL> requiredClassPaths) {
// Make sure the previous registration referred to the same
libraries and class paths.
- // NOTE: the original collections may contain duplicates and may
not already be Set
- // collections with fast checks whether an item is contained
in it.
-
- // lazy construction of a new set for faster comparisons
- if (libraries.size() != requiredLibraries.size()
- || !new
HashSet<>(requiredLibraries).containsAll(libraries)) {
+ // Libraries are compared by BLOB content rather than by key, see
#libraryHashes.
+ final Set<String> requiredLibraryHashes =
hashesOf(requiredLibraries);
+ if (!libraryHashes.equals(requiredLibraryHashes)) {
throw new IllegalStateException(
- "The library registration references a different set
of library BLOBs than"
- + " previous registrations for this job:\nold:"
+ "The library registration references a different set
of library BLOB"
+ + " contents than previous registrations for
this job:\nold:"
+ libraries
- + "\nnew:"
- + requiredLibraries);
+ + " (contents: "
+ + libraryHashes
+ + ")\nnew:"
+ + requiredLibraries
+ + " (contents: "
+ + requiredLibraryHashes
+ + ")");
}
+ // NOTE: the original collection may contain duplicates and may
not already be a Set
+ // collection with fast checks whether an item is contained
in it.
// lazy construction of a new set with String representations of
the URLs
if (classPaths.size() != requiredClassPaths.size()
|| !requiredClassPaths.stream()
@@ -467,7 +488,7 @@ public class BlobLibraryCacheManager implements
LibraryCacheManager {
.containsAll(classPaths)) {
throw new IllegalStateException(
- "The library registration references a different set
of library BLOBs than"
+ "The library registration references a different set
of class paths than"
+ " previous registrations for this job:\nold:"
+ classPaths
+ "\nnew:"
@@ -475,6 +496,14 @@ public class BlobLibraryCacheManager implements
LibraryCacheManager {
}
}
+ private static Set<String> hashesOf(Collection<PermanentBlobKey>
libraries) {
+ final Set<String> hashes =
CollectionUtil.newHashSetWithExpectedSize(libraries.size());
+ for (PermanentBlobKey library : libraries) {
+ hashes.add(StringUtils.byteToHexString(library.getHash()));
+ }
+ return hashes;
+ }
+
/**
* Release the class loader to ensure any file descriptors are closed
and the cached
* libraries are deleted immediately.
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 e7f9e82f1a6..a0e7be8830d 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
@@ -67,6 +67,9 @@ public interface LibraryCacheManager {
* Every subsequent call to this method, will ensure that created user
code class loader can
* fulfill the required jar files and class paths.
*
+ * <p>Required jar files are matched by content, not by {@link
PermanentBlobKey} identity,
+ * so a jar that is re-uploaded without changing still resolves to the
cached class loader.
+ *
* @param requiredJarFiles requiredJarFiles the user code class loader
needs to load
* @param requiredClasspaths requiredClasspaths the user code class
loader needs to be
* started with
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 1047ce577a0..cf11611647a 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
@@ -56,8 +56,10 @@ import static
org.apache.flink.runtime.blob.TestingBlobHelpers.checkFileCountFor
import static org.apache.flink.runtime.blob.TestingBlobHelpers.checkFilesExist;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -615,6 +617,85 @@ public class BlobLibraryCacheManagerTest extends
TestLogger {
releaseHookLatch.await();
}
+ /**
+ * A {@link PermanentBlobKey} carries a random component on top of the
content hash, so
+ * re-uploading an unchanged JAR - as happens when a JobManager fails over
and resubmits the job
+ * - yields a key that is unequal to the one the cached class loader was
resolved with. Such a
+ * key must resolve to the cached class loader, while a key referring to
different content must
+ * still be rejected (FLINK-32212).
+ */
+ @Test
+ public void libraryBlobsAreComparedByContentNotByBlobKey() throws
Exception {
+ final JobID jobId = new JobID();
+ final byte[] content = new byte[] {1, 2, 3, 4, 5, 6, 7, 8};
+ final byte[] otherContent = new byte[] {8, 7, 6, 5, 4, 3, 2, 1};
+
+ BlobServer server = null;
+ PermanentBlobCache cache = null;
+ BlobLibraryCacheManager libCache = null;
+
+ try {
+ Configuration config = new Configuration();
+ config.set(BlobServerOptions.CLEANUP_INTERVAL, 1_000_000L);
+
+ server = new BlobServer(config, temporaryFolder.newFolder(), new
VoidBlobStore());
+ server.start();
+ InetSocketAddress serverAddress = new
InetSocketAddress("localhost", server.getPort());
+ cache =
+ new PermanentBlobCache(
+ config,
+ temporaryFolder.newFolder(),
+ new VoidBlobStore(),
+ serverAddress);
+
+ final PermanentBlobKey key = server.putPermanent(jobId, content);
+ final PermanentBlobKey reuploadedKey = server.putPermanent(jobId,
content);
+ final PermanentBlobKey otherKey = server.putPermanent(jobId,
otherContent);
+
+ assertNotEquals(key, reuploadedKey);
+ assertArrayEquals(key.getHash(), reuploadedKey.getHash());
+
+ libCache = createBlobLibraryCacheManager(cache);
+ cache.registerJob(jobId, applicationId);
+
+ final LibraryCacheManager.ClassLoaderLease lease =
+ libCache.registerClassLoaderLease(jobId, applicationId);
+ final UserCodeClassLoader classLoader =
+ lease.getOrResolveClassLoader(
+ Collections.singletonList(key),
Collections.emptyList());
+
+ assertThat(
+ lease.getOrResolveClassLoader(
+ Collections.singletonList(reuploadedKey),
Collections.emptyList()),
+ sameInstance(classLoader));
+
+ // a duplicate does not add content, so the required set is still
the same
+ assertThat(
+ lease.getOrResolveClassLoader(Arrays.asList(key, key),
Collections.emptyList()),
+ sameInstance(classLoader));
+
+ try {
+ lease.getOrResolveClassLoader(
+ Collections.singletonList(otherKey),
Collections.emptyList());
+ fail("Should fail with an IllegalStateException");
+ } catch (IllegalStateException e) {
+ // that's what we want
+ }
+
+ lease.release();
+ } finally {
+ if (libCache != null) {
+ libCache.shutdown();
+ }
+ if (cache != null) {
+ cache.close();
+ }
+ if (server != null) {
+ server.close();
+ }
+ }
+ }
+
private BlobLibraryCacheManager createSimpleBlobLibraryCacheManager()
throws IOException {
return new TestingBlobLibraryCacheManagerBuilder().build();
}
