You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by GitBox <gi...@apache.org> on 2020/11/06 19:17:44 UTC
[GitHub] [helix] jiajunwang opened a new pull request #1514: Add default message handling retry count for state transition messages.
jiajunwang opened a new pull request #1514:
URL: https://github.com/apache/helix/pull/1514
### Issues
- [X] My PR addresses the following Helix issues and references them in the PR description:
Resolves #1513
### Description
- [X] Here are some details about my PR, including screenshots of any UI changes:
The retry count will be counted when message handler creation fails or state transition times out.
In addition, this PR also improves several ambiguous behaviors that may block a normal retry.
1. The cache of known messages list in HelixTaskExecutor may not be updated correctly if a message is going to be retried.
2. The retry count is not strictly followed in some conditions.
3. The to-be-retried message won't be automatically read when there are no other new message changes.
Finally, improve the tests to cover the new changes.
### Tests
- [X] The following tests are written for this issue:
TestStateTransitionAppFailureHandling
- [X] The following is the result of the "mvn test" command on the appropriate module:
[ERROR] Failures:
[ERROR] TestTaskRebalancerFailover.test ยป ThreadTimeout Method org.testng.internal.Tes...
[INFO]
[ERROR] Tests run: 1247, Failures: 1, Errors: 0, Skipped: 0
[INFO]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 01:36 h
[INFO] Finished at: 2020-11-06T00:46:51-08:00
[INFO] ------------------------------------------------------------------------
Rerun the test and it will pass. But it is not stable due to issue #1512
### Documentation (Optional)
- In case of new functionality, my PR adds documentation in the following wiki page:
(Link the GitHub wiki you added)
### Commits
- My commits all reference appropriate Apache Helix GitHub issues in their subject lines. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Code Quality
- My diff has been formatted using helix-style.xml
(helix-style-intellij.xml if IntelliJ IDE is used)
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520783510
##########
File path: helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
##########
@@ -563,23 +564,42 @@ public void testCreateHandlerException() throws InterruptedException {
HelixManager manager = new MockClusterManager();
HelixDataAccessor dataAccessor = manager.getHelixDataAccessor();
PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder();
-
- TestMessageHandlerFactory factory = new TestMessageHandlerFactory();
- for (String type : factory.getMessageTypes()) {
- executor.registerMessageHandlerFactory(type, factory);
- }
-
NotificationContext changeContext = new NotificationContext(manager);
+ TestMessageHandlerFactory factory = new TestMessageHandlerFactory();
+ // Sending message without registering the factory.
+ // The message won't be processed since creating handler returns null.
int nMsgs1 = 5;
+ List<Message> msgList = new ArrayList<>();
for (int i = 0; i < nMsgs1; i++) {
Message msg = new Message(factory.getMessageTypes().get(0), UUID.randomUUID().toString());
msg.setTgtSessionId(manager.getSessionId());
msg.setTgtName("Localhost_1123");
msg.setSrcName("127.101.1.23_2234");
msg.setCorrelationId(UUID.randomUUID().toString());
dataAccessor.setProperty(keyBuilder.message(manager.getInstanceName(), msg.getMsgId()), msg);
+ msgList.add(msg);
+ }
+
+ changeContext.setChangeType(HelixConstants.ChangeType.MESSAGE);
+ executor.onMessage(manager.getInstanceName(), Collections.emptyList(), changeContext);
+
+ Thread.sleep(1000);
Review comment:
We decided to get rid of sleep, right?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java
##########
@@ -0,0 +1,203 @@
+package org.apache.helix.integration.paticipant;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.TestHelper;
+import org.apache.helix.controller.stages.MessageGenerationPhase;
+import org.apache.helix.integration.common.ZkStandAloneCMTestBase;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.mock.participant.MockMSStateModel;
+import org.apache.helix.mock.participant.MockTransition;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestStateTransitionAppFailureHandling extends ZkStandAloneCMTestBase {
+ private static Logger LOG = LoggerFactory.getLogger(TestStateTransitionAppFailureHandling.class);
+ private final static int REPLICAS = 3;
+
+ @Override
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ super.beforeClass();
+ // Clean up the resource that is created in the super cluster beforeClass method.
+ _gSetupTool.dropResourceFromCluster(CLUSTER_NAME, TEST_DB);
+ _clusterVerifier.verifyByPolling();
+ }
+
+ public static class RetryStateModelFactory extends StateModelFactory<MockMSStateModel> {
+ int _retryCountUntilSucceed;
+
+ public RetryStateModelFactory(int retryCountUntilSucceed) {
+ _retryCountUntilSucceed = retryCountUntilSucceed;
+ }
+
+ public int getRemainingRetryCountUntilSucceed() {
+ return _retryCountUntilSucceed;
+ }
+
+ @Override
+ public MockMSStateModel createNewStateModel(String resource, String stateUnitKey) {
+ if (_retryCountUntilSucceed > 0) {
+ _retryCountUntilSucceed--;
+ throw new HelixException("You Shall Not PASS!!!");
+ } else {
+ return new MockMSStateModel(new MockTransition());
+ }
+ }
+ }
+
+ @Test
+ public void testSTHandlerInitFailureRetry() throws Exception {
+ int retryCountUntilSucceed =
+ Integer.MAX_VALUE; // ensure the retry count is large so the message retry will fail.
+ Map<String, RetryStateModelFactory> retryFactoryMap = resetParticipants(retryCountUntilSucceed);
+
+ _gSetupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
+ _gSetupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, REPLICAS);
+
+ HelixDataAccessor accessor = _controller.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ // Verify and wait until all messages have been retried and failed.
+ Map<String, List<Message>> partitionMessageMap = new HashMap<>();
+ Assert.assertTrue(TestHelper.verify(() -> {
+ int totalMessageCount = 0;
+ for (int i = 0; i < NODE_NR; i++) {
+ String instanceName = _participants[i].getInstanceName();
+ List<Message> messageList = accessor.getProperty(
+ accessor.getChildNames(keyBuilder.messages(instanceName)).stream()
+ .map(childName -> keyBuilder.message(instanceName, childName))
+ .collect(Collectors.toList()), true);
+ for (Message message : messageList) {
+ if (message.getMsgState() != Message.MessageState.UNPROCESSABLE) {
+ return false;
+ }
+ }
+ partitionMessageMap.put(instanceName, messageList);
+ totalMessageCount += messageList.size();
+ }
+ return totalMessageCount == _PARTITIONS * REPLICAS;
+ }, TestHelper.WAIT_DURATION));
+
+ // Verify that the correct numbers of retry has been done on each node.
+ for (String instanceName : partitionMessageMap.keySet()) {
+ List<Message> instanceMessages = partitionMessageMap.get(instanceName);
+ for (Message message : instanceMessages) {
+ Assert.assertTrue(message.getRetryCount() <= 0);
+ Assert.assertEquals(message.getMsgState(), Message.MessageState.UNPROCESSABLE);
+ }
+ // Check if the factory has tried enough times before fail the message.
+ Assert.assertEquals(retryCountUntilSucceed - retryFactoryMap.get(instanceName)
+ .getRemainingRetryCountUntilSucceed(), instanceMessages.size()
+ * MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
+ }
+
+ // Verify that the partition is not initialized.
+ for (int i = 0; i < NODE_NR; i++) {
+ String instanceName = _participants[i].getInstanceName();
+ String sessionId = _participants[i].getSessionId();
+ List<CurrentState> currentStates = accessor.getProperty(
+ accessor.getChildNames(keyBuilder.currentStates(instanceName, sessionId)).stream()
+ .map(childName -> keyBuilder.currentState(instanceName, sessionId, childName))
+ .collect(Collectors.toList()), true);
+ for (CurrentState currentState : currentStates) {
+ Assert.assertTrue(currentState.getPartitionStateMap().isEmpty());
+ }
+ }
+
+ _gSetupTool.dropResourceFromCluster(CLUSTER_NAME, TEST_DB);
+ }
+
+ @Test(dependsOnMethods = "testSTHandlerInitFailureRetry")
+ public void testSTHandlerInitFailureRetrySucceed() {
+ // Make the mock StateModelFactory return handler before last retry. So it will successfully
+ // finish handler initialization.
+ int retryCountUntilSucceed =
+ MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT - 1;
+ Map<String, RetryStateModelFactory> retryFactoryMap = resetParticipants(retryCountUntilSucceed);
+
+ _gSetupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
+ _gSetupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, REPLICAS);
+
+ HelixDataAccessor accessor = _controller.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ // Verify and wait until all messages have been retried and failed.
Review comment:
You mean retried and succeeded finally, right?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] dasahcc commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
dasahcc commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520811867
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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) {
+ 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
Review comment:
nit: an known
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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) {
Review comment:
Why we dont move this check in the previous block? Since it already does the looping of the msgsToBeUpdated.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -70,6 +70,8 @@
.getSystemPropertyAsLong(SystemPropertyKeys.CONTROLLER_MESSAGE_PURGE_DELAY, 60 * 1000);
private final static String PENDING_MESSAGE = "pending message";
private final static String STALE_MESSAGE = "stale message";
+ // TODO: Make the message retry count configurable through the Cluster Config or IdealStates.
+ public final static int DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT = 3;
Review comment:
Is this the same thing as the message attribute of retrying? Otherwise, we need an accurate name since it is the retry for state model creation.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520824923
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -821,32 +843,29 @@ public void onMessage(String instanceName, List<Message> messages,
LOG.error(
"Exception happens when creating Message Handler for message {}. Current remaining retry count is {}.",
message.getMsgId(), remainingRetryCount);
- // Set the message retry count to avoid infinite retrying.
+ // Reduce the message retry count to avoid infinite retrying.
message.setRetryCount(remainingRetryCount - 1);
message.setExecuteSessionId(sessionId);
- // continue processing in the next section where handler object is double-checked.
- }
-
- if (msgHandler == null) {
// Note that we are re-using the retry count of Message that was original designed to control
// timeout retries. So it is not checked before the first try in order to ensure consistent
// behavior. It is possible that we introduce a new behavior for this method. But it requires
// us to split the configuration item so as to avoid confusion.
- if (message.getRetryCount() < 0) {
+ if (message.getRetryCount() <= 0) {
// If no more retry count remains, then mark the message to be UNPROCESSABLE.
- String errorMsg = String
- .format("No available message Handler found!"
- + " Stop processing message %s since it has a negative remaining retry count %d!",
- message.getMsgId(), message.getRetryCount());
+ String errorMsg = String.format("No available message Handler found!"
+ + " Stop processing message %s since it has negative remaining retry count %d!",
Review comment:
Good catch, I forget to change the comment after the code change.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520824143
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -1288,6 +1308,19 @@ private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
}
}
+ private void sendNopMessage(HelixDataAccessor accessor, String instanceName) {
+ try {
+ Message nopMsg = new Message(MessageType.NO_OP, UUID.randomUUID().toString());
+ nopMsg.setSrcName(instanceName);
+ nopMsg.setTgtName(instanceName);
+ accessor
+ .setProperty(accessor.keyBuilder().message(nopMsg.getTgtName(), nopMsg.getId()), nopMsg);
+ LOG.info("Send NO_OP message to " + nopMsg.getTgtName() + ", msgId: " + nopMsg.getId());
+ } catch (Exception e) {
+ LOG.error(e.toString());
Review comment:
I agree, it was copy-pasted. Let me make it complete.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] alirezazamani commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
alirezazamani commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520977896
##########
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:
+1 I was thinking that an additional log for this case would be helpful.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang merged pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang merged pull request #1514:
URL: https://github.com/apache/helix/pull/1514
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520867684
##########
File path: helix-core/src/test/java/org/apache/helix/integration/paticipant/TestStateTransitionAppFailureHandling.java
##########
@@ -0,0 +1,203 @@
+package org.apache.helix.integration.paticipant;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixDefinedState;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.TestHelper;
+import org.apache.helix.controller.stages.MessageGenerationPhase;
+import org.apache.helix.integration.common.ZkStandAloneCMTestBase;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.mock.participant.MockMSStateModel;
+import org.apache.helix.mock.participant.MockTransition;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestStateTransitionAppFailureHandling extends ZkStandAloneCMTestBase {
+ private static Logger LOG = LoggerFactory.getLogger(TestStateTransitionAppFailureHandling.class);
+ private final static int REPLICAS = 3;
+
+ @Override
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ super.beforeClass();
+ // Clean up the resource that is created in the super cluster beforeClass method.
+ _gSetupTool.dropResourceFromCluster(CLUSTER_NAME, TEST_DB);
+ _clusterVerifier.verifyByPolling();
+ }
+
+ public static class RetryStateModelFactory extends StateModelFactory<MockMSStateModel> {
+ int _retryCountUntilSucceed;
+
+ public RetryStateModelFactory(int retryCountUntilSucceed) {
+ _retryCountUntilSucceed = retryCountUntilSucceed;
+ }
+
+ public int getRemainingRetryCountUntilSucceed() {
+ return _retryCountUntilSucceed;
+ }
+
+ @Override
+ public MockMSStateModel createNewStateModel(String resource, String stateUnitKey) {
+ if (_retryCountUntilSucceed > 0) {
+ _retryCountUntilSucceed--;
+ throw new HelixException("You Shall Not PASS!!!");
+ } else {
+ return new MockMSStateModel(new MockTransition());
+ }
+ }
+ }
+
+ @Test
+ public void testSTHandlerInitFailureRetry() throws Exception {
+ int retryCountUntilSucceed =
+ Integer.MAX_VALUE; // ensure the retry count is large so the message retry will fail.
+ Map<String, RetryStateModelFactory> retryFactoryMap = resetParticipants(retryCountUntilSucceed);
+
+ _gSetupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
+ _gSetupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, REPLICAS);
+
+ HelixDataAccessor accessor = _controller.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ // Verify and wait until all messages have been retried and failed.
+ Map<String, List<Message>> partitionMessageMap = new HashMap<>();
+ Assert.assertTrue(TestHelper.verify(() -> {
+ int totalMessageCount = 0;
+ for (int i = 0; i < NODE_NR; i++) {
+ String instanceName = _participants[i].getInstanceName();
+ List<Message> messageList = accessor.getProperty(
+ accessor.getChildNames(keyBuilder.messages(instanceName)).stream()
+ .map(childName -> keyBuilder.message(instanceName, childName))
+ .collect(Collectors.toList()), true);
+ for (Message message : messageList) {
+ if (message.getMsgState() != Message.MessageState.UNPROCESSABLE) {
+ return false;
+ }
+ }
+ partitionMessageMap.put(instanceName, messageList);
+ totalMessageCount += messageList.size();
+ }
+ return totalMessageCount == _PARTITIONS * REPLICAS;
+ }, TestHelper.WAIT_DURATION));
+
+ // Verify that the correct numbers of retry has been done on each node.
+ for (String instanceName : partitionMessageMap.keySet()) {
+ List<Message> instanceMessages = partitionMessageMap.get(instanceName);
+ for (Message message : instanceMessages) {
+ Assert.assertTrue(message.getRetryCount() <= 0);
+ Assert.assertEquals(message.getMsgState(), Message.MessageState.UNPROCESSABLE);
+ }
+ // Check if the factory has tried enough times before fail the message.
+ Assert.assertEquals(retryCountUntilSucceed - retryFactoryMap.get(instanceName)
+ .getRemainingRetryCountUntilSucceed(), instanceMessages.size()
+ * MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT);
+ }
+
+ // Verify that the partition is not initialized.
+ for (int i = 0; i < NODE_NR; i++) {
+ String instanceName = _participants[i].getInstanceName();
+ String sessionId = _participants[i].getSessionId();
+ List<CurrentState> currentStates = accessor.getProperty(
+ accessor.getChildNames(keyBuilder.currentStates(instanceName, sessionId)).stream()
+ .map(childName -> keyBuilder.currentState(instanceName, sessionId, childName))
+ .collect(Collectors.toList()), true);
+ for (CurrentState currentState : currentStates) {
+ Assert.assertTrue(currentState.getPartitionStateMap().isEmpty());
+ }
+ }
+
+ _gSetupTool.dropResourceFromCluster(CLUSTER_NAME, TEST_DB);
+ }
+
+ @Test(dependsOnMethods = "testSTHandlerInitFailureRetry")
+ public void testSTHandlerInitFailureRetrySucceed() {
+ // Make the mock StateModelFactory return handler before last retry. So it will successfully
+ // finish handler initialization.
+ int retryCountUntilSucceed =
+ MessageGenerationPhase.DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT - 1;
+ Map<String, RetryStateModelFactory> retryFactoryMap = resetParticipants(retryCountUntilSucceed);
+
+ _gSetupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
+ _gSetupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, REPLICAS);
+
+ HelixDataAccessor accessor = _controller.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ // Verify and wait until all messages have been retried and failed.
Review comment:
Fixed.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520822104
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -1288,6 +1308,19 @@ private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
}
}
+ private void sendNopMessage(HelixDataAccessor accessor, String instanceName) {
Review comment:
Oh yeah, good catch. I changed the design since during the test I find that would be a very bad way if we leave the message there and pray for some new messages comming.
I will modify the description.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520827174
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -70,6 +70,8 @@
.getSystemPropertyAsLong(SystemPropertyKeys.CONTROLLER_MESSAGE_PURGE_DELAY, 60 * 1000);
private final static String PENDING_MESSAGE = "pending message";
private final static String STALE_MESSAGE = "stale message";
+ // TODO: Make the message retry count configurable through the Cluster Config or IdealStates.
+ public final static int DEFAULT_STATE_TRANSITION_MESSAGE_RETRY_COUNT = 3;
Review comment:
It is the same. So in theory it will be applied to the timeout condition too. Although ST message does not have a timeout setup for now, we may have it in the future. So make this name too explicit may be confusing as well. I propose to make it a general retry count for ST messages. But it is debatable. Let me know what's your thought.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520829622
##########
File path: helix-core/src/test/java/org/apache/helix/messaging/handling/TestHelixTaskExecutor.java
##########
@@ -563,23 +564,42 @@ public void testCreateHandlerException() throws InterruptedException {
HelixManager manager = new MockClusterManager();
HelixDataAccessor dataAccessor = manager.getHelixDataAccessor();
PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder();
-
- TestMessageHandlerFactory factory = new TestMessageHandlerFactory();
- for (String type : factory.getMessageTypes()) {
- executor.registerMessageHandlerFactory(type, factory);
- }
-
NotificationContext changeContext = new NotificationContext(manager);
+ TestMessageHandlerFactory factory = new TestMessageHandlerFactory();
+ // Sending message without registering the factory.
+ // The message won't be processed since creating handler returns null.
int nMsgs1 = 5;
+ List<Message> msgList = new ArrayList<>();
for (int i = 0; i < nMsgs1; i++) {
Message msg = new Message(factory.getMessageTypes().get(0), UUID.randomUUID().toString());
msg.setTgtSessionId(manager.getSessionId());
msg.setTgtName("Localhost_1123");
msg.setSrcName("127.101.1.23_2234");
msg.setCorrelationId(UUID.randomUUID().toString());
dataAccessor.setProperty(keyBuilder.message(manager.getInstanceName(), msg.getMsgId()), msg);
+ msgList.add(msg);
+ }
+
+ changeContext.setChangeType(HelixConstants.ChangeType.MESSAGE);
+ executor.onMessage(manager.getInstanceName(), Collections.emptyList(), changeContext);
+
+ Thread.sleep(1000);
Review comment:
I think removing it will be fine. It was from some old code. Let me have a try.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520834813
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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) {
Review comment:
Good point, I don't remember any limitation makes me do this. Let me have a try : )
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520824549
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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;
Review comment:
Good name, let me change it.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520822104
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -1288,6 +1308,19 @@ private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
}
}
+ private void sendNopMessage(HelixDataAccessor accessor, String instanceName) {
Review comment:
Oh yeah, good catch. I changed the design since during the test I find that would be a very bad way if we leave the message there and pray for some new messages coming.
The PR description is about what has been fixed. Let me make it more clear.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on pull request #1514:
URL: https://github.com/apache/helix/pull/1514#issuecomment-725766096
Thank @dasahcc for the approval. I will merge the code soon.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on pull request #1514:
URL: https://github.com/apache/helix/pull/1514#issuecomment-725024414
Rebased to the current master and retested.
[INFO] Results:
[INFO]
[INFO] Tests run: 1247, Failures: 0, Errors: 0, Skipped: 0
[INFO]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 01:29 h
[INFO] Finished at: 2020-11-10T14:54:05-08:00
[INFO] ------------------------------------------------------------------------
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520777177
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -1288,6 +1308,19 @@ private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
}
}
+ private void sendNopMessage(HelixDataAccessor accessor, String instanceName) {
Review comment:
This function will trigger another event and retry the message, right? Why in the PR it said the message will not be automatically retried?
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -1288,6 +1308,19 @@ private void removeMessageFromZK(HelixDataAccessor accessor, Message message,
}
}
+ private void sendNopMessage(HelixDataAccessor accessor, String instanceName) {
+ try {
+ Message nopMsg = new Message(MessageType.NO_OP, UUID.randomUUID().toString());
+ nopMsg.setSrcName(instanceName);
+ nopMsg.setTgtName(instanceName);
+ accessor
+ .setProperty(accessor.keyBuilder().message(nopMsg.getTgtName(), nopMsg.getId()), nopMsg);
+ LOG.info("Send NO_OP message to " + nopMsg.getTgtName() + ", msgId: " + nopMsg.getId());
+ } catch (Exception e) {
+ LOG.error(e.toString());
Review comment:
Can we have a more detailed error message here?
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -821,32 +843,29 @@ public void onMessage(String instanceName, List<Message> messages,
LOG.error(
"Exception happens when creating Message Handler for message {}. Current remaining retry count is {}.",
message.getMsgId(), remainingRetryCount);
- // Set the message retry count to avoid infinite retrying.
+ // Reduce the message retry count to avoid infinite retrying.
message.setRetryCount(remainingRetryCount - 1);
message.setExecuteSessionId(sessionId);
- // continue processing in the next section where handler object is double-checked.
- }
-
- if (msgHandler == null) {
// Note that we are re-using the retry count of Message that was original designed to control
// timeout retries. So it is not checked before the first try in order to ensure consistent
// behavior. It is possible that we introduce a new behavior for this method. But it requires
// us to split the configuration item so as to avoid confusion.
- if (message.getRetryCount() < 0) {
+ if (message.getRetryCount() <= 0) {
// If no more retry count remains, then mark the message to be UNPROCESSABLE.
- String errorMsg = String
- .format("No available message Handler found!"
- + " Stop processing message %s since it has a negative remaining retry count %d!",
- message.getMsgId(), message.getRetryCount());
+ String errorMsg = String.format("No available message Handler found!"
+ + " Stop processing message %s since it has negative remaining retry count %d!",
Review comment:
"negative" doesn't match if condition.
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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) {
+ 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.
+ _knownMessageIds.add(msg.getId());
+ }
+ }
+ if (isNewMessageUpdated) {
+ // Sending a NO-OP message to trigger another message callback to re-process the New and
+ // updated messsages.
Review comment:
same here: what does "new and updated" mean?
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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;
Review comment:
The naming here makes me a bit confusing. This message is updated as NEW because it is cannot be processed right? Maybe call it MessageUpdatedAsNew or something.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1514: Add default message handling retry count for state transition messages.
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1514:
URL: https://github.com/apache/helix/pull/1514#discussion_r520853964
##########
File path: helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -521,29 +523,51 @@ public void finishTask(MessageTask task) {
}
}
- private void updateMessageState(List<Message> readMsgs, HelixDataAccessor accessor,
+ private void updateMessageState(Collection<Message> msgsToBeUpdated, HelixDataAccessor accessor,
String instanceName) {
- Builder keyBuilder = accessor.keyBuilder();
- List<String> readMsgPaths = new ArrayList<>();
- List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
- for (Message msg : readMsgs) {
- readMsgPaths.add(msg.getKey(keyBuilder, instanceName).getPath());
- _knownMessageIds.add(msg.getId());
- /**
- * 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();
- });
+ 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);
+ }
+
+ // 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) {
Review comment:
It turns out that I can move the check to the loop above, but I cannot remove the loop here after the update call. For the reason, please check the comment in code. Paste it here as well.
> // Note that only cache the known message Ids after the update to ZK is successfully done.
> // This is to avoid inconsistent cache.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org