You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2017/08/30 21:07:54 UTC
[31/50] [abbrv] storm git commit: STORM-2682: Fix issues with null
owner on rolling upgrade
STORM-2682: Fix issues with null owner on rolling upgrade
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f10e26da
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f10e26da
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f10e26da
Branch: refs/heads/metrics_v2
Commit: f10e26dac3777de88d0c425a52329fd2724804ff
Parents: 4f19fee
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Mon Aug 7 15:44:08 2017 -0500
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Sat Aug 19 11:23:01 2017 +0900
----------------------------------------------------------------------
.../src/jvm/org/apache/storm/daemon/supervisor/Slot.java | 11 +++++++++++
.../storm/daemon/supervisor/timer/UpdateBlobs.java | 11 ++++++++---
2 files changed, 19 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f10e26da/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Slot.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Slot.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Slot.java
index c92db0c..11f800a 100644
--- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Slot.java
+++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Slot.java
@@ -749,6 +749,17 @@ public class Slot extends Thread implements AutoCloseable {
LOG.info("SLOT {}: Changing current assignment from {} to {}", staticState.port, dynamicState.currentAssignment, nextState.currentAssignment);
saveNewAssignment(nextState.currentAssignment);
}
+
+ if (equivalent(nextState.newAssignment, nextState.currentAssignment) &&
+ nextState.currentAssignment != null && nextState.currentAssignment.get_owner() == null &&
+ nextState.newAssignment != null && nextState.newAssignment.get_owner() != null) {
+ //This is an odd case for a rolling upgrade where the user on the old assignment may be null,
+ // but not on the new one. Although in all other ways they are the same.
+ // If this happens we want to use the assignment with the owner.
+ LOG.info("Updating assignment to save owner {}", nextState.newAssignment.get_owner());
+ saveNewAssignment(nextState.newAssignment);
+ nextState = nextState.withCurrentAssignment(nextState.container, nextState.newAssignment);
+ }
// clean up the profiler actions that are not being processed
removed.removeAll(dynamicState.profileActions);
http://git-wip-us.apache.org/repos/asf/storm/blob/f10e26da/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/UpdateBlobs.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/UpdateBlobs.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/UpdateBlobs.java
index 746578e..68c4e27 100644
--- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/UpdateBlobs.java
+++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/UpdateBlobs.java
@@ -67,9 +67,14 @@ public class UpdateBlobs implements Runnable {
for (String stormId : downloadedStormIds) {
LocalAssignment la = assignedStormIds.get(stormId);
if (la != null) {
- String stormRoot = ConfigUtils.supervisorStormDistRoot(conf, stormId);
- LOG.debug("Checking Blob updates for storm topology id {} With target_dir: {}", stormId, stormRoot);
- updateBlobsForTopology(conf, stormId, supervisor.getLocalizer(), la.get_owner());
+ if (la.get_owner() == null) {
+ //We got a case where the local assignment is not up to date, no point in going on...
+ LOG.warn("The blobs will not be updated for {} until the local assignment is updated...", stormId);
+ } else {
+ String stormRoot = ConfigUtils.supervisorStormDistRoot(conf, stormId);
+ LOG.debug("Checking Blob updates for storm topology id {} With target_dir: {}", stormId, stormRoot);
+ updateBlobsForTopology(conf, stormId, supervisor.getLocalizer(), la.get_owner());
+ }
}
}
} catch (Exception e) {