Reject a topology submission that lists a blob key which is not a dependency blob key

[rzo1]

StormTopology.dependency_jars and dependency_artifacts are filled in by the client and nothing on the submit path checks them. Not validateTopologyName, ConfigValidation.validateTopoConf, DefaultTopologyValidator.validate, validateTopologyWorkerMaxHeapSizeConfigs, Utils.validateTopologyBlobStoreMap, validateTopologySize, StormCommon.validateStructure or normalizeTopology. Nimbus then acts on whatever they contain: it deletes those keys during cleanup, and on gaining leadership it compares them against the blobstore.

LeaderListenerCallback computes activeTopologyDependencies - blobStore.listKeys() and surrenders leadership when the difference is not empty. One dependency key that does not exist, on one active topology, makes every Nimbus win leadership, surrender and re-queue indefinitely, so the cluster does no scheduling, no cleanup and accepts no submissions.

Submission now requires every entry to carry the dep- prefix and to exist in the blobstore. Existence is checked with blobStore.getBlobMeta(key, subject) as the submitting subject, the same way Utils.validateTopologyBlobStoreMap does twenty lines below. A key listed twice is looked up once.

StormSubmitter uploads the blobs, then sets the fields, then submits, and DependencyUploader returns a key only after the blob is committed, so a legitimate submission always finds its blobs present. On a replicated blobstore getBlobMeta falls through to checkForBlobOrDownload, so the lookup also pulls a blob that is missing locally.

BLOB_DEPENDENCIES_PREFIX becomes public and gains an isDependencyBlobKey predicate so the producer and the check use one definition.

[reiabreu]

Disclaimer: this comment was generated with the help of an LLM.

Minor nit: NimbusTest.java adds a duplicate import javax.security.auth.Subject; — it's already imported a few lines above (from the earlier rebalance change). Checkstyle's RedundantImport will likely flag it, so it's worth dropping the extra line.