Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/1895#discussion_r97902409
--- Diff: storm-core/src/jvm/org/apache/storm/zookeeper/Zookeeper.java ---
@@ -336,29 +340,53 @@ public static NimbusInfo toNimbusInfo(Participant
participant) {
public static LeaderLatchListener leaderLatchListenerImpl(final Map
conf, final CuratorFramework zk, final BlobStore blobStore, final LeaderLatch
leaderLatch) throws UnknownHostException {
final String hostName =
InetAddress.getLocalHost().getCanonicalHostName();
return new LeaderLatchListener() {
+ final String STORM_JAR_SUFFIX = "-stormjar.jar";
+ final String STORM_CODE_SUFFIX = "-stormcode.ser";
+ final String STORM_CONF_SUFFIX = "-stormconf.ser";
+
@Override
public void isLeader() {
- Set<String> activeTopologyIds = new
HashSet<>(Zookeeper.getChildren(zk, conf.get(Config.STORM_ZOOKEEPER_ROOT) +
ClusterUtils.STORMS_SUBTREE, false));
- Set<String> localTopologyIds =
blobStore.filterAndListKeys(new KeyFilter<String>() {
- @Override
- public String filter(String key) {
- return ConfigUtils.getIdFromBlobKey(key);
- }
- });
- Sets.SetView<String> diffTopology =
Sets.difference(activeTopologyIds, localTopologyIds);
- LOG.info("active-topology-ids [{}] local-topology-ids [{}]
diff-topology [{}]",
- generateJoinedString(activeTopologyIds),
generateJoinedString(localTopologyIds),
+ Set<String> activeTopologyIds = new
TreeSet<>(Zookeeper.getChildren(zk, conf.get(Config.STORM_ZOOKEEPER_ROOT) +
ClusterUtils.STORMS_SUBTREE, false));
+
+ Set<String> activeTopologyBlobKeys =
populateTopologyBlobKeys(activeTopologyIds);
+ Set<String> activeTopologyCodeKeys =
filterTopologyCodeKeys(activeTopologyBlobKeys);
+ Set<String> allLocalBlobKeys =
Sets.newHashSet(blobStore.listKeys());
+ Set<String> allLocalTopologyBlobKeys =
filterTopologyBlobKeys(allLocalBlobKeys);
+
+ // this finds all active topologies blob keys from all
local topology blob keys
+ Sets.SetView<String> diffTopology =
Sets.difference(activeTopologyBlobKeys, allLocalTopologyBlobKeys);
+ LOG.info("active-topology-blobs [{}] local-topology-blobs
[{}] diff-topology-blobs [{}]",
+ generateJoinedString(activeTopologyIds),
generateJoinedString(allLocalTopologyBlobKeys),
generateJoinedString(diffTopology));
- if (diffTopology.isEmpty()) {
- LOG.info("Accepting leadership, all active topology
found locally.");
- } else {
+ if (!diffTopology.isEmpty()) {
LOG.info("code for all active topologies not available
locally, giving up leadership.");
try {
leaderLatch.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
+
+ return;
--- End diff --
OK I'll try to remove `return` at this place.
Btw, synchronizing blobs are done for each, so I don't see benefit to clean
up. If we're worried about inconsistency between versions of the topology
blobs, I think it should be handled properly from BlobStore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
