From a9fb80480d60210d4a6e4503b5b6d3296ca98380 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Thu, 10 Sep 2026 11:16:26 +0200 Subject: [PATCH 1/2] STORM-3871: sweep dependency blobs that outlive their topology's cleanup A dependency blob key carries no topology id, so a blob that survives the cleanup pass of its topology (a failed submission, or a blob that another nimbus hands back to the leader after it was deleted) stays in the blob store forever. doCleanup now also removes dependency blobs whose key is provably unique to one topology once no topology has referred to them for nimbus.inbox.jar.expiration.secs. --- .../java/org/apache/storm/DaemonConfig.java | 4 + .../apache/storm/daemon/nimbus/Nimbus.java | 47 ++++++++ .../blobstore/KeySequenceNumberTest.java | 80 +++++++++++++ .../storm/daemon/nimbus/NimbusTest.java | 109 ++++++++++++++++++ 4 files changed, 240 insertions(+) create mode 100644 storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java index b62702abc5a..4805f51957b 100644 --- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java +++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java @@ -270,6 +270,10 @@ public class DaemonConfig implements Validated { * it takes to delete an inbox jar file is going to be somewhat more than NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS * (depending on how often NIMBUS_CLEANUP_FREQ_SECS is set to). * + *

This is also how long a dependency blob uploaded by a client may go without any topology referring to it + * before nimbus deletes it from the blob store, which covers the time between uploading the dependencies of a + * topology and submitting it. + * * @see #NIMBUS_CLEANUP_INBOX_FREQ_SECS */ @IsInteger diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java index cc08fe2201f..72f4a39d26d 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java @@ -440,6 +440,8 @@ public static List getNimbusAcls(Map conf) { private final TimeCacheMap uploaders; private final BlobStore blobStore; private final TopoCache topoCache; + //When a dependency blob was first seen unreferenced by any topology, only used by the cleanup pass. + private final Map orphanedDependencyKeysDetectedMs = new HashMap<>(); @SuppressWarnings("deprecation") private final TimeCacheMap blobDownloaders; @SuppressWarnings("deprecation") @@ -3095,6 +3097,45 @@ public void forceDeleteTopoDistDir(String topoId) throws IOException { Utils.forceDelete(ServerConfigUtils.masterStormDistRoot(conf, topoId)); } + /** + * Remove the dependency blobs that no topology has referred to for longer than the inbox jar expiration. + * + *

{@link #rmDependencyBlobsInTopology} only runs in the pass that cleans up the owning topology, and a dependency + * blob key carries no topology id, so a blob that outlives that pass can never be traced back to it. That happens + * when a submission fails after its dependencies were uploaded, or when another nimbus still holds a copy of the blob + * and it is downloaded back after it was removed here. This sweep reclaims those. + * + *

Only keys that are provably unique to one topology are considered: an older client that finds a shareable key + * in the store does not upload it again but refers to it, so such a key may be about to be used. A client uploads the + * dependencies before it submits the topology, so a key is only removed once it has been seen unreferenced for + * {@link DaemonConfig#NIMBUS_INBOX_JAR_EXPIRATION_SECS}, the time nimbus gives an uploaded topology jar to be + * submitted. When a key was first seen unreferenced is only kept in memory, so a new leader starts the wait over. + * + * @param referenced the dependency blob keys referenced by topologies that are not being cleaned up + */ + @VisibleForTesting + void sweepOrphanedDependencyBlobs(Set referenced) { + try { + long graceMs = TimeUnit.SECONDS.toMillis( + ObjectReader.getInt(conf.get(DaemonConfig.NIMBUS_INBOX_JAR_EXPIRATION_SECS), 3600)); + long nowMs = Time.currentTimeMillis(); + Set orphaned = blobStore.filterAndListKeys( + key -> isProvablyUniqueDependencyKey(key) && !referenced.contains(key) ? key : null); + //Forget the keys that are gone or referenced again, so that being orphaned later waits the full time again. + orphanedDependencyKeysDetectedMs.keySet().retainAll(orphaned); + for (String key : orphaned) { + long unreferencedMs = nowMs - orphanedDependencyKeysDetectedMs.computeIfAbsent(key, k -> nowMs); + if (unreferencedMs >= graceMs) { + LOG.info("Removing dependency blob {}, no topology has referred to it for {} ms", key, unreferencedMs); + rmBlobKey(blobStore, key, stormClusterState); + orphanedDependencyKeysDetectedMs.remove(key); + } + } + } catch (Exception e) { + LOG.warn("Could not sweep the dependency blobs that no topology refers to", e); + } + } + /** * Cleanup topologies and Jars. */ @@ -3134,6 +3175,12 @@ public void doCleanup() { idToExecutors.getAndUpdate(new Dissoc<>(topoId)); } + //Catches the dependency blobs that outlived the pass that cleaned up their topology. Without knowing the + //references nothing can be told to be unused, so nothing is swept then. + if (stillReferenced != null) { + sweepOrphanedDependencyBlobs(stillReferenced); + } + long cleanupDurationMs = Time.deltaMs(cleanupStartMs); if (cleanupDurationMs > 10000) { LOG.warn("doCleanup is taking too long, topoIdSelectionDurationMs={}, cleanupDurationMs={}", diff --git a/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java new file mode 100644 index 00000000000..bb2ccde65aa --- /dev/null +++ b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.blobstore; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.apache.storm.nimbus.NimbusInfo; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.storm.shade.org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.storm.testing.InProcessZookeeper; +import org.junit.jupiter.api.Test; + +class KeySequenceNumberTest { + private static final String KEY = "dep-lib-11111111-1111-1111-1111-111111111111.jar"; + private static final NimbusInfo LEADER = new NimbusInfo("nimbus-1", 6627, false); + private static final NimbusInfo PEER = new NimbusInfo("nimbus-0", 6627, false); + + /** + * Replays the blob store state changes behind the nimbus log reported on STORM-3871, where a dependency blob that + * was removed when its topology was cleaned up showed up on the leader again, registered at version 0 and then 1. + */ + @Test + void aBlobThatAnotherNimbusRegistersAgainAfterItWasDeletedIsRecreatedOnTheLeader() throws Exception { + try (InProcessZookeeper zk = new InProcessZookeeper(); + CuratorFramework zkClient = CuratorFrameworkFactory.newClient("localhost:" + zk.getPort(), + new ExponentialBackoffRetry(1000, 3))) { + zkClient.start(); + + // the client uploads the dependency: createBlob, then createStateInZookeeper when it closes the stream + assertEquals(1, register(zkClient, LEADER)); + assertEquals(2, register(zkClient, LEADER)); + + // the topology is cleaned up and the leader deletes the blob, as LocalFsBlobStore#deleteBlob does + zkClient.delete().deletingChildrenIfNeeded().forPath("/blobstore/" + KEY); + zkClient.delete().deletingChildrenIfNeeded().forPath("/blobstoremaxkeysequencenumber/" + KEY); + + // another nimbus, which still has a copy, registers the key again as if it were new + assertEquals(1, register(zkClient, PEER)); + + // a request for the key makes the leader download it back from that nimbus: createBlob, then + // createStateInZookeeper, which are the set-path lines ending in -0 and -1 in the report + assertEquals(0, register(zkClient, LEADER)); + assertEquals(1, register(zkClient, LEADER)); + } + } + + /** + * Do what IStormClusterState#setupBlob does with the version KeySequenceNumber hands out. + */ + private static int register(CuratorFramework zkClient, NimbusInfo nimbus) throws Exception { + int version = new KeySequenceNumber(KEY, nimbus).getKeySequenceNumber(zkClient); + String parent = "/blobstore/" + KEY; + if (zkClient.checkExists().forPath(parent) != null) { + for (String child : zkClient.getChildren().forPath(parent)) { + if (child.startsWith(nimbus.toHostPortString())) { + zkClient.delete().forPath(parent + "/" + child); + } + } + } + zkClient.create().creatingParentsIfNeeded().forPath(parent + "/" + nimbus.toHostPortString() + "-" + version); + return version; + } +} diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java index e27b5b4ec2d..dd9b3011c9b 100644 --- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java +++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java @@ -39,6 +39,7 @@ import org.apache.storm.DaemonConfig; import org.apache.storm.blobstore.BlobStore; import org.apache.storm.blobstore.BlobStoreAclHandler; +import org.apache.storm.blobstore.KeyFilter; import org.apache.storm.blobstore.KeySequenceNumber; import org.apache.storm.blobstore.LocalFsBlobStore; import org.apache.storm.cluster.IStormClusterState; @@ -575,6 +576,114 @@ void doCleanupContinuesTheReferenceScanWhenACandidateTopologyHasNoCodeBlob() thr verify(store, never()).deleteBlob(eq(LEGACY_ARTIFACT_KEY), any()); } + /** + * Make the blob store mock list exactly these keys. + */ + private static void storeKeys(BlobStore store, String... keys) { + when(store.filterAndListKeys(any())).thenAnswer(invocation -> { + KeyFilter filter = invocation.getArgument(0); + Set filtered = new HashSet<>(); + for (String key : keys) { + Object result = filter.filter(key); + if (result != null) { + filtered.add(result); + } + } + return filtered; + }); + } + + @Test + void doCleanupSweepsADependencyBlobNoTopologyRefersToOnceTheInboxJarExpirationHasPassed() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of()); + when(state.activeStorms()).thenReturn(List.of()); + // outlived the pass that cleaned up its topology, e.g. because it was downloaded back from another nimbus + storeKeys(store, UNIQUE_JAR_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + // one second short of the default nimbus.inbox.jar.expiration.secs, a submission may still be under way + Time.advanceTimeSecs(3599); + nimbus.doCleanup(); + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + + Time.advanceTimeSecs(1); + nimbus.doCleanup(); + verify(store).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + + @Test + void doCleanupNeverSweepsADependencyBlobThatIsReferencedOrCouldBeShared() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("live-topo")); + when(state.activeStorms()).thenReturn(List.of("live-topo")); + storeTopology(store, "live-topo", List.of(UNIQUE_JAR_KEY), List.of()); + // nothing refers to the legacy key, but an older client that finds it in the store refers to it without + // uploading it again, so it may be about to be used + storeKeys(store, UNIQUE_JAR_KEY, LEGACY_ARTIFACT_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + verify(store, never()).deleteBlob(eq(LEGACY_ARTIFACT_KEY), any()); + } + } + + @Test + void doCleanupDoesNotSweepDependencyBlobsWhenTheReferencesCannotBeRead() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("live-topo")); + when(state.activeStorms()).thenReturn(List.of("live-topo")); + // the live topology's code blob cannot be read, so it may be the one referring to the key + when(store.readBlob(eq(ConfigUtils.masterStormCodeKey("live-topo")), any())) + .thenThrow(new IOException("blob store is unhappy")); + storeKeys(store, UNIQUE_JAR_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + + @Test + void aDependencyBlobThatComesBackAfterItsTopologyWasCleanedUpIsRemovedAgain() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("dead-topo")); + when(state.activeStorms()).thenReturn(List.of()); + storeTopology(store, "dead-topo", List.of(UNIQUE_JAR_KEY), List.of()); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + verify(store).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + + // the topology is gone for good, but another nimbus still had a copy of the blob and it was downloaded + // back, so the pass that knew which topology it belonged to is over + when(store.storedTopoIds()).thenReturn(Set.of()); + storeKeys(store, UNIQUE_JAR_KEY); + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, times(2)).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + @Test void everyDependencyKeyACurrentClientGeneratesIsRecognisedAsUniqueToOneTopology() { for (String fileName : List.of("commons-lang3-3.12.0.jar", "some.lib.tar.gz", "noextension")) { From 8c1fdc19dd734399fd61734d2c8d74ca39838437 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Thu, 10 Sep 2026 13:09:18 +0200 Subject: [PATCH 2/2] STORM-3871: stop a non-leader nimbus from registering a deleted blob again The leader deletes a blob's zookeeper nodes one by one. A non-leader that syncs in between downloads the blob from another nimbus and then registers its copy after the key is gone, which registers the key as new, and the other nimbus instances, the leader included, download the blob back from it. Only the leader storing a blob a client uploads may now register a key zookeeper does not know; any other nimbus gets a KeyNotFoundException instead. --- .../storm/blobstore/KeySequenceNumber.java | 17 +++++ .../storm/blobstore/LocalFsBlobStore.java | 19 +++++- .../apache/storm/daemon/nimbus/Nimbus.java | 16 ++++- .../blobstore/KeySequenceNumberTest.java | 66 ++++++++++++------- .../storm/daemon/nimbus/NimbusTest.java | 19 +++++- 5 files changed, 106 insertions(+), 31 deletions(-) diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java b/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java index 39e1747259c..ae0f998603b 100644 --- a/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java +++ b/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java @@ -124,10 +124,27 @@ public KeySequenceNumber(String key, NimbusInfo nimbusInfo) { } public synchronized int getKeySequenceNumber(CuratorFramework zkClient) throws KeyNotFoundException { + return getKeySequenceNumber(zkClient, true); + } + + /** + * Hand over the sequence number for the copy of the key held by this nimbus. + * + * @param zkClient the zookeeper client + * @param mayCreateKey whether the key may be created when zookeeper does not know it, which is only right for the + * leader storing a blob that a client uploads. Any other nimbus mirrors a key the leader created, + * so a key zookeeper does not know was deleted and must not be registered again. + * @return the sequence number + * @throws KeyNotFoundException if the key is not in zookeeper and may not be created, or it is deleted meanwhile + */ + public synchronized int getKeySequenceNumber(CuratorFramework zkClient, boolean mayCreateKey) throws KeyNotFoundException { TreeSet sequenceNumbers = new TreeSet(); try { // Key has not been created yet and it is the first time it is being created if (zkClient.checkExists().forPath(BlobStoreUtils.getBlobStoreSubtree() + "/" + key) == null) { + if (!mayCreateKey) { + throw new KeeperException.NoNodeException(BlobStoreUtils.getBlobStoreSubtree() + "/" + key); + } zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT) .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(BLOBSTORE_MAX_KEY_SEQUENCE_SUBTREE + "/" + key); zkClient.setData().forPath(BLOBSTORE_MAX_KEY_SEQUENCE_SUBTREE + "/" + key, diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java index 5ec0b2357a4..a5d5c5bcf8e 100644 --- a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java +++ b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java @@ -139,7 +139,7 @@ private void setupBlobstore() throws AuthorizationException, KeyNotFoundExceptio LOG.debug("Creating list of key entries for blobstore inside zookeeper {} local {}", activeKeys, activeLocalKeys); for (String key : activeLocalKeys) { try { - state.setupBlob(key, nimbusInfo, getVersionForKey(key, nimbusInfo, zkClient)); + state.setupBlob(key, nimbusInfo, getVersionForKey(key, nimbusInfo, zkClient, false)); } catch (KeyNotFoundException e) { // invalid key, remove it from blobstore store.deleteBlob(key, NIMBUS_SUBJECT); @@ -147,6 +147,18 @@ private void setupBlobstore() throws AuthorizationException, KeyNotFoundExceptio } } + /** + * Tell whether this nimbus is the leader, which is the only one that may register a key zookeeper does not know. + * Without a leader elector there is a single nimbus, which is then the leader. + */ + private boolean isLeader() { + try { + return leaderElector == null || leaderElector.isLeader(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + private void blobSync() throws Exception { if ("distributed".equals(conf.get(Config.STORM_CLUSTER_MODE))) { @@ -216,11 +228,14 @@ public AtomicOutputStream createBlob(String key, SettableBlobMeta meta, Subject } BlobStoreFileOutputStream outputStream = null; try { + //Taken before anything is written, so that a non-leader downloading a key that was deleted meanwhile is + //refused without leaving a copy behind. + int version = getVersionForKey(key, this.nimbusInfo, zkClient, isLeader()); outputStream = new BlobStoreFileOutputStream(fbs.write(META_PREFIX + key, true)); outputStream.write(Utils.thriftSerialize(meta)); outputStream.close(); outputStream = null; - this.stormClusterState.setupBlob(key, this.nimbusInfo, getVersionForKey(key, this.nimbusInfo, zkClient)); + this.stormClusterState.setupBlob(key, this.nimbusInfo, version); return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX + key, true)); } catch (IOException e) { throw new RuntimeException(e); diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java index 72f4a39d26d..75ef22c6a05 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java @@ -777,8 +777,18 @@ static List getKeyListFromId(Map conf, String id) { public static int getVersionForKey(String key, NimbusInfo nimbusInfo, CuratorFramework zkClient) throws KeyNotFoundException { + return getVersionForKey(key, nimbusInfo, zkClient, true); + } + + /** + * Get the version to register the copy of a blob held by a nimbus under. + * + * @see KeySequenceNumber#getKeySequenceNumber(CuratorFramework, boolean) + */ + public static int getVersionForKey(String key, NimbusInfo nimbusInfo, + CuratorFramework zkClient, boolean mayCreateKey) throws KeyNotFoundException { KeySequenceNumber kseq = new KeySequenceNumber(key, nimbusInfo); - return kseq.getKeySequenceNumber(zkClient); + return kseq.getKeySequenceNumber(zkClient, mayCreateKey); } private static StormTopology readStormTopology(String topoId, TopoCache tc) throws KeyNotFoundException, AuthorizationException, @@ -4425,7 +4435,9 @@ public void createStateInZookeeper(String key) throws TException { BlobStore store = blobStore; NimbusInfo ni = nimbusHostPortInfo; if (store instanceof LocalFsBlobStore) { - state.setupBlob(key, ni, getVersionForKey(key, ni, zkClient)); + //A non-leader only registers its copy of a key the leader created. If zookeeper does not know the key + //any more it was deleted while the copy was downloaded, and registering it would bring it back. + state.setupBlob(key, ni, getVersionForKey(key, ni, zkClient, isLeader())); } LOG.debug("Created state in zookeeper {} {} {}", state, store, ni); } catch (KeyNotFoundException e) { diff --git a/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java index bb2ccde65aa..8b4e8a08ba5 100644 --- a/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java +++ b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java @@ -19,7 +19,10 @@ package org.apache.storm.blobstore; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import org.apache.storm.generated.KeyNotFoundException; import org.apache.storm.nimbus.NimbusInfo; import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; import org.apache.storm.shade.org.apache.curator.framework.CuratorFrameworkFactory; @@ -29,52 +32,65 @@ class KeySequenceNumberTest { private static final String KEY = "dep-lib-11111111-1111-1111-1111-111111111111.jar"; + private static final String KEY_PATH = "/blobstore/" + KEY; + private static final String MAX_SEQUENCE_PATH = "/blobstoremaxkeysequencenumber/" + KEY; private static final NimbusInfo LEADER = new NimbusInfo("nimbus-1", 6627, false); - private static final NimbusInfo PEER = new NimbusInfo("nimbus-0", 6627, false); + private static final NimbusInfo PEER = new NimbusInfo("nimbus-2", 6627, false); /** - * Replays the blob store state changes behind the nimbus log reported on STORM-3871, where a dependency blob that - * was removed when its topology was cleaned up showed up on the leader again, registered at version 0 and then 1. + * Replays the blob store state changes behind the nimbus logs reported on STORM-3871. A non-leader that downloaded + * a blob while the leader deleted it registered the key again as new, and every other nimbus, the leader included, + * then downloaded the blob back from it. */ @Test - void aBlobThatAnotherNimbusRegistersAgainAfterItWasDeletedIsRecreatedOnTheLeader() throws Exception { + void aNonLeaderCannotRegisterAKeyAgainThatWasDeletedWhileItDownloadedIt() throws Exception { try (InProcessZookeeper zk = new InProcessZookeeper(); - CuratorFramework zkClient = CuratorFrameworkFactory.newClient("localhost:" + zk.getPort(), - new ExponentialBackoffRetry(1000, 3))) { - zkClient.start(); - + CuratorFramework zkClient = newClient(zk)) { // the client uploads the dependency: createBlob, then createStateInZookeeper when it closes the stream - assertEquals(1, register(zkClient, LEADER)); - assertEquals(2, register(zkClient, LEADER)); + assertEquals(1, register(zkClient, LEADER, true)); + assertEquals(2, register(zkClient, LEADER, true)); // the topology is cleaned up and the leader deletes the blob, as LocalFsBlobStore#deleteBlob does - zkClient.delete().deletingChildrenIfNeeded().forPath("/blobstore/" + KEY); - zkClient.delete().deletingChildrenIfNeeded().forPath("/blobstoremaxkeysequencenumber/" + KEY); + zkClient.delete().deletingChildrenIfNeeded().forPath(KEY_PATH); + zkClient.delete().deletingChildrenIfNeeded().forPath(MAX_SEQUENCE_PATH); + + // the non-leader finishes its download and would register the key as if it were new + assertThrows(KeyNotFoundException.class, () -> register(zkClient, PEER, false)); + assertNull(zkClient.checkExists().forPath(KEY_PATH)); + assertNull(zkClient.checkExists().forPath(MAX_SEQUENCE_PATH)); + } + } - // another nimbus, which still has a copy, registers the key again as if it were new - assertEquals(1, register(zkClient, PEER)); + @Test + void aNonLeaderStillRegistersItsCopyOfAKeyTheLeaderCreated() throws Exception { + try (InProcessZookeeper zk = new InProcessZookeeper(); + CuratorFramework zkClient = newClient(zk)) { + assertEquals(1, register(zkClient, LEADER, true)); - // a request for the key makes the leader download it back from that nimbus: createBlob, then - // createStateInZookeeper, which are the set-path lines ending in -0 and -1 in the report - assertEquals(0, register(zkClient, LEADER)); - assertEquals(1, register(zkClient, LEADER)); + assertEquals(0, register(zkClient, PEER, false)); } } + private static CuratorFramework newClient(InProcessZookeeper zk) { + CuratorFramework zkClient = CuratorFrameworkFactory.newClient("localhost:" + zk.getPort(), + new ExponentialBackoffRetry(1000, 3)); + zkClient.start(); + return zkClient; + } + /** * Do what IStormClusterState#setupBlob does with the version KeySequenceNumber hands out. */ - private static int register(CuratorFramework zkClient, NimbusInfo nimbus) throws Exception { - int version = new KeySequenceNumber(KEY, nimbus).getKeySequenceNumber(zkClient); - String parent = "/blobstore/" + KEY; - if (zkClient.checkExists().forPath(parent) != null) { - for (String child : zkClient.getChildren().forPath(parent)) { + private static int register(CuratorFramework zkClient, NimbusInfo nimbus, boolean mayCreateKey) throws Exception { + int version = new KeySequenceNumber(KEY, nimbus).getKeySequenceNumber(zkClient, mayCreateKey); + if (zkClient.checkExists().forPath(KEY_PATH) != null) { + for (String child : zkClient.getChildren().forPath(KEY_PATH)) { if (child.startsWith(nimbus.toHostPortString())) { - zkClient.delete().forPath(parent + "/" + child); + zkClient.delete().forPath(KEY_PATH + "/" + child); } } } - zkClient.create().creatingParentsIfNeeded().forPath(parent + "/" + nimbus.toHostPortString() + "-" + version); + zkClient.create().creatingParentsIfNeeded().forPath(KEY_PATH + "/" + nimbus.toHostPortString() + "-" + version); return version; } } diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java index dd9b3011c9b..0b5215c6f14 100644 --- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java +++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java @@ -94,6 +94,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; @@ -266,14 +267,28 @@ void testCreateStateInZookeeperWhenFailToSetupBlobWithRuntimeExceptionThrowsRunt @Test void testCreateStateInZookeeperWhenKeyNotFoundHandlesException() throws Exception { try (MockedConstruction keySequenceNumber = mockConstruction(KeySequenceNumber.class, (mock, context) -> - when(mock.getKeySequenceNumber(any())).thenThrow(new KeyNotFoundException("Failed to setup blob")))) { + when(mock.getKeySequenceNumber(any(), anyBoolean())).thenThrow(new KeyNotFoundException("Failed to setup blob")))) { nimbus.createStateInZookeeper(BLOB_FILE_KEY); - verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any()); + verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any(), anyBoolean()); verify(stormClusterState, never()).setupBlob(eq(BLOB_FILE_KEY), eq(nimbusInfo), any()); } } + @Test + void testCreateStateInZookeeperOnlyLetsTheLeaderRegisterAKeyZookeeperDoesNotKnow() throws Exception { + try (MockedConstruction keySequenceNumber = mockConstruction(KeySequenceNumber.class)) { + when(leaderElector.isLeader()).thenReturn(false); + nimbus.createStateInZookeeper(BLOB_FILE_KEY); + when(leaderElector.isLeader()).thenReturn(true); + nimbus.createStateInZookeeper(BLOB_FILE_KEY); + + // a non-leader registering a key zookeeper does not know would bring back a key that was deleted + verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any(), eq(false)); + verify(keySequenceNumber.constructed().get(1)).getKeySequenceNumber(any(), eq(true)); + } + } + @Test void testListBlobsOnlyReturnsKeysTheCallerMayReadTheMetadataOf() throws Exception { when(localBlobStore.listKeys()).thenReturn(List.of("readable-key", "other-users-key").iterator());