jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520936115
##########
File path:
helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -525,47 +525,51 @@ public void finishTask(MessageTask task) {
private void updateMessageState(Collection<Message> msgsToBeUpdated,
HelixDataAccessor accessor,
String instanceName) {
- if (!msgsToBeUpdated.isEmpty()) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> updateMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : msgsToBeUpdated) {
- updateMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- /**
- * We use the updater to avoid race condition between writing message
to zk as READ state and removing message after ST is done
- * If there is no message at this path, meaning the message is removed
so we do not write the message
- */
- updaters.add(currentData -> {
- if (currentData == null) {
- LOG.warn(
- "Message {} targets at {} has already been removed before it
is set as READ on instance {}",
- msg.getId(), msg.getTgtName(), instanceName);
- return null;
- }
- return msg.getRecord();
- });
- }
- accessor.updateChildren(updateMsgPaths, updaters,
AccessOption.PERSISTENT);
+ if (msgsToBeUpdated.isEmpty()) {
+ return;
+ }
+
+ Builder keyBuilder = accessor.keyBuilder();
+ List<Message> updateMsgs = new ArrayList<>();
+ List<String> updateMsgPaths = new ArrayList<>();
+ List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
+ for (Message msg : msgsToBeUpdated) {
+ updateMsgs.add(msg);
+ updateMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
+ /**
+ * We use the updater to avoid race condition between writing message to
zk as READ state and removing message after ST is done
+ * If there is no message at this path, meaning the message is removed
so we do not write the message
+ */
+ updaters.add(currentData -> {
+ if (currentData == null) {
+ LOG.warn(
+ "Message {} targets at {} has already been removed before it is
set as READ on instance {}",
+ msg.getId(), msg.getTgtName(), instanceName);
+ return null;
+ }
+ return msg.getRecord();
+ });
}
+ boolean[] updateResults =
+ accessor.updateChildren(updateMsgPaths, updaters,
AccessOption.PERSISTENT);
+ boolean isMessageUpdatedAsNew = false;
// Note that only cache the known message Ids after the update to ZK is
successfully done.
// This is to avoid inconsistent cache.
-
- // if a message in "NEW" state is updated, then we might need to process
it soon.
- boolean isNewMessageUpdated = false;
- for (Message msg : msgsToBeUpdated) {
+ for (int i = 0; i < updateMsgs.size(); i++) {
+ Message msg = updateMsgs.get(i);
if (msg.getMsgState().equals(MessageState.NEW)) {
- isNewMessageUpdated = true;
- // If a message is still "NEW", it is not a known message. The message
may not be able to
- // processed now in an expected way.
- } else {
- // else, cache the known messages.
+ // If a message is updated as NEW state, then we might need to process
it again soon.
+ isMessageUpdatedAsNew = true;
+ // And it shall not be treated as a known messages.
+ } else if (updateResults[i]) {
Review comment:
On my second thought, I think this is dangerous since changed the
behavior unnecessarily. Let me just log in this case instead of modifying the
original 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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
