XWIKI-23613: When using udp configuration it takes a long time to notice a cluster member went down

Author: tmortagne

xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/internal/jgroups/JGroupsNetworkChannel.java

@@ -37,6 +37,8 @@
 import org.jgroups.JChannel;
 import org.jgroups.Message;
 import org.jgroups.ObjectMessage;
+import org.jgroups.Receiver;
+import org.jgroups.View;
 import org.jgroups.blocks.MessageDispatcher;
 import org.jgroups.blocks.RequestHandler;
 import org.jgroups.blocks.RequestOptions;
@@ -70,7 +72,7 @@
 // It's unfortunately not so easy to reduce the class fan out complexity (which is currently just above the accepted
 // threshold)
 @SuppressWarnings("checkstyle:ClassFanOutComplexity")
-public class JGroupsNetworkChannel implements NetworkChannel
+public class JGroupsNetworkChannel implements NetworkChannel, Receiver
 {
     /**
      * Relative path where to find JGroups channels configurations.
@@ -170,6 +172,7 @@ public void start(String id) throws Exception
 
         // Register the member handler dispatcher
         this.memberIdDispatcher = new MessageDispatcher(this.jchannel, new JGroupsMemberHandler());
+        this.memberIdDispatcher.setReceiver(this);
 
         // Start the channel
         this.jchannel.connect("event");
@@ -192,15 +195,15 @@ public void start(String id) throws Exception
         this.members =
             Map.of(currentMemberId, new JGroupsNetworkMember(this, currentMemberId, this.jchannel.getAddress()));
 
-        // Send the to other members and wait for their ids in response (wait for 1min max).
+        // Send the message to other members and wait for their ids in response (wait for 1min max).
         RspList<String> responses = this.memberIdDispatcher.castMessage(null, new ObjectMessage(null, currentMemberId),
             new RequestOptions(ResponseMode.GET_ALL, 60000));
         for (Map.Entry<Address, Rsp<String>> response : responses.entrySet()) {
             this.membersIdMap.put(response.getKey().toString(), response.getValue().getValue());
         }
 
         // Initialize the channel
-        // It's actually the second time that #onViewChanged is called, but the first time we did not had enough
+        // It's actually the second time that #updateMembers is called, but the first time we did not had enough
         // information for it to do much
         updateMembers(false);
     }
@@ -297,7 +300,24 @@ public void send(Message message) throws Exception
         this.memberIdDispatcher.castMessage(null, new ObjectMessage(null, message), RequestOptions.ASYNC());
     }
 
-    private void updateMembers(boolean notifyChanges)
+    @Override
+    public void viewAccepted(View view)
+    {
+        if (this.membersIdMap == null) {
+            // Not ready
+            return;
+        }
+
+        // Check if a member has been removed
+        for (JGroupsNetworkMember member : this.members.values()) {
+            if (!this.jchannel.getView().containsMember(member.getAddress())) {
+                // At least one member left, update the members cache
+                updateMembers(true);
+            }
+        }
+    }
+
+    private synchronized void updateMembers(boolean notifyChanges)
     {
         if (this.membersIdMap == null) {
             // Not ready