Fix race condition in ChannelCoordinator startup: wait for receiver readiness before reading localMember

java/org/apache/catalina/tribes/ChannelReceiver.java

@@ -17,6 +17,7 @@
 package org.apache.catalina.tribes;
 
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The <code>ChannelReceiver</code> interface is the data receiver component at the bottom layer, the IO layer (for
@@ -29,6 +30,11 @@ public interface ChannelReceiver extends Heartbeat {
      */
     int MAX_UDP_SIZE = 65535;
 
+    /**
+     * Default timeout in milliseconds for {@link #waitForReady(long, TimeUnit)}.
+     */
+    long DEFAULT_READY_TIMEOUT_MS = 5000;
+
     /**
      * Start listening for incoming messages on the host/port
      *
@@ -41,6 +47,23 @@ public interface ChannelReceiver extends Heartbeat {
      */
     void stop();
 
+    /**
+     * Wait until the receiver is ready to accept connections, or the timeout expires.
+     * <p>
+     * The default implementation returns immediately, preserving backward compatibility
+     * for receivers that do not implement readiness signaling. Implementations that
+     * start background listener threads should override this method to block until
+     * the listener thread has entered its accept/select loop.
+     *
+     * @param timeout the maximum time to wait
+     * @param unit the time unit of the timeout argument
+     * @return {@code true} if the receiver is ready; {@code false} if the timeout elapsed
+     * @throws InterruptedException if the current thread is interrupted while waiting
+     */
+    default boolean waitForReady(long timeout, TimeUnit unit) throws InterruptedException {
+        return true;
+    }
+
     /**
      * String representation of the IPv4 or IPv6 address that this host is listening to.
      *

java/org/apache/catalina/tribes/group/ChannelCoordinator.java

@@ -16,6 +16,8 @@
  */
 package org.apache.catalina.tribes.group;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.catalina.tribes.Channel;
 import org.apache.catalina.tribes.ChannelException;
 import org.apache.catalina.tribes.ChannelMessage;
@@ -160,7 +162,22 @@ protected synchronized void internalStart(int svc) throws ChannelException {
                 clusterReceiver.setMessageListener(this);
                 clusterReceiver.setChannel(getChannel());
                 clusterReceiver.start();
-                // synchronize, big time FIXME
+                // Wait for the receiver's background thread to enter the listen loop
+                // before reading the local member. Without this synchronization, there
+                // is a race window where start() has returned but the listener thread
+                // has not yet initialized, potentially causing getLocalMember() to
+                // observe an incomplete or null member state.
+                try {
+                    boolean ready = clusterReceiver.waitForReady(
+                            ChannelReceiver.DEFAULT_READY_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+                    if (!ready) {
+                        throw new ChannelException(sm.getString("channelCoordinator.receiverNotReady",
+                                Long.toString(ChannelReceiver.DEFAULT_READY_TIMEOUT_MS)));
+                    }
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new ChannelException(sm.getString("channelCoordinator.receiverWaitInterrupted"), ie);
+                }
                 Member localMember = getChannel().getLocalMember(false);
                 if (localMember instanceof StaticMember staticMember) {
                     // static member

java/org/apache/catalina/tribes/group/LocalStrings.properties

@@ -16,6 +16,8 @@
 channelCoordinator.alreadyStarted=Channel already started for level:[{0}]
 channelCoordinator.invalid.startLevel=Invalid start level, valid levels are:SND_RX_SEQ,SND_TX_SEQ,MBR_TX_SEQ,MBR_RX_SEQ
 channelCoordinator.invalidState.notStopped=Configuration may not be changed until the channel has been fully stopped
+channelCoordinator.receiverNotReady=Channel receiver did not become ready within [{0}] ms during startup.
+channelCoordinator.receiverWaitInterrupted=Interrupted while waiting for channel receiver to become ready during startup.
 
 groupChannel.listener.alreadyExist=Listener already exists:[{0}][{1}]
 groupChannel.noDestination=No destination given

java/org/apache/catalina/tribes/transport/nio/NioReceiver.java

@@ -30,6 +30,8 @@
 import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.catalina.tribes.io.ObjectReader;
@@ -59,6 +61,13 @@ public class NioReceiver extends ReceiverBase implements Runnable, NioReceiverMB
     private ServerSocketChannel serverChannel = null;
     private DatagramChannel datagramChannel = null;
 
+    /**
+     * Latch that counts down when the listener thread has entered the select loop.
+     * A count of 0 means the receiver is ready (or not started). A count of 1 means
+     * the listener thread is still initializing.
+     */
+    private volatile CountDownLatch readyLatch = new CountDownLatch(0);
+
     /**
      * Queue of events to be processed by the selector thread.
      */
@@ -92,6 +101,9 @@ public void start() throws IOException {
         try {
             getBind();
             bind();
+            // Create a fresh latch with count 1 before launching the listener thread.
+            // The latch will be counted down in listen() after setListen(true).
+            readyLatch = new CountDownLatch(1);
             String channelName = "";
             if (getChannel().getName() != null) {
                 channelName = "[" + getChannel().getName() + "]";
@@ -100,6 +112,8 @@ public void start() throws IOException {
             t.setDaemon(true);
             t.start();
         } catch (Exception e) {
+            // Reset latch to avoid blocking callers if start fails
+            readyLatch = new CountDownLatch(0);
             log.fatal(sm.getString("nioReceiver.start.fail"), e);
             if (e instanceof IOException) {
                 throw (IOException) e;
@@ -109,6 +123,19 @@ public void start() throws IOException {
         }
     }
 
+    /**
+     * Wait until the receiver's listener thread has entered the select loop.
+     *
+     * @param timeout the maximum time to wait
+     * @param unit the time unit of the timeout argument
+     * @return {@code true} if the receiver is ready; {@code false} if the timeout elapsed
+     * @throws InterruptedException if the current thread is interrupted while waiting
+     */
+    @Override
+    public boolean waitForReady(long timeout, TimeUnit unit) throws InterruptedException {
+        return readyLatch.await(timeout, unit);
+    }
+
     @Override
     public AbstractRxTask createRxTask() {
         NioReplicationTask thread = new NioReplicationTask(this, this);
@@ -309,6 +336,10 @@ protected void listen() throws Exception {
 
         setListen(true);
 
+        // Signal that the listener thread has entered the listen loop and is
+        // ready to accept connections. This must happen after setListen(true).
+        readyLatch.countDown();
+
         // Avoid NPEs if selector is set to null on stop.
         Selector selector = this.selector.get();
 
@@ -399,6 +430,9 @@ protected void listen() throws Exception {
      */
     protected void stopListening() {
         setListen(false);
+        // Reset the latch so that a subsequent start() can create a fresh one.
+        // A count of 0 means "not waiting" / "already ready".
+        readyLatch = new CountDownLatch(0);
         Selector selector = this.selector.get();
         if (selector != null) {
             try {