This is an automated email from the ASF dual-hosted git repository.
ethanli pushed a commit to branch 2.1.x-branch
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/2.1.x-branch by this push:
new 8de8c6a STORM-3516 Kill or Rebalance Topology not processed on Nimbus
restart (#3133)
new c03af8f Merge pull request #3134 from dandsager1/STORM-3516-2.1.x
8de8c6a is described below
commit 8de8c6aac2b482fa7c630d09f2749c89bda82787
Author: dandsager1 <[email protected]>
AuthorDate: Thu Oct 3 19:51:49 2019 -0500
STORM-3516 Kill or Rebalance Topology not processed on Nimbus restart
(#3133)
---
.../org/apache/storm/daemon/nimbus/Nimbus.java | 37 +++++++++++++++-------
.../storm/daemon/nimbus/TopologyActions.java | 2 +-
2 files changed, 26 insertions(+), 13 deletions(-)
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 6c8734c..d9dd550 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
@@ -360,12 +360,12 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
return sb;
};
- private static final TopologyStateTransition
STARTUP_WHEN_KILLED_TRANSITION = (args, nimbus, topoId, base) -> {
+ private static final TopologyStateTransition
GAIN_LEADERSHIP_WHEN_KILLED_TRANSITION = (args, nimbus, topoId, base) -> {
int delay =
base.get_topology_action_options().get_kill_options().get_wait_secs();
nimbus.delayEvent(topoId, delay, TopologyActions.REMOVE, null);
return null;
};
- private static final TopologyStateTransition
STARTUP_WHEN_REBALANCING_TRANSITION = (args, nimbus, topoId, base) -> {
+ private static final TopologyStateTransition
GAIN_LEADERSHIP_WHEN_REBALANCING_TRANSITION = (args, nimbus, topoId, base) -> {
int delay =
base.get_topology_action_options().get_rebalance_options().get_wait_secs();
nimbus.delayEvent(topoId, delay, TopologyActions.DO_REBALANCE, null);
return null;
@@ -385,12 +385,12 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
.put(TopologyActions.KILL, KILL_TRANSITION)
.build())
.put(TopologyStatus.KILLED, new
ImmutableMap.Builder<TopologyActions, TopologyStateTransition>()
- .put(TopologyActions.STARTUP, STARTUP_WHEN_KILLED_TRANSITION)
+ .put(TopologyActions.GAIN_LEADERSHIP,
GAIN_LEADERSHIP_WHEN_KILLED_TRANSITION)
.put(TopologyActions.KILL, KILL_TRANSITION)
.put(TopologyActions.REMOVE, REMOVE_TRANSITION)
.build())
.put(TopologyStatus.REBALANCING, new
ImmutableMap.Builder<TopologyActions, TopologyStateTransition>()
- .put(TopologyActions.STARTUP,
STARTUP_WHEN_REBALANCING_TRANSITION)
+ .put(TopologyActions.GAIN_LEADERSHIP,
GAIN_LEADERSHIP_WHEN_REBALANCING_TRANSITION)
.put(TopologyActions.KILL, KILL_TRANSITION)
.put(TopologyActions.DO_REBALANCE, DO_REBALANCE_TRANSITION)
.build())
@@ -462,6 +462,7 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
private AtomicReference<Map<String, Set<List<Integer>>>> idToExecutors;
//May be null if worker tokens are not supported by the thrift transport.
private WorkerTokenManager workerTokenManager;
+ private boolean wasLeader = false;
public Nimbus(Map<String, Object> conf, INimbus inimbus,
StormMetricsRegistry metricsRegistry) throws Exception {
this(conf, inimbus, null, null, null, null, null, metricsRegistry);
@@ -1624,8 +1625,8 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
throw new RuntimeException(message);
}
- if (TopologyActions.STARTUP != event) {
- //STARTUP is a system event so don't log an issue
+ if (TopologyActions.GAIN_LEADERSHIP != event) {
+ //GAIN_LEADERSHIP is a system event so don't log an
issue
LOG.info(message);
}
transition = NOOP_TRANSITION;
@@ -2875,12 +2876,24 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
exec.prepare();
}
- if (isLeader()) {
- for (String topoId : state.activeStorms()) {
- transition(topoId, TopologyActions.STARTUP, null);
- }
- clusterMetricSet.setActive(true);
- }
+ // Leadership coordination may be incomplete when launchServer is
called. Previous behavior did a one time check
+ // which could cause Nimbus to not process
TopologyActions.GAIN_LEADERSHIP transitions. Similar problem exists for
+ // HA Nimbus on being newly elected as leader. Change to a
recurring pattern addresses these problems.
+ timer.scheduleRecurring(3, 5,
+ () -> {
+ try {
+ boolean isLeader = isLeader();
+ if (isLeader && !wasLeader) {
+ for (String topoId : state.activeStorms()) {
+ transition(topoId,
TopologyActions.GAIN_LEADERSHIP, null);
+ }
+ clusterMetricSet.setActive(true);
+ }
+ wasLeader = isLeader;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
final boolean doNotReassign = (Boolean)
conf.getOrDefault(ServerConfigUtils.NIMBUS_DO_NOT_REASSIGN, false);
timer.scheduleRecurring(0,
ObjectReader.getInt(conf.get(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS)),
diff --git
a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/TopologyActions.java
b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/TopologyActions.java
index bc41da5..999d52d 100644
---
a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/TopologyActions.java
+++
b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/TopologyActions.java
@@ -16,7 +16,7 @@ package org.apache.storm.daemon.nimbus;
* Actions that can be done to a topology in nimbus
*/
public enum TopologyActions {
- STARTUP,
+ GAIN_LEADERSHIP,
INACTIVATE,
ACTIVATE,
REBALANCE,
