Reject any reconnect attempts outside the reconnect window. Unless something very strange happens to the underlying network comms, this should not happen. If it does, any reconnect handshake will be rejected when outside the reconnect window. The server has no idea how to handle reconnects extra processing outside the reconnect window.

Author: Myron Scott

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

@@ -54,7 +54,7 @@
 public class ServerStackProvider implements NetworkStackProvider, MessageTransportListener, ProtocolAdaptorFactory {
   private static final Logger logger = LoggerFactory.getLogger(ServerStackProvider.class);
 
-  private final Map<ClientID, ServerNetworkStackHarness> harnesses          = new ConcurrentHashMap<ClientID, ServerNetworkStackHarness>();
+  private final Map<ClientID, ServerNetworkStackHarness> harnesses          = new ConcurrentHashMap<>();
   private final NetworkStackHarnessFactory       harnessFactory;
   private final ServerMessageChannelFactory      channelFactory;
   private final TransportHandshakeMessageFactory handshakeMessageFactory;
@@ -63,7 +63,7 @@ public class ServerStackProvider implements NetworkStackProvider, MessageTranspo
   private final WireProtocolAdaptorFactory       wireProtocolAdaptorFactory;
   private final WireProtocolMessageSink          wireProtoMsgsink;
   private final MessageTransportFactory          messageTransportFactory;
-  private final List<MessageTransportListener>   transportListeners = new ArrayList<MessageTransportListener>();
+  private final List<MessageTransportListener>   transportListeners = new ArrayList<>();
   private final ReentrantLock                    licenseLock;
   private final RedirectAddressProvider                 activeProvider;
   private final Predicate<MessageTransport>                validateTransport;
@@ -213,7 +213,7 @@ public void notifyTransportClosed(MessageTransport transport) {
       if (!connectionId.isJvmIDNull()) {
         boolean removed = this.connectionPolicy.clientDisconnected(connectionId);
         if (removed) {
-          logger.warn("connectionid not removed be transport disconnect");
+          logger.warn("connectionid not removed by transport disconnect");
         }
       }
     }

common/src/test/java/com/tc/net/protocol/transport/ServerMessageTransportTest.java

@@ -21,6 +21,7 @@
 import com.tc.net.core.TCConnection;
 import com.tc.net.core.event.TCConnectionEvent;
 import com.tc.net.core.event.TCConnectionEventListener;
+import com.tc.net.protocol.IllegalReconnectException;
 import com.tc.util.Assert;
 import java.util.ArrayList;
 import java.util.List;
@@ -124,4 +125,47 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
     verify(checker, times(2)).notifyTransportDisconnected(eq(transport), eq(false));
     verify(checker, times(1)).notifyTransportDisconnected(eq(transport), eq(true));
   }
+  
+  @Test
+  public void testDoubleAttachFails() throws Exception {
+    ConnectionID id = new ConnectionID("JVM", 1);
+    TCConnection connection = mock(TCConnection.class);
+    
+    MessageTransportListener checker = mock(MessageTransportListener.class);
+    
+    final List<TCConnectionEventListener> listeners = new ArrayList<TCConnectionEventListener>();
+    doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) throws Throwable {
+        listeners.add((TCConnectionEventListener)invocation.getArguments()[0]);
+        return null;
+      }
+    }).when(connection).addListener(ArgumentMatchers.any(TCConnectionEventListener.class));
+    when(connection.isConnected()).thenReturn(Boolean.TRUE);
+    TransportHandshakeErrorHandler errHdr = mock(TransportHandshakeErrorHandler.class);
+    TransportHandshakeMessageFactory factory = mock(TransportHandshakeMessageFactory.class);
+    ServerMessageTransport transport = new ServerMessageTransport(errHdr, factory);
+    transport.initConnectionID(id);
+    transport.addTransportListener(checker);
+
+    Assert.assertFalse(transport.status.isConnected());
+    
+    transport.attachNewConnection(connection);
+    Assert.assertTrue(transport.status.isConnected());
+        
+    try {
+      transport.attachNewConnection(connection);
+      Assert.fail();
+    } catch (IllegalReconnectException illegal) {
+      //  expected
+    }
+    transport.close();
+    Assert.assertFalse(transport.status.isConnected());
+    try {
+      transport.attachNewConnection(connection);
+      Assert.fail();
+    } catch (IllegalReconnectException illegal) {
+      //  expected
+    }
+  }
 }

galvan-support/src/test/java/org/terracotta/functional/ReconnectRejectIT.java

@@ -0,0 +1,74 @@
+/*
+ *
+ *  The contents of this file are subject to the Terracotta Public License Version
+ *  2.0 (the "License"); You may not use this file except in compliance with the
+ *  License. You may obtain a copy of the License at
+ *
+ *  http://terracotta.org/legal/terracotta-public-license.
+ *
+ *  Software distributed under the License is distributed on an "AS IS" basis,
+ *  WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for
+ *  the specific language governing rights and limitations under the License.
+ *
+ *  The Covered Software is Terracotta Core.
+ *
+ *  The Initial Developer of the Covered Software is
+ *  Terracotta, Inc., a Software AG company
+ *
+ */
+package org.terracotta.functional;
+
+
+import java.io.StringReader;
+import java.net.InetSocketAddress;
+import java.util.Properties;
+import junit.framework.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.terracotta.connection.Connection;
+import org.terracotta.connection.Diagnostics;
+import org.terracotta.connection.DiagnosticsFactory;
+import org.terracotta.connection.entity.EntityRef;
+import org.terracotta.entity.map.ConcurrentClusteredMap;
+import org.terracotta.entity.map.MapConfig;
+import org.terracotta.exception.ConnectionClosedException;
+import org.terracotta.testing.rules.BasicExternalClusterBuilder;
+import org.terracotta.testing.rules.Cluster;
+
+/**
+ *
+ */
+public class ReconnectRejectIT {
+
+  @Rule
+  public final Cluster CLUSTER = BasicExternalClusterBuilder.newCluster(1).withClientReconnectWindowTime(30)
+      .build();
+
+  @Test
+  public void testClusterHostPorts() throws Exception {
+    Connection connection = CLUSTER.newConnection();
+    String hp = CLUSTER.getClusterHostPorts()[0];
+    String[] shp = hp.split(":");
+    int port = Integer.parseInt(shp[1]);
+    String id = null;
+    try (Diagnostics d = DiagnosticsFactory.connect(InetSocketAddress.createUnresolved(shp[0], port), new Properties())) {
+      Properties clients = new Properties();
+      clients.load(new StringReader(d.invoke("Server", "getConnectedClients")));
+      id = clients.getProperty("clients.0.id");
+    }
+    try (Diagnostics d = DiagnosticsFactory.connect(InetSocketAddress.createUnresolved(shp[0], port), new Properties())) {
+      System.out.println(d.invokeWithArg("Server", "disconnectClient", id));
+    }
+    try {
+      EntityRef<ConcurrentClusteredMap, MapConfig, Void>  ref = connection.getEntityRef(ConcurrentClusteredMap.class, 1L, "ROOT");
+      ref.create(new MapConfig(1, "ROOT"));
+      ConcurrentClusteredMap map = ref.fetchEntity(null);
+      Assert.fail();
+    } catch (ConnectionClosedException e) {
+      //expected
+      // There should be TRANSPORT_RECONNECTION_REJECTED_EVENT in the client logs
+      e.printStackTrace();
+    }
+    
+  }
+}

management/src/main/java/com/tc/management/beans/TCServerInfoMBean.java

@@ -102,6 +102,8 @@ public interface TCServerInfoMBean extends TerracottaMBean, RuntimeStatisticCons
 
   void setPipelineMonitoring(boolean monitor);
 
+  boolean disconnectClient(String id);
+  
   String getClusterState(boolean shortForm);
 
   String getConnectedClients() throws IOException;

tc-client/src/main/java/com/tc/object/DistributedObjectClient.java

@@ -240,7 +240,7 @@ private synchronized ClientMessageChannel internalStart(int socketTimeout) {
 
     final ProductInfo pInfo = ProductInfo.getInstance(getClass().getClassLoader());
 
-    ClientHandshakeMessageFactory chmf = (u, n, c, r)->{
+    ClientHandshakeMessageFactory chmf = (u, n, c, r, reconnect)->{
       ClientMessageChannel cmc = getClientMessageChannel();
       if (cmc != null) {
         final ClientHandshakeMessage rv = (ClientHandshakeMessage)cmc.createMessage(TCMessageType.CLIENT_HANDSHAKE_MESSAGE);
@@ -249,6 +249,7 @@ private synchronized ClientMessageChannel internalStart(int socketTimeout) {
         rv.setClientPID(getPID());
         rv.setUUID(u);
         rv.setName(n);
+        rv.setReconnect(reconnect);
         return rv;
       } else {
         return null;

tc-client/src/main/java/com/tc/object/handshakemanager/ClientHandshakeManagerImpl.java

@@ -59,6 +59,7 @@ private enum State {
 
   private State state;
   private volatile boolean disconnected;
+  private volatile boolean wasConnected;
   private volatile boolean isShutdown = false;
 
   public ClientHandshakeManagerImpl(Logger logger, ClientHandshakeMessageFactory chmf,
@@ -73,6 +74,7 @@ public ClientHandshakeManagerImpl(Logger logger, ClientHandshakeMessageFactory c
     this.callBacks = entities;
     this.state = State.PAUSED;
     this.disconnected = true;
+    this.wasConnected = false;
     pauseCallbacks();
   }
 
@@ -106,7 +108,7 @@ private void initiateHandshake() {
     ClientHandshakeMessage handshakeMessage;
 
     changeToStarting();
-    handshakeMessage = this.chmf.newClientHandshakeMessage(this.uuid, this.name, this.clientVersion, this.clientRevision);
+    handshakeMessage = this.chmf.newClientHandshakeMessage(this.uuid, this.name, this.clientVersion, this.clientRevision, this.wasConnected);
     if (handshakeMessage != null) {
       notifyCallbackOnHandshake(handshakeMessage);
 
@@ -242,5 +244,6 @@ private void changeToRunning() {
     state = State.RUNNING;
 
     this.disconnected = false;
+    this.wasConnected = true;
   }
 }

tc-client/src/test/java/com/tc/object/ClientEntityManagerTest.java

@@ -579,6 +579,7 @@ public void result(byte[] response) {
       result.get(1, TimeUnit.SECONDS);
       Assert.fail();
     } catch (TimeoutException to) {
+      assertTrue(!result.isDone());
       assertThat(System.currentTimeMillis() - start, Matchers.greaterThanOrEqualTo(1000L));
       //  expected
     }
@@ -587,6 +588,7 @@ public void result(byte[] response) {
       result.get(2, TimeUnit.SECONDS);
       Assert.fail();
     } catch (TimeoutException to) {
+      assertTrue(!result.isDone());
       assertThat(System.currentTimeMillis() - start, Matchers.greaterThanOrEqualTo(2000L));
       //  expected
     }

tc-messaging/src/main/java/com/tc/object/msg/ClientHandshakeMessage.java

@@ -26,6 +26,10 @@
 
 public interface ClientHandshakeMessage extends TCAction {
 
+  void setReconnect(boolean isReconnect);
+  
+  boolean isReconnect();
+  
   void setUUID(String uuid);
 
   String getUUID();

tc-messaging/src/main/java/com/tc/object/msg/ClientHandshakeMessageFactory.java

@@ -20,6 +20,6 @@
 
 public interface ClientHandshakeMessageFactory {
 
-  public ClientHandshakeMessage newClientHandshakeMessage(String uuid, String name, String clientVersion, String clientRevision);
+  public ClientHandshakeMessage newClientHandshakeMessage(String uuid, String name, String clientVersion, String clientRevision, boolean reconnect);
 
 }

tc-messaging/src/main/java/com/tc/object/msg/ClientHandshakeMessageImpl.java

@@ -39,7 +39,7 @@
 
 
 public class ClientHandshakeMessageImpl extends DSOMessageBase implements ClientHandshakeMessage {
-  private static final byte   UNUSED_1        = 1;
+  private static final byte   RECONNECT        = 1;
   private static final byte   CLIENT_VERSION           = 2;
   private static final byte   UNUSED_2        = 3;
   private static final byte   LOCAL_TIME_MILLS         = 4;
@@ -58,6 +58,7 @@ public class ClientHandshakeMessageImpl extends DSOMessageBase implements Client
   private String              clientRevision           = "";
   private String              clientAddress            = ""; 
   private int                 pid                      = -1;
+  private boolean             reconnect                = false;
   private final Set<ClientEntityReferenceContext> reconnectReferences = new HashSet<ClientEntityReferenceContext>();
   private final Set<ResendVoltronEntityMessage> resendMessages = new TreeSet<ResendVoltronEntityMessage>(new Comparator<ResendVoltronEntityMessage>() {
     @Override
@@ -80,6 +81,16 @@ public ClientHandshakeMessageImpl(SessionID sessionID, MessageMonitor monitor, M
     clientAddress = TCSocketAddress.getStringForm(channel.getLocalAddress());
   }
 
+  @Override
+  public void setReconnect(boolean isReconnect) {
+    this.reconnect = isReconnect;
+  }
+
+  @Override
+  public boolean isReconnect() {
+    return this.reconnect;
+  }
+
   @Override
   public String getClientVersion() {
     return this.clientVersion;
@@ -142,7 +153,7 @@ public String getClientAddress() {
 
   @Override
   protected void dehydrateValues() {
-    putNVPair(UNUSED_1, false);  // unused but keep for compatibility
+    putNVPair(RECONNECT, reconnect);  // unused but keep for compatibility
     putNVPair(UNUSED_2, false);  // unused but keep for compatibility
     putNVPair(CLIENT_UUID, this.uuid);
     putNVPair(CLIENT_NAME, this.name);
@@ -162,8 +173,8 @@ protected void dehydrateValues() {
   @Override
   protected boolean hydrateValue(byte name) throws IOException {
     switch (name) {
-      case UNUSED_1:
-        getBooleanValue();  // unused but keep for compatibility
+      case RECONNECT:
+        this.reconnect = getBooleanValue();
         return true;
       case UNUSED_2:
         getBooleanValue();  // unused but keep for compatibility