Fixes #1802 - messages intended for instances that are no longer in the cluster (#1951)
In MessageGenerationPhase.java, - process() method populates the list of live instances from cache.
But while generateMessage() method has the sessionIdMap information, it still goes through partition/resource/instance map without checking if instance is still part of the cluster or not.
It is possible that cache has stale entry but that logic needs to be worked separately. But while generating message, we should check if the instance is still there.
So this is a simple change. We need to still look further if cache is getting invalidated properly.
To make sure that the cache properly is handled/refreshed under instance being replaced or deletion - have filled another bug: #1956
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
index 98313b9..7981302 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
@@ -193,7 +193,7 @@
}
}
- if (pendingMessage != null && shouldCleanUpPendingMessage(pendingMessage, currentState,
+ if (shouldCleanUpPendingMessage(pendingMessage, sessionIdMap, instanceName, currentState,
currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
logAndAddToCleanUp(messagesToCleanUp, pendingMessage, instanceName, resourceName,
partition, currentState, PENDING_MESSAGE);
@@ -203,7 +203,8 @@
// staleMessage can be simple or batch mode
if ((System.currentTimeMillis() - currentStateOutput
.getEndTime(resourceName, partition, instanceName) > DEFAULT_OBSELETE_MSG_PURGE_DELAY)
- && staleMessage.getResourceName().equals(resourceName) && (
+ && staleMessage.getResourceName().equals(resourceName) && sessionIdMap
+ .containsKey(instanceName) && (
staleMessage.getPartitionName().equals(partition.getPartitionName()) || (
staleMessage.getBatchMessageMode() && staleMessage.getPartitionNames()
.contains(partition.getPartitionName())))) {
@@ -404,9 +405,9 @@
});
}
- private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentState,
- Long currentStateTransitionEndTime) {
- if (pendingMsg == null) {
+ private boolean shouldCleanUpPendingMessage(Message pendingMsg, Map<String, String> sessionIdMap,
+ String instanceName, String currentState, Long currentStateTransitionEndTime) {
+ if (pendingMsg == null || !sessionIdMap.containsKey(instanceName)) {
return false;
}
if (currentState.equalsIgnoreCase(pendingMsg.getToState())) {
