code: public static void cleanup_key(String blobKey, BlobStore blobStore, StormClusterState clusterState) {
          // we skip to clean up the non-exist keys
          if (blobKey.startsWith(JStormMetrics.NIMBUS_METRIC_KEY)
                  || blobKey.startsWith(JStormMetrics.CLUSTER_METRIC_KEY)
                  || blobKey.startsWith(JStormMetrics.SUPERVISOR_METRIC_KEY)) {
              return;
          }
          try {
              blobStore.deleteBlob(blobKey);
          } catch (Exception e) {
              LOG.warn("cleanup blob key {} error {}", blobKey, e);
          }
  
          try {
              if (blobStore instanceof LocalFsBlobStore) {
                  clusterState.remove_blobstore_key(blobKey);
                  clusterState.remove_key_version(blobKey);
              }
          } catch (Exception e) {
              LOG.warn("cleanup blob key {} error {}", blobKey, e);
          }
      }
docs: remove blob information in zk for the blobkey