0003305: Removing a channel can cause errors during push/pull

# Conflicts: # symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/NodeCommunicationService.java
JumpMind · Nov 18, 2020 · 1f47b95 · 1f47b95
1 parent c34d48f
commit 1f47b95
Showing 1 changed file with 18 additions and 1 deletion.
diff --git a/...tric-core/src/main/java/org/jumpmind/symmetric/service/impl/NodeCommunicationService.java b/...tric-core/src/main/java/org/jumpmind/symmetric/service/impl/NodeCommunicationService.java
@@ -38,6 +38,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DateUtils;
 import org.jumpmind.db.sql.ISqlRowMapper;
 import org.jumpmind.db.sql.Row;
@@ -223,7 +224,7 @@ public List<NodeCommunication> list(CommunicationType communicationType) {
 
             NodeCommunication nodeToCommunicateWith = nodesToCommunicateWithListMap.get(nodeCommunication.getNodeId()); 
             Node node = nodeToCommunicateWith != null ? nodeToCommunicateWith.getNode() : null;
-            if (node == null) {
+            if (node == null || (! isQueueValid(nodeCommunication))) {
                 delete(nodeCommunication);
                 it.remove();
             }
@@ -250,6 +251,22 @@ public List<NodeCommunication> list(CommunicationType communicationType) {
         return communicationRows;
     }
 
+    private boolean isQueueValid(NodeCommunication nodeCommunication) {
+        boolean ret = false;
+        String queue = nodeCommunication.getQueue();
+        Map<String, Channel> channels = configurationService.getChannels(false);
+        for(String key : channels.keySet()) {
+            Channel channel = channels.get(key);
+            if (channel != null) {
+                if (StringUtils.equalsIgnoreCase(queue,  channel.getQueue())) {
+                    ret = true;
+                    break;
+                }
+            }
+        }
+        return ret;
+    }
+
     protected List<String> getNodeIdsWithUnsentCount() {
         return sqlTemplate.query(getSql("selectNodeIdsWithUnsentBatchsSql"),
                 new StringMapper(), OutgoingBatch.Status.ER.name(), OutgoingBatch.Status.NE.name(), OutgoingBatch.Status.QY.name(),