[storm] branch master updated: STORM-3804 Don't allow deleting blobs if they are required for an active topology (#3421)

[ag...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

agresch pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git


The following commit(s) were added to refs/heads/master by this push:
     new 6c3a6be  STORM-3804 Don't allow deleting blobs if they are required for an active topology (#3421)
6c3a6be is described below

commit 6c3a6be54514ebba6f7e0fbd0c9b89a9192e9eff
Author: snikhil5 <ns...@yahoo.com>
AuthorDate: Wed Nov 3 15:54:44 2021 -0500

    STORM-3804 Don't allow deleting blobs if they are required for an active topology (#3421)
    
    STORM-3804 Don't allow deleting blobs if they are required for an active topology
---
 .../org/apache/storm/daemon/nimbus/Nimbus.java     | 33 ++++++++++++++++++++++
 1 file changed, 33 insertions(+)

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 ce55a85..f95c5ff 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
@@ -985,6 +985,33 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
         return isTopologyActive(state, topoName) || state.activeStorms().contains(topoName);
     }
 
+    /**
+     * Returns the topologyId of the topology using this blob.
+     * @param state the cluster state
+     * @param topoCache the topology cache
+     * @param key the blob key
+     * @return null or id
+     */
+    private static String topologyUsingThisBlob(IStormClusterState state, TopoCache topoCache, String key) {
+        for (String topologyId : state.activeStorms()) {
+            Map<String, Object> topoConf = null;
+            try {
+                topoConf = readTopoConfAsNimbus(topologyId, topoCache);
+            } catch (KeyNotFoundException | AuthorizationException | IOException e) {
+                continue;
+            }
+            if (topoConf == null) {
+                continue;
+            }
+            @SuppressWarnings("unchecked")
+            Map<String, Map<String, Object>> blobstoreMap = (Map<String, Map<String, Object>>) topoConf.get(Config.TOPOLOGY_BLOBSTORE_MAP);
+            if (blobstoreMap != null && blobstoreMap.containsKey(key)) {
+                return topologyId;
+            }
+        }
+        return null;
+    }
+
     private static Map<String, Object> tryReadTopoConf(String topoId, TopoCache tc)
         throws NotAliveException, AuthorizationException, IOException {
         try {
@@ -3907,6 +3934,12 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
                     throw new WrappedIllegalStateException(message);
                 }
             }
+            String topoId = topologyUsingThisBlob(stormClusterState, topoCache,  key);
+            if (topoId != null) {
+                String message = "Attempting to delete active blob " + key + " used by topology " + topoId;
+                LOG.warn(message);
+                throw new WrappedIllegalStateException(message);
+            }
             blobStore.deleteBlob(key, getSubject());
             LOG.info("Deleted blob for key {} with {}", key, ReqContext.context());
         } catch (Exception e) {