simplify health-checker callback ports to only use ports known by the client

Author: Myron Scott

common/src/main/java/com/tc/net/protocol/tcm/CommunicationsManagerImpl.java

@@ -99,7 +99,6 @@ public class CommunicationsManagerImpl implements CommunicationsManager {
 
   private final ConnectionHealthChecker                                              connectionHealthChecker;
   private ServerID                                                             serverID                  = ServerID.NULL_ID;
-  private int                                                                  callbackPort              = TransportHandshakeMessage.NO_CALLBACK_PORT;
   private final TransportHandshakeErrorHandler                                 handshakeErrHandler;
 
   /**
@@ -276,7 +275,7 @@ public ClientMessageChannel createClientChannel(ProductID productId,
                                                                                      connectionHealthChecker,
                                                                                      connectionManager,
                                                                                      timeout,
-                                                                                     callbackPort, handshakeErrHandler,
+                                                                                     handshakeErrHandler,
                                                                                      reconnectionRejectedHandler
     );
     NetworkStackHarness stackHarness = this.stackHarnessFactory.createClientHarness(transportFactory, rv,
@@ -328,13 +327,6 @@ public MessageChannelInternal createNewChannel(ChannelID id) {
       }
     };
 
-    // XXX: since we don't create multiple listeners per commsMgr, its OK to set
-    // L2's callbackPort here. Otherwise, have interface method and set after starting
-    // commsMgr listener.
-    if (!this.healthCheckerConfig.isCallbackPortListenerNeeded()) {
-      this.callbackPort = addr.getPort();
-    }
-
     final ChannelManagerImpl channelManager = new ChannelManagerImpl(transportDisconnectRemovesChannel, channelFactory);
     return new NetworkListenerImpl(addr, this, channelManager, msgFactory, reuseAddr,
                                    connectionIdFactory, wireProtoMsgSnk, activeProvider, validation);

common/src/main/java/com/tc/net/protocol/tcm/MessageTransportFactoryImpl.java

@@ -27,6 +27,7 @@
 import com.tc.net.protocol.transport.ReconnectionRejectedHandler;
 import com.tc.net.protocol.transport.ServerMessageTransport;
 import com.tc.net.protocol.transport.TransportHandshakeErrorHandler;
+import com.tc.net.protocol.transport.TransportHandshakeMessage;
 import com.tc.net.protocol.transport.TransportHandshakeMessageFactory;
 import com.tc.net.protocol.transport.WireProtocolAdaptorFactory;
 import com.tc.net.protocol.transport.WireProtocolAdaptorFactoryImpl;
@@ -38,21 +39,19 @@ public class MessageTransportFactoryImpl implements MessageTransportFactory {
   private final ConnectionHealthChecker          connectionHealthChecker;
   private final TCConnectionManager              connectionMgr;
   private final int                              timeout;
-  private final int                              callbackport;
   private final TransportHandshakeErrorHandler   defaultHandshakeErrorHandler;
   private final ReconnectionRejectedHandler      reconnectionRejectedHandler;
 
   public MessageTransportFactoryImpl(TransportHandshakeMessageFactory transportMessageFactory,
                                      ConnectionHealthChecker connectionHealthChecker,
                                      TCConnectionManager connectionManager,
-                                     int timeout, int callbackPort,
+                                     int timeout, 
                                      TransportHandshakeErrorHandler defaultHandshakeErrorHandler,
                                      ReconnectionRejectedHandler reconnectionRejectedBehaviour) {
     this.transportMessageFactory = transportMessageFactory;
     this.connectionHealthChecker = connectionHealthChecker;
     this.connectionMgr = connectionManager;
     this.timeout = timeout;
-    this.callbackport = callbackPort;
     this.defaultHandshakeErrorHandler = defaultHandshakeErrorHandler;
     this.reconnectionRejectedHandler = reconnectionRejectedBehaviour;
   }
@@ -61,17 +60,16 @@ public MessageTransportFactoryImpl(TransportHandshakeMessageFactory transportMes
   public ClientMessageTransport createNewTransport() {
     ClientMessageTransport cmt = createClientMessageTransport(
                                                               defaultHandshakeErrorHandler, transportMessageFactory,
-                                                              new WireProtocolAdaptorFactoryImpl(), callbackport);
+                                                              new WireProtocolAdaptorFactoryImpl());
     cmt.addTransportListener(connectionHealthChecker);
     return cmt;
   }
 
   protected ClientMessageTransport createClientMessageTransport(TransportHandshakeErrorHandler handshakeErrorHandler,
                                                                 TransportHandshakeMessageFactory messageFactory,
-                                                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory,
-                                                                int callbackPortNum) {
+                                                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory) {
     return new ClientMessageTransport(this.connectionMgr, handshakeErrorHandler, transportMessageFactory,
-                                      wireProtocolAdaptorFactory, callbackPortNum, this.timeout, reconnectionRejectedHandler);
+                                      wireProtocolAdaptorFactory, this.timeout, reconnectionRejectedHandler);
   }
 
   @Override

common/src/main/java/com/tc/net/protocol/transport/ClientMessageTransport.java

@@ -59,15 +59,14 @@ public class ClientMessageTransport extends MessageTransportBase {
   private CompletableFuture<NetworkStackID>                         opener;
   private CompletableFuture<SynAckMessage>                          waitForSynAckResult;
   private final WireProtocolAdaptorFactory  wireProtocolAdaptorFactory;
-  private final int                         callbackPort;
   private final int                         timeout;
   private final ReconnectionRejectedHandler reconnectionRejectedHandler;
 
   public ClientMessageTransport(TCConnectionManager clientConnectionEstablisher,
                                 TransportHandshakeErrorHandler handshakeErrorHandler,
                                 TransportHandshakeMessageFactory messageFactory,
-                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory, int callbackPort, int timeout) {
-    this(clientConnectionEstablisher, handshakeErrorHandler, messageFactory, wireProtocolAdaptorFactory, callbackPort, timeout,
+                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory, int timeout) {
+    this(clientConnectionEstablisher, handshakeErrorHandler, messageFactory, wireProtocolAdaptorFactory, timeout,
          ReconnectionRejectedHandlerL1.SINGLETON);
   }
 
@@ -78,13 +77,12 @@ public ClientMessageTransport(TCConnectionManager clientConnectionEstablisher,
   public ClientMessageTransport(TCConnectionManager connectionManager,
                                 TransportHandshakeErrorHandler handshakeErrorHandler,
                                 TransportHandshakeMessageFactory messageFactory,
-                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory, int callbackPort, int timeout,
+                                WireProtocolAdaptorFactory wireProtocolAdaptorFactory, int timeout,
                                 ReconnectionRejectedHandler reconnectionRejectedHandler) {
 
     super(MessageTransportState.STATE_START, handshakeErrorHandler, messageFactory, LoggerFactory.getLogger(ClientMessageTransport.class));
     this.wireProtocolAdaptorFactory = wireProtocolAdaptorFactory;
     this.connectionManager = connectionManager;
-    this.callbackPort = callbackPort;
     this.timeout = timeout;
     this.reconnectionRejectedHandler = reconnectionRejectedHandler;
   }
@@ -309,7 +307,6 @@ private void handleSynAck(WireProtocolMessage message) {
       }
       getConnection().setTransportEstablished();
       setSynAckResult(synAck);
-      setRemoteCallbackPort(synAck.getCallbackPort());
     }
   }
 
@@ -400,7 +397,7 @@ private Future<SynAckMessage> sendSyn() throws TransportHandshakeException {
       // get the stack layer list and pass it in
       short stackLayerFlags = getCommunicationStackFlags(this);
       TransportHandshakeMessage syn = this.messageFactory.createSyn(getConnectionID(), getConnection(),
-                                                                    stackLayerFlags, this.callbackPort);
+                                                                    stackLayerFlags);
       // send syn message
       try {
         this.sendToConnection(syn);

common/src/main/java/com/tc/net/protocol/transport/ConnectionHealthCheckerContextImpl.java

@@ -67,7 +67,6 @@ class ConnectionHealthCheckerContextImpl implements ConnectionHealthCheckerConte
 
   // Context info
   private final HealthCheckerConfig              config;
-  private final int                              callbackPort;
   private final String                           remoteNodeDesc;
 
   // Socket Connect probes
@@ -96,15 +95,8 @@ public ConnectionHealthCheckerContextImpl(MessageTransportBase mtb, HealthChecke
     this.remoteNodeDesc = mtb.getRemoteAddress().toString();
     logger.info("Health monitoring agent started for " + remoteNodeDesc);
     currentState = INIT;
-    callbackPort = transport.getRemoteCallbackPort();
     configFactor = 1;
-    // only verify the callback port if the callback is different 
-    // from the connection port
-    if (callbackPort != transport.getRemoteAddress().getPort()) {
-      initCallbackPortVerification();
-    } else {
-      changeState(START);
-    }
+    changeState(START);
   }
 
   @Override
@@ -113,29 +105,6 @@ public synchronized void close() {
     sockectConnect.stop();
   }
 
-  // RMP-343
-  private void initCallbackPortVerification() {
-    if (config.isSocketConnectOnPingFail()) {
-      SocketConnectStartStatus status = initSocketConnectProbe();
-      if (status == SocketConnectStartStatus.FAILED) {
-        // 1. callback port handshaked and not reachable -- upgrade the config factor
-        callbackPortVerificationFailed();
-      } else if (status == SocketConnectStartStatus.NOT_STARTED) {
-        // 2. callback port not handshaked -- just log it, no config upgrade, move to next state
-        changeState(START);
-      } else if (status == SocketConnectStartStatus.STARTED) {
-        // async socket connect to callback port has started. HC state remains at INIT. state change happens on
-        // connection events
-      } else {
-        throw new AssertionError("initCallbackPortVerification: Unexpected SocketConnectStart Status");
-      }
-    } else {
-      logger
-          .info("HealthCheck SocketConnect disabled for " + remoteNodeDesc + ". HealthCheckCallbackPort not verified");
-      changeState(START);
-    }
-  }
-
   /* all callers of this method are already synchronized */
   private void changeState(State newState) {
     if (logger.isDebugEnabled() && currentState != newState) {
@@ -188,15 +157,8 @@ protected TCConnection getNewConnection(TCConnectionManager connManager) throws
   protected HealthCheckerSocketConnect getHealthCheckerSocketConnector(TCConnection connection,
                                                                        MessageTransportBase transportBase,
                                                                        Logger loger, HealthCheckerConfig cnfg) {
-    if (TransportHandshakeMessage.NO_CALLBACK_PORT == callbackPort) {
-      logger.info("No HealthCheckCallbackPort handshaked for node " + remoteNodeDesc);
-      return new NullHealthCheckerSocketConnectImpl();
-    }
-
-    // TODO: do we need to exchange the address as well ??? (since it might be different than the remote IP on this
-    // conn)
-    InetSocketAddress sa = new InetSocketAddress(transportBase.getRemoteAddress().getAddress(), callbackPort);
-    return new HealthCheckerSocketConnectImpl(sa, connection, remoteNodeDesc + "(callbackport:" + callbackPort + ")",
+    InetSocketAddress sa = transportBase.getRemoteAddress();
+    return new HealthCheckerSocketConnectImpl(sa, connection, remoteNodeDesc,
                                               loger, cnfg.getSocketConnectTimeout());
   }
 
@@ -310,17 +272,14 @@ public synchronized boolean probeIfAlive() {
   }
 
   private void callbackPortVerificationFailed() {
-    transport.setRemoteCallbackPort(TransportHandshakeMessage.NO_CALLBACK_PORT);
     updateConfigFactor(CONFIG_UPGRADE_FACTOR);
     changeState(START);
-    logger.debug("HealthCheckCallbackPort verification FAILED for " + remoteNodeDesc + "(callbackport: " + callbackPort
-                + ")");
+    logger.debug("HealthCheckCallbackPort verification FAILED for " + remoteNodeDesc);
   }
 
   private void callbackPortVerificationSuccess() {
     changeState(START);
-    logger.debug("HealthCheckCallbackPort verification PASSED for " + remoteNodeDesc + "(callbackport: " + callbackPort
-                + ")");
+    logger.debug("HealthCheckCallbackPort verification PASSED for " + remoteNodeDesc);
   }
 
   @Override

common/src/main/java/com/tc/net/protocol/transport/DisabledHealthCheckerConfigImpl.java

@@ -65,11 +65,6 @@ public int getSocketConnectTimeout() {
     throw new AssertionError("Disabled HealthChecker");
   }
 
-  @Override
-  public boolean isCallbackPortListenerNeeded() {
-    return false;
-  }
-
   @Override
   public boolean isCheckTimeEnabled() {
     throw new AssertionError("Disabled HealthChecker");

common/src/main/java/com/tc/net/protocol/transport/HealthCheckerConfig.java

@@ -52,10 +52,6 @@ public interface HealthCheckerConfig {
 
   int getSocketConnectTimeout();
 
-  /**
-   * RMP-343: L2 SocketConnect L1
-   */
-  boolean isCallbackPortListenerNeeded();
   /**
    * Checking time difference between hosts enabled/disabled.
    */

common/src/main/java/com/tc/net/protocol/transport/HealthCheckerConfigClientImpl.java

@@ -31,13 +31,8 @@ public HealthCheckerConfigClientImpl(String name, String bindPort) {
     super(name);
   }
 
-  public HealthCheckerConfigClientImpl(long idle, long interval, int probes, String name, boolean extraCheck,
+  public HealthCheckerConfigClientImpl(long idle, long interval, int probes, String name,
                                        int socketConnectMaxCount, int socketConnectTimeout) {
-    super(idle, interval, probes, name, extraCheck, socketConnectMaxCount, socketConnectTimeout);
-  }
-
-  @Override
-  public boolean isCallbackPortListenerNeeded() {
-    return true;
+    super(idle, interval, probes, name, false, socketConnectMaxCount, socketConnectTimeout);
   }
 }

common/src/main/java/com/tc/net/protocol/transport/HealthCheckerConfigImpl.java

@@ -139,11 +139,6 @@ public int getSocketConnectTimeout() {
     return this.socketConnectTimeout;
   }
 
-  @Override
-  public boolean isCallbackPortListenerNeeded() {
-    return false;
-  }
-
   @Override
   public boolean isCheckTimeEnabled() {
     return this.checkTimeEnabled;

common/src/main/java/com/tc/net/protocol/transport/MessageTransport.java

@@ -50,9 +50,5 @@ public interface MessageTransport extends NetworkLayer, PrettyPrintable {
 
   public String getCommunicationStackNames(NetworkLayer parentLayer);
 
-  public void setRemoteCallbackPort(int callbackPort);
-
-  public int getRemoteCallbackPort();
-
   public void initConnectionID(ConnectionID cid);
 }

common/src/main/java/com/tc/net/protocol/transport/MessageTransportBase.java

@@ -55,7 +55,6 @@ abstract class MessageTransportBase extends AbstractMessageTransport implements
 
   private final AtomicReference<TCConnectionEvent> connectionCloseEvent   = new AtomicReference<>();
   private volatile ConnectionHealthCheckerContext  healthCheckerContext   = new ConnectionHealthCheckerContextDummyImpl();
-  private int                                      remoteCallbackPort     = TransportHandshakeMessage.NO_CALLBACK_PORT;
 
   protected MessageTransportBase(MessageTransportState initialState,
                                  TransportHandshakeErrorHandler handshakeErrorHandler,
@@ -386,16 +385,6 @@ public String getStackLayerName() {
     return NAME_TRANSPORT_LAYER;
   }
 
-  @Override
-  public synchronized int getRemoteCallbackPort() {
-    return this.remoteCallbackPort;
-  }
-
-  @Override
-  public synchronized void setRemoteCallbackPort(int remoteCallbackPort) {
-    this.remoteCallbackPort = remoteCallbackPort;
-  }
-
   @Override
   public synchronized final void initConnectionID(ConnectionID cid) {
     connectionId = cid;