junkaixue commented on code in PR #2770:
URL: https://github.com/apache/helix/pull/2770#discussion_r1509634366
##########
helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java:
##########
@@ -485,6 +487,56 @@ public AccessResult doUpdate(String path, DataUpdater<T>
updater, int options) {
return result;
}
+ /**
+ * transactional sync set
+ */
+ @Override
+ public boolean multiSet(Map<String, DataUpdater<T>> updaterByPath) {
+ AccessResult result = doMultiSet(updaterByPath);
+ return result._retCode == RetCode.OK;
+ }
+
+ private AccessResult doMultiSet(Map<String, DataUpdater<T>> updaterByPath) {
+ AccessResult result = new AccessResult();
+ boolean retry;
+ do {
+ retry = false;
+ try {
+ List<Op> ops = new ArrayList<>();
+ for (Map.Entry<String, DataUpdater<T>> entry :
updaterByPath.entrySet()) {
+ String path = entry.getKey();
+ DataUpdater<T> updater = entry.getValue();
+ Stat readStat = new Stat();
+ T oldData = (T) _zkClient.readData(path, readStat);
+ T newData = updater.update(oldData);
+ if (newData != null) {
+ ops.add(Op.setData(path, _zkClient.serialize(newData, path),
readStat.getVersion()));
+ }
+ }
+ for (OpResult opResult : _zkClient.multi(ops)) {
+ if (opResult instanceof OpResult.SetDataResult) {
+ DataTree.copyStat(((OpResult.SetDataResult) opResult).getStat(),
result._stat);
+ } else if (opResult instanceof OpResult.ErrorResult) {
+ if (((OpResult.ErrorResult) opResult).getErr() ==
Code.BADVERSION.intValue()) {
Review Comment:
Does this contain the part of update failure due to version mismatch?
##########
helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java:
##########
@@ -485,6 +487,56 @@ public AccessResult doUpdate(String path, DataUpdater<T>
updater, int options) {
return result;
}
+ /**
+ * transactional sync set
+ */
+ @Override
+ public boolean multiSet(Map<String, DataUpdater<T>> updaterByPath) {
+ AccessResult result = doMultiSet(updaterByPath);
+ return result._retCode == RetCode.OK;
+ }
+
+ private AccessResult doMultiSet(Map<String, DataUpdater<T>> updaterByPath) {
+ AccessResult result = new AccessResult();
+ boolean retry;
+ do {
+ retry = false;
+ try {
+ List<Op> ops = new ArrayList<>();
+ for (Map.Entry<String, DataUpdater<T>> entry :
updaterByPath.entrySet()) {
+ String path = entry.getKey();
+ DataUpdater<T> updater = entry.getValue();
+ Stat readStat = new Stat();
+ T oldData = (T) _zkClient.readData(path, readStat);
+ T newData = updater.update(oldData);
+ if (newData != null) {
+ ops.add(Op.setData(path, _zkClient.serialize(newData, path),
readStat.getVersion()));
+ }
+ }
+ for (OpResult opResult : _zkClient.multi(ops)) {
+ if (opResult instanceof OpResult.SetDataResult) {
+ DataTree.copyStat(((OpResult.SetDataResult) opResult).getStat(),
result._stat);
+ } else if (opResult instanceof OpResult.ErrorResult) {
+ if (((OpResult.ErrorResult) opResult).getErr() ==
Code.BADVERSION.intValue()) {
+ retry = true;
+ } else {
+ LOG.error("Failed to update path: " + updaterByPath.keySet());
+ result._retCode = RetCode.ERROR;
+ return result;
+ }
+ }
+ }
+ } catch (Exception e) {
Review Comment:
Would be good to specific the failure exceptions to handle different logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
