import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@Override
public boolean update(String path, DataUpdater<T> updater, int options) {
  String clientPath = path;
  String serverPath = prependChroot(clientPath);

  Cache<T> cache = getCache(serverPath);

  if (cache != null) {
    try {
      cache.lockWrite();
      ZkBaseDataAccessor<T>.AccessResult result =
          _baseAccessor.doUpdate(serverPath, updater, options);
      boolean success = (result._retCode == RetCode.OK);
      updateCache(cache, result._pathCreated, success, serverPath, result._updatedValue,
          result._stat);

      return success;
    } finally {
      cache.unlockWrite();
    }
  }

  // no cache
  return _groupCommit.commit(_baseAccessor, options, serverPath, updater);
  // return _baseAccessor.update(serverPath, updater, options);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * Updates data of an existing znode. The current content of the znode is passed to the {@link DataUpdater} that is
 * passed into this method, which returns the new content. The new content is only written back to ZooKeeper if
 * nobody has modified the given znode in between. If a concurrent change has been detected the new data of the
 * znode is passed to the updater once again until the new contents can be successfully written back to ZooKeeper.
 *
 * @param <T>
 * @param path
 *            The path of the znode.
 * @param updater
 *            Updater that creates the new contents.
 */
@SuppressWarnings("unchecked") public <T extends Object> void updateDataSerialized(String path,
    DataUpdater<T> updater) {
  Stat stat = new Stat();
  boolean retry;
  do {
    retry = false;
    try {
      T oldData = (T) readData(path, stat);
      T newData = updater.update(oldData);
      writeData(path, newData, stat.getVersion());
    } catch (ZkBadVersionException e) {
      retry = true;
    }
  } while (retry);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@Override
public void setConstraint(String clusterName, final ConstraintType constraintType,
    final String constraintId, final ConstraintItem constraintItem) {
  BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<ZNRecord>(_zkClient);

  Builder keyBuilder = new Builder(clusterName);
  String path = keyBuilder.constraint(constraintType.toString()).getPath();

  baseAccessor.update(path, new DataUpdater<ZNRecord>() {
    @Override
    public ZNRecord update(ZNRecord currentData) {
      ClusterConstraints constraints = currentData == null
          ? new ClusterConstraints(constraintType)
          : new ClusterConstraints(currentData);

      constraints.addConstraintItem(constraintId, constraintItem);
      return constraints.getRecord();
    }
  }, AccessOption.PERSISTENT);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@Override
public void removeConstraint(String clusterName, final ConstraintType constraintType,
    final String constraintId) {
  BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<ZNRecord>(_zkClient);

  Builder keyBuilder = new Builder(clusterName);
  String path = keyBuilder.constraint(constraintType.toString()).getPath();

  baseAccessor.update(path, new DataUpdater<ZNRecord>() {
    @Override
    public ZNRecord update(ZNRecord currentData) {
      if (currentData != null) {
        ClusterConstraints constraints = new ClusterConstraints(currentData);

        constraints.removeConstraintItem(constraintId);
        return constraints.getRecord();
      }
      return null;
    }
  }, AccessOption.PERSISTENT);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
private void enableSingleInstance(final String clusterName, final String instanceName,
    final boolean enabled, BaseDataAccessor<ZNRecord> baseAccessor) {
  String path = PropertyPathBuilder.instanceConfig(clusterName, instanceName);

  if (!baseAccessor.exists(path, 0)) {
    throw new HelixException("Cluster " + clusterName + ", instance: " + instanceName
        + ", instance config does not exist");
  }

  baseAccessor.update(path, new DataUpdater<ZNRecord>()

  {
    @Override
    public ZNRecord update(ZNRecord currentData) {
      if (currentData == null) {
        throw new HelixException("Cluster: " + clusterName + ", instance: " + instanceName
            + ", participant config is null");
      }

      InstanceConfig config = new InstanceConfig(currentData);
      config.setInstanceEnabled(enabled);
      return config.getRecord();
    }
  }, AccessOption.PERSISTENT);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@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);

  boolean success = false;
  switch (type) {
  case CURRENTSTATES:
    success = _groupCommit.commit(_baseDataAccessor, options, path, value.getRecord(), true);
    break;
  case STATUSUPDATES:
    if (LOG.isTraceEnabled()) {
      LOG.trace("Update status. path: " + key.getPath() + ", record: " + value.getRecord());
    }
    break;
  default:
    success = _baseDataAccessor.update(path, updater, options);
    break;
  }
  return success;
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
public static void createOrReplace(ZkClient client, String path, final ZNRecord record,
    final boolean persistent) {
  int retryCount = 0;
  while (retryCount < RETRYLIMIT) {
    try {
      if (client.exists(path)) {
        DataUpdater<Object> updater = new DataUpdater<Object>() {
          @Override
          public Object update(Object currentData) {
            return record;
          }
        };
        client.updateDataSerialized(path, updater);
      } else {
        CreateMode mode = (persistent) ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL;
        client.create(path, record, mode);
      }
      break;
    } catch (Exception e) {
      retryCount = retryCount + 1;
      logger.warn("Exception trying to createOrReplace " + path + " Exception:" + e.getMessage()
          + ". Will retry.");
    }
  }
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
public static void subtract(ZkClient client, String path, final ZNRecord recordTosubtract) {
  int retryCount = 0;
  while (retryCount < RETRYLIMIT) {
    try {
      if (client.exists(path)) {
        DataUpdater<ZNRecord> updater = new DataUpdater<ZNRecord>() {
          @Override
          public ZNRecord update(ZNRecord currentData) {
            currentData.subtract(recordTosubtract);
            return currentData;
          }
        };
        client.updateDataSerialized(path, updater);
        break;
      }
    } catch (Exception e) {
      retryCount = retryCount + 1;
      logger.warn("Exception trying to createOrReplace " + path + ". Will retry.", e);
    }
  }

}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * Add an user defined key-value pair data to task level
 *
 * @param manager       a connection to Helix
 * @param job   the name of job
 * @param task  the name of task
 * @param key           the key of key-value pair
 * @param value         the value of key-value pair
 */
protected static void addTaskUserContent(final HelixManager manager, String job,
    final String task, final String key, final String value) {
  String path =
      Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, job, USER_CONTENT_NODE);

  manager.getHelixPropertyStore().update(path, new DataUpdater<ZNRecord>() {
    @Override public ZNRecord update(ZNRecord znRecord) {
      if (znRecord.getMapField(task) == null) {
        znRecord.setMapField(task, new HashMap<String, String>());
      }
      znRecord.getMapField(task).put(key, value);
      return znRecord;
    }
  }, AccessOption.PERSISTENT);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@Override
public boolean update(String path, DataUpdater<T> updater, int options) {
  if (_fallbackStore == null) {
    return super.update(path, updater, options);
  } else {
    Stat stat = super.getStat(path, options);
    if (stat == null) {
      // create znode at new location with fallback-value
      T fallbackValue = _fallbackStore.get(path, null, options);
      boolean succeed = super.create(path, fallbackValue, AccessOption.PERSISTENT);
      if (!succeed) {
        LOG.error("Can't update " + path + " since there are concurrent updates");
        return false;
      }
    }
    return super.update(path, updater, options);
  }
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * Updates data of an existing znode. The current content of the znode is passed to the {@link DataUpdater} that is
 * passed into this method, which returns the new content. The new content is only written back to ZooKeeper if
 * nobody has modified the given znode in between. If a concurrent change has been detected the new data of the
 * znode is passed to the updater once again until the new contents can be successfully written back to ZooKeeper.
 * 
 * @param <T>
 * @param path The path of the znode.
 * @param updater Updater that creates the new contents.
 */
public <T extends Object> void updateDataSerialized(String path, DataUpdater<T> updater) {
    Stat stat = new Stat();
    boolean retry;
    do {
        retry = false;
        try {
            T oldData = (T) readData(path, stat);
            T newData = updater.update(oldData);
            writeData(path, newData, stat.getVersion());
        } catch (ZkBadVersionException e) {
            retry = true;
        }
    } while (retry);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * 修改节点数据
 * 
 * @param path
 * @param data
 */
public boolean updateNote(String path, final Object data) {
	if (zkClient.exists(path)) {
		zkClient.updateDataSerialized(path, new DataUpdater<Object>() {
			public Object update(Object currentData) {
				currentData = data;
				return currentData;
			}
		});
		return true;
	} else {
		LOGGER.warn("update path = " + path + " exists!");
		return false;
	}
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
@Override
public boolean[] updateChildren(List<String> paths, List<DataUpdater<T>> updaters, int options) {
  final int size = paths.size();
  List<String> serverPaths = prependChroot(paths);

  Cache<T> cache = getCache(serverPaths);
  if (cache != null) {
    try {
      cache.lockWrite();

      List<Stat> setStats = new ArrayList<Stat>();
      boolean[] success = new boolean[size];
      List<List<String>> pathsCreatedList =
          new ArrayList<List<String>>(Collections.<List<String>> nCopies(size, null));
      List<T> updateData =
          _baseAccessor.update(serverPaths, updaters, pathsCreatedList, setStats, options);

      // System.out.println("updateChild: ");
      // for (T data : updateData)
      // {
      // System.out.println(data);
      // }

      for (int i = 0; i < size; i++) {
        success[i] = (updateData.get(i) != null);
        updateCache(cache, pathsCreatedList.get(i), success[i], serverPaths.get(i),
            updateData.get(i), setStats.get(i));
      }
      return success;
    } finally {
      cache.unlockWrite();
    }
  }

  // no cache
  return _baseAccessor.updateChildren(serverPaths, updaters, options);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * async update
 */
@Override
public boolean[] updateChildren(List<String> paths, List<DataUpdater<T>> updaters, int options) {

  List<T> updateData = update(paths, updaters, null, null, options);
  boolean[] success = new boolean[paths.size()]; // init to false
  for (int i = 0; i < paths.size(); i++) {
    T data = updateData.get(i);
    success[i] = (data != null);
  }
  return success;
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
public static void updateIfExists(ZkClient client, String path, final ZNRecord record,
    boolean mergeOnUpdate) {
  if (client.exists(path)) {
    DataUpdater<Object> updater = new DataUpdater<Object>() {
      @Override
      public Object update(Object currentData) {
        return record;
      }
    };
    client.updateDataSerialized(path, updater);
  }
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
public static void createOrMerge(ZkClient client, String path, final ZNRecord record,
    final boolean persistent, final boolean mergeOnUpdate) {
  int retryCount = 0;
  while (retryCount < RETRYLIMIT) {
    try {
      if (client.exists(path)) {
        DataUpdater<ZNRecord> updater = new DataUpdater<ZNRecord>() {
          @Override
          public ZNRecord update(ZNRecord currentData) {
            if (currentData != null && mergeOnUpdate) {
              currentData.merge(record);
              return currentData;
            }
            return record;
          }
        };
        client.updateDataSerialized(path, updater);
      } else {
        CreateMode mode = (persistent) ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL;
        if (record.getDeltaList().size() > 0) {
          ZNRecord value = new ZNRecord(record.getId());
          value.merge(record);
          client.create(path, value, mode);
        } else {
          client.create(path, record, mode);
        }
      }
      break;
    } catch (Exception e) {
      retryCount = retryCount + 1;
      logger.warn("Exception trying to update " + path + " Exception:" + e.getMessage()
          + ". Will retry.");
    }
  }
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
public static void createOrUpdate(ZkClient client, String path, final ZNRecord record,
    final boolean persistent, final boolean mergeOnUpdate) {
  int retryCount = 0;
  while (retryCount < RETRYLIMIT) {
    try {
      if (client.exists(path)) {
        DataUpdater<ZNRecord> updater = new DataUpdater<ZNRecord>() {
          @Override public ZNRecord update(ZNRecord currentData) {
            if (currentData != null && mergeOnUpdate) {
              currentData.update(record);
              return currentData;
            }
            return record;
          }
        };
        client.updateDataSerialized(path, updater);
      } else {
        CreateMode mode = (persistent) ? CreateMode.PERSISTENT : CreateMode.EPHEMERAL;
        client.create(path, record, mode);
      }
      break;
    } catch (Exception e) {
      retryCount = retryCount + 1;
      logger.warn("Exception trying to update " + path + " Exception:" + e.getMessage()
          + ". Will retry.");
    }
  }
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * Helper function to change target state for a given workflow
 */
private void setSingleWorkflowTargetState(String workflow, final TargetState state) {
  LOG.info("Set " + workflow + " to target state " + state);

  WorkflowConfig workflowConfig = TaskUtil.getWorkflowConfig(_accessor, workflow);
  if (workflowConfig == null) {
    LOG.warn("WorkflowConfig for " + workflow + " not found!");
    return;
  }

  WorkflowContext workflowContext = TaskUtil.getWorkflowContext(_propertyStore, workflow);
  if (state != TargetState.DELETE && workflowContext != null &&
      workflowContext.getFinishTime() != WorkflowContext.UNFINISHED) {
    // Should not update target state for completed workflow
    LOG.info("Workflow " + workflow + " is already completed, skip to update its target state "
        + state);
    return;
  }

  DataUpdater<ZNRecord> updater = new DataUpdater<ZNRecord>() {
    @Override public ZNRecord update(ZNRecord currentData) {
      if (currentData != null) {
        currentData.setSimpleField(WorkflowConfig.WorkflowConfigProperty.TargetState.name(),
            state.name());
      } else {
        LOG.warn("TargetState DataUpdater: Fails to update target state. CurrentData is "
            + currentData);
      }
      return currentData;
    }
  };

  PropertyKey workflowConfigKey = TaskUtil.getWorkflowConfigKey(_accessor, workflow);
  _accessor.getBaseDataAccessor()
      .update(workflowConfigKey.getPath(), updater, AccessOption.PERSISTENT);
  RebalanceScheduler.invokeRebalance(_accessor, workflow);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/**
 * Add an user defined key-value pair data to workflow or job level
 *
 * @param manager             a connection to Helix
 * @param workflowJobResource the name of workflow or job
 * @param key                 the key of key-value pair
 * @param value               the value of key-value pair
 */
protected static void addWorkflowJobUserContent(final HelixManager manager,
    String workflowJobResource, final String key, final String value) {
  String path = Joiner.on("/")
      .join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowJobResource, USER_CONTENT_NODE);

  manager.getHelixPropertyStore().update(path, new DataUpdater<ZNRecord>() {
    @Override public ZNRecord update(ZNRecord znRecord) {
      znRecord.setSimpleField(key, value);
      return znRecord;
    }
  }, AccessOption.PERSISTENT);
}
 

import org.I0Itec.zkclient.DataUpdater; //导入依赖的package包/类
/** Remove the job name from the DAG from the queue configuration */
// Job name should be namespaced job name here.
protected static boolean removeJobsFromDag(final HelixDataAccessor accessor, final String workflow,
    final Set<String> jobsToRemove, final boolean maintainDependency) {
  // Now atomically clear the DAG
  DataUpdater<ZNRecord> dagRemover = new DataUpdater<ZNRecord>() {
    @Override
    public ZNRecord update(ZNRecord currentData) {
      if (currentData != null) {
        JobDag jobDag = JobDag.fromJson(
            currentData.getSimpleField(WorkflowConfig.WorkflowConfigProperty.Dag.name()));
        if (jobDag == null) {
          LOG.warn("Could not update DAG for workflow: " + workflow + " JobDag is null.");
          return null;
        }
        for (String job : jobsToRemove) {
          jobDag.removeNode(job, maintainDependency);
        }
        try {
          currentData
              .setSimpleField(WorkflowConfig.WorkflowConfigProperty.Dag.name(), jobDag.toJson());
        } catch (IOException e) {
          throw new IllegalArgumentException(e);
        }
      }
      return currentData;
    }
  };

  String configPath = accessor.keyBuilder().resourceConfig(workflow).getPath();
  if (!accessor.getBaseDataAccessor().update(configPath, dagRemover, AccessOption.PERSISTENT)) {
    LOG.warn("Failed to remove jobs " + jobsToRemove + " from DAG of workflow " + workflow);
    return false;
  }

  return true;
}