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.

Author: rzo1

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<Integer> sequenceNumbers = new TreeSet<Integer>();
         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,

storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java

@@ -139,14 +139,26 @@ 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);
             }
         }
     }
 
+    /**
+     * 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);

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -777,8 +777,18 @@ static List<String> getKeyListFromId(Map<String, Object> 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) {

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;
     }
 }

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> 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> 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());