You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2024/02/27 01:47:28 UTC
(helix) branch master updated: Expose addListener and support listening on single instance config (#2752)
This is an automated email from the ASF dual-hosted git repository.
jxue 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 6bdc88804 Expose addListener and support listening on single instance config (#2752)
6bdc88804 is described below
commit 6bdc888042daefa7e75d93df44270db6d4a26849
Author: Zachary Pinto <za...@linkedin.com>
AuthorDate: Mon Feb 26 17:47:22 2024 -0800
Expose addListener and support listening on single instance config (#2752)
Expose addListener method in HelixManager to allow users to register any listeners that implement one of the Listener interfaces. Add support in CallbackHandler for InstanceConfigChangeListener to be used on a single instance config.
---
.../main/java/org/apache/helix/HelixManager.java | 14 +++-
.../apache/helix/manager/zk/CallbackHandler.java | 10 ++-
.../apache/helix/manager/zk/ZKHelixManager.java | 3 +-
.../event/MockCloudEventAwareHelixManager.java | 7 ++
.../controller/stages/DummyClusterManager.java | 7 ++
.../rebalancer/TestInstanceOperation.java | 74 ++++++++++++++++++----
.../java/org/apache/helix/mock/MockManager.java | 8 ++-
.../helix/participant/MockZKHelixManager.java | 7 ++
8 files changed, 113 insertions(+), 17 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/HelixManager.java b/helix-core/src/main/java/org/apache/helix/HelixManager.java
index b5c2a6448..c1d2ad18c 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixManager.java
@@ -51,7 +51,7 @@ import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.spectator.RoutingTableProvider;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
-
+import org.apache.zookeeper.Watcher;
/**
* Class that represents the Helix Agent.
@@ -108,6 +108,18 @@ public interface HelixManager {
*/
void disconnect();
+ /**
+ * Add a change listener on the specified propertyKey for the specified
+ * changeType and eventTypes.
+ * @see org.apache.helix.api.listeners for the list of available listeners
+ * @param listener the listener to add
+ * @param propertyKey the property key to listen to
+ * @param changeType the type of change to listen to
+ * @param eventType the event type to listen for
+ */
+ void addListener(Object listener, PropertyKey propertyKey, HelixConstants.ChangeType changeType,
+ Watcher.Event.EventType[] eventType);
+
/**
* @see IdealStateChangeListener#onIdealStateChange(List, NotificationContext)
* @param listener
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
index 2a19a7a4e..c2809620e 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java
@@ -358,7 +358,15 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener {
configChangeListener.onConfigChange(configs, changeContext);
} else if (_listener instanceof InstanceConfigChangeListener) {
InstanceConfigChangeListener listener = (InstanceConfigChangeListener) _listener;
- List<InstanceConfig> configs = preFetch(_propertyKey);
+ List<InstanceConfig> configs = Collections.emptyList();
+ if (_propertyKey.getParams().length > 2 && _preFetchEnabled) {
+ // If there are more than 2 params, that means the property key is for a specific instance
+ // and will not have children.
+ InstanceConfig config = _accessor.getProperty(_propertyKey);
+ configs = config != null ? Collections.singletonList(config) : Collections.emptyList();
+ } else {
+ configs = preFetch(_propertyKey);
+ }
listener.onInstanceConfigChange(configs, changeContext);
}
} else if (_changeType == RESOURCE_CONFIG) {
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 7a62d1923..3473bed08 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
@@ -419,7 +419,8 @@ public class ZKHelixManager implements HelixManager, IZkStateListener {
}
}
- void addListener(Object listener, PropertyKey propertyKey, ChangeType changeType,
+ @Override
+ public void addListener(Object listener, PropertyKey propertyKey, ChangeType changeType,
EventType[] eventType) {
checkConnected(_waitForConnectedTimeout);
diff --git a/helix-core/src/test/java/org/apache/helix/cloud/event/MockCloudEventAwareHelixManager.java b/helix-core/src/test/java/org/apache/helix/cloud/event/MockCloudEventAwareHelixManager.java
index de46257d3..740c932fc 100644
--- a/helix-core/src/test/java/org/apache/helix/cloud/event/MockCloudEventAwareHelixManager.java
+++ b/helix-core/src/test/java/org/apache/helix/cloud/event/MockCloudEventAwareHelixManager.java
@@ -26,6 +26,7 @@ import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixCloudProperty;
+import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
@@ -59,6 +60,7 @@ import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.zookeeper.Watcher;
public class MockCloudEventAwareHelixManager implements HelixManager {
private final HelixManagerProperty _helixManagerProperty;
@@ -110,6 +112,11 @@ public class MockCloudEventAwareHelixManager implements HelixManager {
return false;
}
+ @Override
+ public void addListener(Object listener, PropertyKey propertyKey, HelixConstants.ChangeType changeType,
+ Watcher.Event.EventType[] eventType) {
+ }
+
@Override
public void addIdealStateChangeListener(IdealStateChangeListener listener) throws Exception {
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/DummyClusterManager.java b/helix-core/src/test/java/org/apache/helix/controller/stages/DummyClusterManager.java
index 5d23a321c..4903a3d8b 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/DummyClusterManager.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/DummyClusterManager.java
@@ -24,6 +24,7 @@ import java.util.Set;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
@@ -53,6 +54,7 @@ import org.apache.helix.healthcheck.ParticipantHealthReportCollector;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.zookeeper.Watcher;
public class DummyClusterManager implements HelixManager {
HelixDataAccessor _accessor;
@@ -303,6 +305,11 @@ public class DummyClusterManager implements HelixManager {
return null;
}
+ @Override
+ public void addListener(Object listener, PropertyKey propertyKey, HelixConstants.ChangeType changeType,
+ Watcher.Event.EventType[] eventType) {
+ }
+
@Override
public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener)
throws Exception {
diff --git a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
index 276757741..010153e64 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestInstanceOperation.java
@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
@@ -25,8 +26,10 @@ import org.apache.helix.HelixManagerFactory;
import org.apache.helix.HelixRollbackException;
import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
+import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyType;
import org.apache.helix.TestHelper;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
import org.apache.helix.common.ZkTestBase;
import org.apache.helix.constants.InstanceConstants;
import org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy;
@@ -54,6 +57,7 @@ import org.apache.helix.spectator.RoutingTableProvider;
import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier;
import org.apache.helix.tools.ClusterVerifiers.StrictMatchExternalViewVerifier;
import org.apache.helix.tools.ClusterVerifiers.ZkHelixClusterVerifier;
+import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -546,12 +550,16 @@ public class TestInstanceOperation extends ZkTestBase {
validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
Collections.emptySet(), Collections.emptySet());
+ CustomIndividualInstanceConfigChangeListener instanceToSwapInInstanceConfigListener =
+ new CustomIndividualInstanceConfigChangeListener();
// Add instance with InstanceOperation set to SWAP_IN
String instanceToSwapInName = PARTICIPANT_PREFIX + "_" + _nextStartPort;
swapOutInstancesToSwapInInstances.put(instanceToSwapOutName, instanceToSwapInName);
addParticipant(instanceToSwapInName, instanceToSwapOutInstanceConfig.getLogicalId(LOGICAL_ID),
instanceToSwapOutInstanceConfig.getDomainAsMap().get(ZONE),
- InstanceConstants.InstanceOperation.SWAP_IN, true, -1);
+ InstanceConstants.InstanceOperation.SWAP_IN, true, -1, instanceToSwapInInstanceConfigListener);
+
+ Assert.assertFalse(instanceToSwapInInstanceConfigListener.isThrottlesEnabled());
// Validate that partitions on SWAP_OUT instance does not change after setting the InstanceOperation to SWAP_OUT
// and adding the SWAP_IN instance to the cluster.
@@ -583,6 +591,9 @@ public class TestInstanceOperation extends ZkTestBase {
Assert.assertFalse(_gSetupTool.getClusterManagementTool()
.getInstanceConfig(CLUSTER_NAME, instanceToSwapOutName).getInstanceEnabled());
+ // Check to make sure the throttle was enabled again after the swap was completed.
+ Assert.assertTrue(instanceToSwapInInstanceConfigListener.isThrottlesEnabled());
+
// Validate that the SWAP_IN instance has the same partitions the SWAP_OUT instance had before
// swap was completed.
verifier(() -> (validateEVsCorrect(getEVs(), originalEVs, swapOutInstancesToSwapInInstances,
@@ -1095,7 +1106,7 @@ public class TestInstanceOperation extends ZkTestBase {
}
@Test(expectedExceptions = HelixException.class, dependsOnMethods = "testNodeSwapWithSwapOutInstanceOffline")
- public void testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet() {
+ public void testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet() throws Exception {
System.out.println(
"START TestInstanceOperation.testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet() at "
+ new Date(System.currentTimeMillis()));
@@ -1113,7 +1124,7 @@ public class TestInstanceOperation extends ZkTestBase {
}
@Test(expectedExceptions = HelixException.class, dependsOnMethods = "testNodeSwapAddSwapInFirstEnabledBeforeSwapOutSet")
- public void testNodeSwapAddSwapInFirstEnableBeforeSwapOutSet() {
+ public void testNodeSwapAddSwapInFirstEnableBeforeSwapOutSet() throws Exception {
System.out.println(
"START TestInstanceOperation.testNodeSwapAddSwapInFirstEnableBeforeSwapOutSet() at "
+ new Date(System.currentTimeMillis()));
@@ -1136,7 +1147,7 @@ public class TestInstanceOperation extends ZkTestBase {
}
@Test(expectedExceptions = HelixException.class, dependsOnMethods = "testNodeSwapAddSwapInFirstEnableBeforeSwapOutSet")
- public void testUnsetInstanceOperationOnSwapInWhenAlreadyUnsetOnSwapOut() {
+ public void testUnsetInstanceOperationOnSwapInWhenAlreadyUnsetOnSwapOut() throws Exception {
System.out.println(
"START TestInstanceOperation.testUnsetInstanceOperationOnSwapInWhenAlreadyUnsetOnSwapOut() at "
+ new Date(System.currentTimeMillis()));
@@ -1380,7 +1391,7 @@ public class TestInstanceOperation extends ZkTestBase {
}
@Test(expectedExceptions = HelixException.class, dependsOnMethods = "testEvacuationWithOfflineInstancesInCluster")
- public void testSwapEvacuateAddRemoveEvacuate() {
+ public void testSwapEvacuateAddRemoveEvacuate() throws Exception {
System.out.println("START TestInstanceOperation.testSwapEvacuateAddRemoveEvacuate() at " + new Date(
System.currentTimeMillis()));
removeOfflineOrDisabledOrSwapInInstances();
@@ -1426,9 +1437,33 @@ public class TestInstanceOperation extends ZkTestBase {
}, timeout));
}
- private MockParticipantManager createParticipant(String participantName) {
+ private static class CustomIndividualInstanceConfigChangeListener implements InstanceConfigChangeListener {
+ private boolean throttlesEnabled;
+
+ public CustomIndividualInstanceConfigChangeListener() {
+ throttlesEnabled = true;
+ }
+
+ public boolean isThrottlesEnabled() {
+ return throttlesEnabled;
+ }
+
+ @Override
+ public void onInstanceConfigChange(List<InstanceConfig> instanceConfig,
+ NotificationContext context) {
+ if (instanceConfig.get(0).getInstanceOperation()
+ .equals(InstanceConstants.InstanceOperation.SWAP_IN.name())) {
+ throttlesEnabled = false;
+ } else if (instanceConfig.get(0).getInstanceOperation().isEmpty()) {
+ throttlesEnabled = true;
+ }
+ }
+ }
+
+ private MockParticipantManager createParticipant(String participantName) throws Exception {
// start dummy participants
- MockParticipantManager participant = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, participantName);
+ MockParticipantManager participant =
+ new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, participantName, 10, null);
StateMachineEngine stateMachine = participant.getStateMachineEngine();
// Using a delayed state model
StDelayMSStateModelFactory delayFactory = new StDelayMSStateModelFactory();
@@ -1436,8 +1471,20 @@ public class TestInstanceOperation extends ZkTestBase {
return participant;
}
+ private void addParticipant(String participantName) throws Exception {
+ addParticipant(participantName, UUID.randomUUID().toString(),
+ "zone_" + _participants.size() % ZONE_COUNT, null, true, -1);
+ }
+
private void addParticipant(String participantName, String logicalId, String zone,
- InstanceConstants.InstanceOperation instanceOperation, boolean enabled, int capacity) {
+ InstanceConstants.InstanceOperation instanceOperation, boolean enabled, int capacity)
+ throws Exception {
+ addParticipant(participantName, logicalId, zone, instanceOperation, enabled, capacity, null);
+ }
+
+ private void addParticipant(String participantName, String logicalId, String zone,
+ InstanceConstants.InstanceOperation instanceOperation, boolean enabled, int capacity,
+ InstanceConfigChangeListener listener) throws Exception {
InstanceConfig config = new InstanceConfig.Builder().setDomain(
String.format("%s=%s, %s=%s, %s=%s", ZONE, zone, HOST, participantName, LOGICAL_ID,
logicalId)).setInstanceEnabled(enabled).setInstanceOperation(instanceOperation)
@@ -1451,16 +1498,17 @@ public class TestInstanceOperation extends ZkTestBase {
MockParticipantManager participant = createParticipant(participantName);
participant.syncStart();
+ if (listener != null) {
+ participant.addListener(listener,
+ new PropertyKey.Builder(CLUSTER_NAME).instanceConfig(participantName),
+ HelixConstants.ChangeType.INSTANCE_CONFIG,
+ new Watcher.Event.EventType[]{Watcher.Event.EventType.NodeDataChanged});
+ }
_participants.add(participant);
_participantNames.add(participantName);
_nextStartPort++;
}
- private void addParticipant(String participantName) {
- addParticipant(participantName, UUID.randomUUID().toString(),
- "zone_" + _participants.size() % ZONE_COUNT, null, true, -1);
- }
-
private void createTestDBs(long delayTime) throws InterruptedException {
createResourceWithDelayedRebalance(CLUSTER_NAME, "TEST_DB0_CRUSHED",
BuiltInStateModelDefinitions.LeaderStandby.name(), PARTITIONS, REPLICA, REPLICA - 1, -1,
diff --git a/helix-core/src/test/java/org/apache/helix/mock/MockManager.java b/helix-core/src/test/java/org/apache/helix/mock/MockManager.java
index afa8d9880..b2d474a52 100644
--- a/helix-core/src/test/java/org/apache/helix/mock/MockManager.java
+++ b/helix-core/src/test/java/org/apache/helix/mock/MockManager.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
@@ -56,7 +57,7 @@ import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.participant.HelixStateMachineEngine;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
-
+import org.apache.zookeeper.Watcher;
public class MockManager implements HelixManager {
MockAccessor accessor;
@@ -328,6 +329,11 @@ public class MockManager implements HelixManager {
return null;
}
+ @Override
+ public void addListener(Object listener, PropertyKey propertyKey, HelixConstants.ChangeType changeType,
+ Watcher.Event.EventType[] eventType) {
+ }
+
@Override
public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener)
throws Exception {
diff --git a/helix-core/src/test/java/org/apache/helix/participant/MockZKHelixManager.java b/helix-core/src/test/java/org/apache/helix/participant/MockZKHelixManager.java
index 897d99dac..997216518 100644
--- a/helix-core/src/test/java/org/apache/helix/participant/MockZKHelixManager.java
+++ b/helix-core/src/test/java/org/apache/helix/participant/MockZKHelixManager.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerProperties;
@@ -58,6 +59,7 @@ import org.apache.helix.messaging.DefaultMessagingService;
import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.task.TaskConstants;
+import org.apache.zookeeper.Watcher;
import org.testng.collections.Lists;
public class MockZKHelixManager implements HelixManager {
@@ -324,6 +326,11 @@ public class MockZKHelixManager implements HelixManager {
(ZkBaseDataAccessor<ZNRecord>) _accessor.getBaseDataAccessor(), TaskConstants.REBALANCER_CONTEXT_ROOT, Lists.<String>newArrayList());
}
+ @Override
+ public void addListener(Object listener, PropertyKey propertyKey, HelixConstants.ChangeType changeType,
+ Watcher.Event.EventType[] eventType) {
+ }
+
@Override
public void addInstanceConfigChangeListener(InstanceConfigChangeListener listener)
throws Exception {