helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject helix git commit: [HELIX-657] Fix unexpected idealstate overwrite when persist assignment is on.
Date Wed, 24 May 2017 19:11:32 GMT
Repository: helix
Updated Branches:
  refs/heads/helix-0.6.x 8ba068e7b -> 091337c42


[HELIX-657] Fix unexpected idealstate overwrite when persist assignment is on.

1. Change persist method from set to update in PersistAssignmentStage.
The new updater only overwrites map and list fields that the controller will update during
PersistAssignmentStage.
All the other updates from other sources that are made during controller read and write will
be kept, as long as those fields are not purposely updated by the controller.
If current node does not exist, new updater return null.
2. Update accessors who relies on updater to check new data before applying the change. If
the returned new data is null, should skip updating or creating.

Also, add a test case for PersistAssignmentStage to cover the change.


Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/091337c4
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/091337c4
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/091337c4

Branch: refs/heads/helix-0.6.x
Commit: 091337c4249f37fd3235c9cf630a861a6abee557
Parents: 8ba068e
Author: Jiajun Wang <jjwang@linkedin.com>
Authored: Tue May 23 18:02:24 2017 -0700
Committer: Jiajun Wang <jjwang@linkedin.com>
Committed: Wed May 24 12:05:57 2017 -0700

----------------------------------------------------------------------
 .../org/apache/helix/HelixDataAccessor.java     |  9 ++
 .../stages/PersistAssignmentStage.java          | 16 +++-
 .../helix/manager/zk/ZKHelixDataAccessor.java   |  7 +-
 .../helix/manager/zk/ZkBaseDataAccessor.java    | 23 +++--
 .../src/test/java/org/apache/helix/Mocks.java   | 17 +++-
 .../integration/TestPersistAssignmentStage.java | 99 ++++++++++++++++++++
 6 files changed, 159 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java b/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
index 0f48539..e7777c6 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
@@ -60,6 +60,15 @@ public interface HelixDataAccessor {
   <T extends HelixProperty> boolean updateProperty(PropertyKey key, T value);
 
   /**
+   * Updates a property using specified updater
+   * @param key
+   * @param updater an update routine for the data to merge in
+   * @param value
+   * @return true if the update was successful
+   */
+  <T extends HelixProperty> boolean updateProperty(PropertyKey key, DataUpdater<ZNRecord>
updater, T value);
+
+  /**
    * Return the property value, it must be refer to a single Helix Property. i.e
    * PropertyKey.isLeaf() must return true.
    * @param key

http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/main/java/org/apache/helix/controller/stages/PersistAssignmentStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/PersistAssignmentStage.java
b/helix-core/src/main/java/org/apache/helix/controller/stages/PersistAssignmentStage.java
index 425b38b..8255cf4 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/PersistAssignmentStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/PersistAssignmentStage.java
@@ -23,9 +23,11 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.I0Itec.zkclient.DataUpdater;
 import org.apache.helix.HelixDataAccessor;
 import org.apache.helix.HelixManager;
 import org.apache.helix.PropertyKey;
+import org.apache.helix.ZNRecord;
 import org.apache.helix.controller.pipeline.AbstractBaseStage;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.IdealState;
@@ -97,7 +99,19 @@ public class PersistAssignmentStage extends AbstractBaseStage {
         }
 
         if (needPersist) {
-          accessor.setProperty(keyBuilder.idealStates(resourceId), idealState);
+          // Update instead of set to ensure any intermediate changes that the controller
does not update are kept.
+          accessor.updateProperty(keyBuilder.idealStates(resourceId), new DataUpdater<ZNRecord>()
{
+            @Override
+            public ZNRecord update(ZNRecord current) {
+              if (current != null) {
+                // Overwrite MapFields and ListFields items with the same key.
+                // Note that default merge will keep old values in the maps or lists unchanged,
which is not desired.
+                current.getMapFields().putAll(idealState.getRecord().getMapFields());
+                current.getListFields().putAll(idealState.getRecord().getListFields());
+              }
+              return current;
+            }
+          }, idealState);
         }
       }
     }

http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
index 791c6d8..d9818ca 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixDataAccessor.java
@@ -157,6 +157,11 @@ public class ZKHelixDataAccessor implements HelixDataAccessor {
 
   @Override
   public <T extends HelixProperty> boolean updateProperty(PropertyKey key, T value)
{
+    return updateProperty(key, new ZNRecordUpdater(value.getRecord()), value);
+  }
+
+  @Override
+  public <T extends HelixProperty> boolean updateProperty(PropertyKey key, DataUpdater<ZNRecord>
updater, T value) {
     PropertyType type = key.getType();
     String path = key.getPath();
     int options = constructOptions(type);
@@ -172,7 +177,7 @@ public class ZKHelixDataAccessor implements HelixDataAccessor {
       }
       break;
     default:
-      success = _baseDataAccessor.update(path, new ZNRecordUpdater(value.getRecord()), options);
+      success = _baseDataAccessor.update(path, updater, options);
       break;
     }
     return success;

http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
index d79abeb..dacecd9 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
@@ -256,9 +256,10 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T>
{
         Stat readStat = new Stat();
         T oldData = (T) _zkClient.readData(path, readStat);
         T newData = updater.update(oldData);
-        Stat setStat = _zkClient.writeDataGetStat(path, newData, readStat.getVersion());
-        DataTree.copyStat(setStat, result._stat);
-
+        if (newData != null) {
+          Stat setStat = _zkClient.writeDataGetStat(path, newData, readStat.getVersion());
+          DataTree.copyStat(setStat, result._stat);
+        }
         updatedData = newData;
       } catch (ZkBadVersionException e) {
         retry = true;
@@ -266,9 +267,15 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T>
{
         // node not exist, try create, pass null to updater
         try {
           T newData = updater.update(null);
-          AccessResult res = doCreate(path, newData, options);
-          result._pathCreated.addAll(res._pathCreated);
-          RetCode rc = res._retCode;
+          RetCode rc;
+          if (newData != null) {
+            AccessResult res = doCreate(path, newData, options);
+            result._pathCreated.addAll(res._pathCreated);
+            rc = res._retCode;
+          } else {
+            // If update returns null, no need to create.
+            rc = RetCode.OK;
+          }
           switch (rc) {
           case OK:
             updatedData = newData;
@@ -810,6 +817,10 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T>
{
           DataUpdater<T> updater = updaters.get(i);
           T newData = updater.update(curDataList.get(i));
           newDataList.add(newData);
+          if (newData == null) {
+            // No need to create or update if the updater does not return a new version
+            continue;
+          }
           Stat curStat = curStats.get(i);
           if (curStat == null) {
             // node not exists

http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/test/java/org/apache/helix/Mocks.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/Mocks.java b/helix-core/src/test/java/org/apache/helix/Mocks.java
index c60a693..70469d8 100644
--- a/helix-core/src/test/java/org/apache/helix/Mocks.java
+++ b/helix-core/src/test/java/org/apache/helix/Mocks.java
@@ -508,14 +508,21 @@ public class Mocks {
 
     @Override
     public <T extends HelixProperty> boolean updateProperty(PropertyKey key, T value)
{
+      return updateProperty(key, new ZNRecordUpdater(value.getRecord()) , value);
+    }
+
+    @Override
+    public <T extends HelixProperty> boolean updateProperty(PropertyKey key, DataUpdater<ZNRecord>
updater, T value) {
       String path = key.getPath();
       PropertyType type = key.getType();
       if (type.updateOnlyOnExists) {
         if (data.containsKey(path)) {
           if (type.mergeOnUpdate) {
             ZNRecord znRecord = new ZNRecord(data.get(path));
-            znRecord.merge(value.getRecord());
-            data.put(path, znRecord);
+            ZNRecord newZNRecord = updater.update(znRecord);
+            if (newZNRecord != null) {
+              data.put(path, newZNRecord);
+            }
           } else {
             data.put(path, value.getRecord());
           }
@@ -524,8 +531,10 @@ public class Mocks {
         if (type.mergeOnUpdate) {
           if (data.containsKey(path)) {
             ZNRecord znRecord = new ZNRecord(data.get(path));
-            znRecord.merge(value.getRecord());
-            data.put(path, znRecord);
+            ZNRecord newZNRecord = updater.update(znRecord);
+            if (newZNRecord != null) {
+              data.put(path, znRecord);
+            }
           } else {
             data.put(path, value.getRecord());
           }

http://git-wip-us.apache.org/repos/asf/helix/blob/091337c4/helix-core/src/test/java/org/apache/helix/integration/TestPersistAssignmentStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestPersistAssignmentStage.java
b/helix-core/src/test/java/org/apache/helix/integration/TestPersistAssignmentStage.java
new file mode 100644
index 0000000..25bd8a7
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestPersistAssignmentStage.java
@@ -0,0 +1,99 @@
+package org.apache.helix.integration;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.controller.pipeline.Stage;
+import org.apache.helix.controller.pipeline.StageContext;
+import org.apache.helix.controller.stages.AttributeName;
+import org.apache.helix.controller.stages.BestPossibleStateOutput;
+import org.apache.helix.controller.stages.ClusterDataCache;
+import org.apache.helix.controller.stages.ClusterEvent;
+import org.apache.helix.controller.stages.PersistAssignmentStage;
+import org.apache.helix.controller.stages.ReadClusterDataStage;
+import org.apache.helix.controller.stages.ResourceComputationStage;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.Partition;
+import org.apache.helix.tools.DefaultIdealStateCalculator;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class TestPersistAssignmentStage extends ZkStandAloneCMTestBase {
+  ClusterEvent event = new ClusterEvent("sampleEvent");
+
+  /**
+   * Case where we have one resource in IdealState
+   * @throws Exception
+   */
+  @Test
+  public void testSimple() throws Exception {
+    int nodes = 2;
+    List<String> instances = new ArrayList<String>();
+    for (int i = 0; i < nodes; i++) {
+      instances.add("localhost_" + i);
+    }
+    int partitions = 10;
+    int replicas = 1;
+    String resourceName = "testResource";
+    ZNRecord record =
+        DefaultIdealStateCalculator.calculateIdealState(instances, partitions, replicas,
resourceName, "ONLINE",
+            "OFFLINE");
+    IdealState idealState = new IdealState(record);
+    idealState.setStateModelDefRef("OnlineOffline");
+    idealState.setRebalanceMode(IdealState.RebalanceMode.FULL_AUTO);
+
+    // Read and load current state into event
+    HelixDataAccessor accessor = _manager.getHelixDataAccessor();
+    PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+    accessor.setProperty(keyBuilder.idealStates(resourceName), idealState);
+    runStage(event, new ReadClusterDataStage());
+    runStage(event, new ResourceComputationStage());
+
+    // Ensure persist best possible assignment is true
+    ClusterDataCache cache = event.getAttribute("ClusterDataCache");
+    cache.getClusterConfig().setPersistBestPossibleAssignment(true);
+
+    // 1. Change best possible state (simulate a new rebalancer run)
+    BestPossibleStateOutput bestPossibleStateOutput = new BestPossibleStateOutput();
+    for (String partition : idealState.getPartitionSet()) {
+      bestPossibleStateOutput.setState(resourceName, new Partition(partition), "localhost_3",
"OFFLINE");
+    }
+    // 2. At the same time, set DelayRebalanceEnabled = true (simulate a Admin operation
at the same time)
+    idealState.setDelayRebalanceDisabled(true);
+    accessor.setProperty(keyBuilder.idealStates(resourceName), idealState);
+
+    // Persist new assignment
+    PersistAssignmentStage stage = new PersistAssignmentStage();
+    event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
+    runStage(event, stage);
+
+    IdealState newIdealState = accessor.getProperty(keyBuilder.idealStates(resourceName));
+    // 1. New assignment should be set
+    Assert.assertEquals(newIdealState.getPartitionSet().size(), idealState.getPartitionSet().size());
+    for (String partition : idealState.getPartitionSet()) {
+      Map<String, String> assignment = newIdealState.getInstanceStateMap(partition);
+      Assert.assertNotNull(assignment);
+      Assert.assertEquals(assignment.size(),1);
+      Assert.assertTrue(assignment.containsKey("localhost_3") && assignment.get("localhost_3").equals("OFFLINE"));
+    }
+    // 2. Admin config should be set
+    Assert.assertTrue(newIdealState.isDelayRebalanceDisabled());
+  }
+
+  private void runStage(ClusterEvent event, Stage stage) {
+    event.addAttribute("helixmanager", _manager);
+    StageContext context = new StageContext();
+    stage.init(context);
+    stage.preProcess();
+    try {
+      stage.process(event);
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    stage.postProcess();
+  }
+}


Mime
View raw message