You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ji...@apache.org on 2019/08/12 17:58:26 UTC
[helix] branch master updated: Fix the CallbackHandler registration
logic in DistributedLeaderElection (#395)
This is an automated email from the ASF dual-hosted git repository.
jiajunwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new aeeb095 Fix the CallbackHandler registration logic in DistributedLeaderElection (#395)
aeeb095 is described below
commit aeeb0954e86a36c114fd933ecc2be8245a58b1b9
Author: Jiajun Wang <18...@users.noreply.github.com>
AuthorDate: Mon Aug 12 10:58:21 2019 -0700
Fix the CallbackHandler registration logic in DistributedLeaderElection (#395)
* Fix the CallbackHandler registration logic in DistributedLeaderElection that may cause a leader node has no callback registered.
Our current initialization logic assumes a strict leader acquire/relinquish events sequence. However, due to the possible carried over ZK events from the previous ZK session, the controller node change event might be triggered in the following sequence:
1. CALLBACK (from the previous session): Create new leader node and add handlers.
2. FINALIZE (Handle the previous session expire): Clean up handlers.
3. INIT (For the new session establishment): Expect to add the handlers back again.
As a result, if the INIT event processing does not recover the handlers, the leader controller won't be able to manage anything. This fix ensures all the acquireLeadership call will try to initialize the leader controller's callback handlers.
Also, add the additional test logic in TestHandleNewSession to verify the fix.
* Improve the leader history update logic so there is no duplicate entry recorded.
---
.../healthcheck/ParticipantHealthReportTask.java | 4 ++--
.../manager/zk/DistributedLeaderElection.java | 25 ++++++++++++----------
.../apache/helix/manager/zk/ZKHelixManager.java | 4 ++--
.../helix/manager/zk/TestHandleNewSession.java | 10 +++++++++
4 files changed, 28 insertions(+), 15 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/healthcheck/ParticipantHealthReportTask.java b/helix-core/src/main/java/org/apache/helix/healthcheck/ParticipantHealthReportTask.java
index 54c8cbe..f48d6d7 100644
--- a/helix-core/src/main/java/org/apache/helix/healthcheck/ParticipantHealthReportTask.java
+++ b/helix-core/src/main/java/org/apache/helix/healthcheck/ParticipantHealthReportTask.java
@@ -54,7 +54,7 @@ public class ParticipantHealthReportTask extends HelixTimerTask {
}
@Override
- public void start() {
+ public synchronized void start() {
if (_timer == null) {
LOG.info("Start HealthCheckInfoReportingTask");
_timer = new Timer("ParticipantHealthReportTimerTask", true);
@@ -66,7 +66,7 @@ public class ParticipantHealthReportTask extends HelixTimerTask {
}
@Override
- public void stop() {
+ public synchronized void stop() {
if (_timer != null) {
LOG.info("Stop ParticipantHealthReportTimerTask");
_timer.cancel();
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
index 179524f..2218508 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DistributedLeaderElection.java
@@ -19,10 +19,6 @@ package org.apache.helix.manager.zk;
* under the License.
*/
-import java.lang.management.ManagementFactory;
-import java.util.List;
-
-import org.I0Itec.zkclient.DataUpdater;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
@@ -40,6 +36,9 @@ import org.apache.helix.model.LiveInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.management.ManagementFactory;
+import java.util.List;
+
/**
* do distributed leader election
*/
@@ -104,16 +103,20 @@ public class DistributedLeaderElection implements ControllerChangeListener {
HelixDataAccessor accessor = manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
- while (accessor.getProperty(keyBuilder.controllerLeader()) == null) {
+ // Try to acquire leader and init the manager in any case.
+ // Even when a leader node already exists, the election process shall still try to init the manager
+ // in case it is the current leader.
+ do {
+ // Due to the possible carried over ZK events from the previous ZK session, the following
+ // initialization might be triggered multiple times. So the operation must be idempotent.
if (tryCreateController(manager)) {
- LOG.info("{} with session {} acquired leadership for cluster: {}", manager.getInstanceName(),
- manager.getSessionId(), manager.getClusterName());
- updateHistory(manager);
+ LOG.info("{} with session {} acquired leadership for cluster: {}",
+ manager.getInstanceName(), manager.getSessionId(), manager.getClusterName());
manager.getHelixDataAccessor().getBaseDataAccessor().reset();
controllerHelper.addListenersToController(_controller);
controllerHelper.startControllerTimerTasks();
}
- }
+ } while (accessor.getProperty(keyBuilder.controllerLeader()) == null);
}
/**
@@ -128,8 +131,8 @@ public class DistributedLeaderElection implements ControllerChangeListener {
newLeader.setSessionId(manager.getSessionId());
newLeader.setHelixVersion(manager.getVersion());
try {
- boolean success = accessor.createControllerLeader(newLeader);
- if (success) {
+ if (accessor.createControllerLeader(newLeader)) {
+ updateHistory(manager);
return true;
} else {
LOG.info(
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index 2fba83c..31d45c0 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -167,7 +167,7 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
@Override
- public void start() {
+ public synchronized void start() {
long initialDelay = 0;
long period = 15 * 60 * 1000;
long timeThresholdNoChangeForStatusUpdates = 15 * 60 * 1000; // 15 minutes
@@ -185,7 +185,7 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
@Override
- public void stop() {
+ public synchronized void stop() {
if (_timer != null) {
LOG.info("Stop StatusDumpTask");
_timer.cancel();
diff --git a/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleNewSession.java b/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleNewSession.java
index 3f5d66f..da7dcc9 100644
--- a/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleNewSession.java
+++ b/helix-core/src/test/java/org/apache/helix/manager/zk/TestHandleNewSession.java
@@ -21,6 +21,7 @@ package org.apache.helix.manager.zk;
import java.util.Collections;
import java.util.Date;
+import java.util.List;
import java.util.concurrent.Semaphore;
import org.apache.helix.HelixException;
@@ -115,6 +116,8 @@ public class TestHandleNewSession extends ZkTestBase {
final long originalCreationTime =
accessor.getProperty(keyBuilder.controllerLeader()).getStat().getCreationTime();
+ int handlerCount = manager.getHandlers().size();
+
// 1. lock the zk event processing to simulate long backlog queue.
((ZkClient) manager._zkclient).getEventLock().lockInterruptibly();
// 2. add a controller leader node change event to the queue, that will not be processed.
@@ -161,6 +164,9 @@ public class TestHandleNewSession extends ZkTestBase {
manager.proceedNewSessionHandling();
// Since the new session handling will re-create the leader node, a new valid node shall be created.
Assert.assertTrue(TestHelper.verify(() -> manager.isLeader(), 1000));
+ // All the callback handlers shall be recovered.
+ Assert.assertTrue(TestHelper.verify(() -> manager.getHandlers().size() == handlerCount, 3000));
+ Assert.assertTrue(manager.getHandlers().stream().allMatch(handler -> handler.isReady()));
manager.disconnect();
TestHelper.dropCluster(clusterName, _gZkClient);
@@ -183,5 +189,9 @@ public class TestHandleNewSession extends ZkTestBase {
void proceedNewSessionHandling() {
newSessionHandlingCount.release();
}
+
+ List<CallbackHandler> getHandlers() {
+ return _handlers;
+ }
}
}