You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2018/05/04 18:43:17 UTC
[4/6] storm git commit: STORM-2171: Fixing nimbus setup
STORM-2171: Fixing nimbus setup
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/692cc4e2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/692cc4e2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/692cc4e2
Branch: refs/heads/master
Commit: 692cc4e28ae3e13094e425bdf99d088ee0fa37f2
Parents: 82e7cf8
Author: Govind Menon <go...@gmail.com>
Authored: Thu May 3 09:24:26 2018 -0500
Committer: Govind Menon <go...@gmail.com>
Committed: Thu May 3 09:28:01 2018 -0500
----------------------------------------------------------------------
.../storm/hdfs/blobstore/HdfsBlobStore.java | 10 ++++++++++
.../org/apache/storm/blobstore/BlobStore.java | 6 ++++++
.../storm/blobstore/LocalFsBlobStore.java | 10 ++++++++++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 21 ++++++--------------
4 files changed, 32 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/692cc4e2/external/storm-hdfs-blobstore/src/main/java/org/apache/storm/hdfs/blobstore/HdfsBlobStore.java
----------------------------------------------------------------------
diff --git a/external/storm-hdfs-blobstore/src/main/java/org/apache/storm/hdfs/blobstore/HdfsBlobStore.java b/external/storm-hdfs-blobstore/src/main/java/org/apache/storm/hdfs/blobstore/HdfsBlobStore.java
index 75a65de..4872581 100644
--- a/external/storm-hdfs-blobstore/src/main/java/org/apache/storm/hdfs/blobstore/HdfsBlobStore.java
+++ b/external/storm-hdfs-blobstore/src/main/java/org/apache/storm/hdfs/blobstore/HdfsBlobStore.java
@@ -279,6 +279,16 @@ public class HdfsBlobStore extends BlobStore {
return rbm;
}
+ /**
+ * Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi
+ *
+ * @param leaderElector
+ */
+ @Override
+ public void setLeaderElector(ILeaderElector leaderElector) {
+ // NO-OP
+ }
+
@Override
public void setBlobMeta(String key, SettableBlobMeta meta, Subject who)
throws AuthorizationException, KeyNotFoundException {
http://git-wip-us.apache.org/repos/asf/storm/blob/692cc4e2/storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java b/storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java
index 5b8659b..3f3dfde 100644
--- a/storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java
+++ b/storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java
@@ -124,6 +124,12 @@ public abstract class BlobStore implements Shutdownable {
public abstract ReadableBlobMeta getBlobMeta(String key, Subject who) throws AuthorizationException, KeyNotFoundException;
/**
+ * Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi
+ * @param leaderElector
+ */
+
+ public abstract void setLeaderElector(ILeaderElector leaderElector);
+ /**
* Sets the metadata with renewed acls for the blob.
*
* @param key Key for the blob.
http://git-wip-us.apache.org/repos/asf/storm/blob/692cc4e2/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
index ca281f3..fe0748c 100644
--- a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
+++ b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
@@ -298,6 +298,16 @@ public class LocalFsBlobStore extends BlobStore {
return rbm;
}
+ /**
+ * Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi
+ *
+ * @param leaderElector
+ */
+ @Override
+ public void setLeaderElector(ILeaderElector leaderElector) {
+ this.leaderElector = leaderElector;
+ }
+
@Override
public void setBlobMeta(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyNotFoundException {
validateKey(key);
http://git-wip-us.apache.org/repos/asf/storm/blob/692cc4e2/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
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 2d6b825..6768102 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
@@ -280,12 +280,6 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
state.removeStorm(topoId);
notifySupervisorsAsKilled(state, oldAssignment, nimbus.getAssignmentsDistributer());
BlobStore store = nimbus.getBlobStore();
- if (store instanceof LocalFsBlobStore) {
- for (String key : Nimbus.getKeyListFromId(nimbus.getConf(), topoId)) {
- state.removeBlobstoreKey(key);
- state.removeKeyVersion(key);
- }
- }
nimbus.getHeartbeatsCache().getAndUpdate(new Dissoc<>(topoId));
return null;
};
@@ -480,7 +474,6 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
this.heartbeatsRecoveryStrategy = WorkerHeartbeatsRecoveryStrategyFactory.getStrategy(conf);
this.downloaders = fileCacheMap(conf);
this.uploaders = fileCacheMap(conf);
- this.topoCache = topoCache;
this.blobDownloaders = makeBlobCacheMap(conf);
this.blobUploaders = makeBlobCacheMap(conf);
this.blobListers = makeBlobListCacheMap(conf);
@@ -495,21 +488,19 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
this.scheduler = wrapAsBlacklistScheduler(conf, underlyingScheduler);
this.zkClient = makeZKClient(conf);
+ if (blobStore == null) {
+ blobStore = ServerUtils.getNimbusBlobStore(conf, this.nimbusHostPortInfo, null);
+ }
+ this.blobStore = blobStore;
if (leaderElector == null) {
leaderElector = Zookeeper.zkLeaderElector(conf, zkClient, blobStore, topoCache, stormClusterState, getNimbusAcls(conf));
}
this.leaderElector = leaderElector;
- if (blobStore == null) {
- blobStore = ServerUtils.getNimbusBlobStore(conf, this.nimbusHostPortInfo, this.leaderElector);
- }
- this.blobStore = blobStore;
+ this.blobStore.setLeaderElector(this.leaderElector);
if (topoCache == null) {
topoCache = new TopoCache(blobStore, conf);
}
-
- if (leaderElector == null) {
- leaderElector = Zookeeper.zkLeaderElector(conf, zkClient, blobStore, topoCache, stormClusterState, getNimbusAcls(conf));
- }
+ this.topoCache = topoCache;
this.assignmentsDistributer = AssignmentDistributionService.getInstance(conf);
this.idToSchedStatus = new AtomicReference<>(new HashMap<>());
this.nodeIdToResources = new AtomicReference<>(new HashMap<>());