You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ud...@apache.org on 2017/09/14 23:09:22 UTC
[geode] branch develop updated: GEODE-3082 Integrate
GenericProtocolServerConnection with ClientHealthMonitor.
This is an automated email from the ASF dual-hosted git repository.
udo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new b31de81 GEODE-3082 Integrate GenericProtocolServerConnection with ClientHealthMonitor.
b31de81 is described below
commit b31de81a8c19c61ad27a2ab803073214050db154
Author: Alexander Murmann <am...@pivotal.io>
AuthorDate: Wed Sep 6 17:03:40 2017 -0700
GEODE-3082 Integrate GenericProtocolServerConnection with ClientHealthMonitor.
1. Now GenericProtocolServerConnection creates ClientProxyMembershipId.
2. ClientHealthMonitor gets notified about pings
3. added test where CHM closes the connection.
4. added test where CHM doesn't close the connection
This currently results in every connetion being tracked as a separate
client.
---
.../membership/InternalDistributedMember.java | 2 +-
.../cache/tier/sockets/ClientHealthMonitor.java | 28 +-
.../sockets/GenericProtocolServerConnection.java | 55 +++-
.../cache/tier/sockets/ServerConnection.java | 2 +-
.../tier/sockets/ClientHealthMonitorJUnitTest.java | 197 +++++++------
.../GenericProtocolServerConnectionTest.java | 110 ++++++--
.../acceptance/CacheConnectionJUnitTest.java | 305 +++++++++++++++++++++
.../CacheConnectionTimeoutJUnitTest.java | 162 +++++++++++
.../CacheOperationsJUnitTest.java} | 191 +------------
.../LocatorConnectionDUnitTest.java} | 24 +-
.../protocol/{ => acceptance}/default.keystore | Bin
11 files changed, 744 insertions(+), 332 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/InternalDistributedMember.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/InternalDistributedMember.java
index e152756..18f4c33 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/InternalDistributedMember.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/InternalDistributedMember.java
@@ -117,7 +117,7 @@ public class InternalDistributedMember implements DistributedMember, Externaliza
* member for use in the P2P cache. Use of other constructors can break
* network-partition-detection.
*
- * @param i
+ * @param i the inet address
* @param p the membership port
* @param splitBrainEnabled whether this feature is enabled for the member
* @param canBeCoordinator whether the member is eligible to be the membership coordinator
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitor.java
index 226da8a..97abbd8 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitor.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicIntegerArray;
+import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
@@ -38,8 +39,6 @@ import org.apache.geode.internal.cache.IncomingGatewayStatus;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.TXId;
import org.apache.geode.internal.cache.TXManagerImpl;
-import org.apache.geode.internal.cache.tier.Acceptor;
-import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.concurrent.ConcurrentHashSet;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
@@ -55,6 +54,8 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
*/
public class ClientHealthMonitor {
private static final Logger logger = LogService.getLogger();
+ public static final String CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY =
+ "geode.client-health-monitor-interval";
/**
* The map of known clients
@@ -81,6 +82,12 @@ public class ClientHealthMonitor {
*/
private final InternalCache _cache;
+ public int getMaximumTimeBetweenPings() {
+ return maximumTimeBetweenPings;
+ }
+
+ private final int maximumTimeBetweenPings;
+
/**
* A thread that validates client connections
*/
@@ -100,9 +107,9 @@ public class ClientHealthMonitor {
/**
* The interval between client monitor iterations
*/
- final protected static long CLIENT_MONITOR_INTERVAL = 1000;
+ private final static long DEFAULT_CLIENT_MONITOR_INTERVAL_IN_MILLIS = 1000;
- final private CacheClientNotifierStats stats;
+ private final CacheClientNotifierStats stats;
/**
* Used to track the number of handshakes in a VM primary use, license enforcement.
@@ -125,6 +132,12 @@ public class ClientHealthMonitor {
*/
AtomicIntegerArray numOfClientsPerVersion = new AtomicIntegerArray(Version.HIGHEST_VERSION + 1);
+ public long getMonitorInterval() {
+ return monitorInterval;
+ }
+
+ private long monitorInterval;
+
/**
* Factory method to construct or return the singleton <code>ClientHealthMonitor</code> instance.
*
@@ -667,10 +680,15 @@ public class ClientHealthMonitor {
CacheClientNotifierStats stats) {
// Set the Cache
this._cache = cache;
+ this.maximumTimeBetweenPings = maximumTimeBetweenPings;
// Initialize the client threads map
this._clientThreads = new HashMap();
+ this.monitorInterval = Long.getLong(CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY,
+ DEFAULT_CLIENT_MONITOR_INTERVAL_IN_MILLIS);
+ logger.debug("Setting monitorInterval to {}", this.monitorInterval);
+
if (maximumTimeBetweenPings > 0) {
if (logger.isDebugEnabled()) {
logger.debug("{}: Initializing client health monitor thread", this);
@@ -800,7 +818,7 @@ public class ClientHealthMonitor {
while (!this._isStopped) {
SystemFailure.checkFailure();
try {
- Thread.sleep(CLIENT_MONITOR_INTERVAL);
+ Thread.sleep(monitorInterval);
if (logger.isTraceEnabled()) {
logger.trace("Monitoring {} client(s)", getClientHeartbeats().size());
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
index 767b6c5..136a659 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnection.java
@@ -15,19 +15,25 @@
package org.apache.geode.internal.cache.tier.sockets;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+import org.apache.geode.cache.client.PoolFactory;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.ServerLocation;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.security.SecurityManager;
import org.apache.geode.security.server.Authenticator;
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.Socket;
-
/**
* Holds the socket and protocol handler for the new client protocol.
*/
@@ -37,22 +43,26 @@ public class GenericProtocolServerConnection extends ServerConnection {
private final SecurityManager securityManager;
private final Authenticator authenticator;
private boolean cleanedUp;
+ private ClientProxyMembershipID clientProxyMembershipID;
/**
* Creates a new <code>GenericProtocolServerConnection</code> that processes messages received
* from an edge client over a given <code>Socket</code>.
*/
- public GenericProtocolServerConnection(Socket socket, InternalCache cache,
- CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
- String communicationModeStr, byte communicationMode, Acceptor acceptor,
- ClientProtocolMessageHandler newClientProtocol, SecurityService securityService,
- Authenticator authenticator) {
- super(socket, cache, helper, stats, hsTimeout, socketBufferSize, communicationModeStr,
+ public GenericProtocolServerConnection(Socket socket, InternalCache c, CachedRegionHelper helper,
+ CacheServerStats stats, int hsTimeout, int socketBufferSize, String communicationModeStr,
+ byte communicationMode, Acceptor acceptor, ClientProtocolMessageHandler newClientProtocol,
+ SecurityService securityService, Authenticator authenticator) {
+ super(socket, c, helper, stats, hsTimeout, socketBufferSize, communicationModeStr,
communicationMode, acceptor, securityService);
securityManager = securityService.getSecurityManager();
this.messageHandler = newClientProtocol;
this.authenticator = authenticator;
this.messageHandler.getStatistics().clientConnected();
+
+ setClientProxyMembershipId();
+
+ doHandShake(CommunicationMode.ProtobufClientServerProtocol.getModeNumber(), 0);
}
@Override
@@ -75,9 +85,20 @@ public class GenericProtocolServerConnection extends ServerConnection {
logger.warn(e);
this.setFlagProcessMessagesAsFalse();
setClientDisconnectedException(e);
+ } finally {
+ acceptor.getClientHealthMonitor().receivedPing(this.clientProxyMembershipID);
}
}
+ private void setClientProxyMembershipId() {
+ ServerLocation serverLocation = new ServerLocation(
+ ((InetSocketAddress) this.getSocket().getRemoteSocketAddress()).getHostName(),
+ this.getSocketPort());
+ DistributedMember distributedMember = new InternalDistributedMember(serverLocation);
+ // no handshake for new client protocol.
+ clientProxyMembershipID = new ClientProxyMembershipID(distributedMember);
+ }
+
@Override
public boolean cleanup() {
synchronized (this) {
@@ -91,11 +112,19 @@ public class GenericProtocolServerConnection extends ServerConnection {
@Override
protected boolean doHandShake(byte epType, int qSize) {
- // no handshake for new client protocol.
+ ClientHealthMonitor clientHealthMonitor = getAcceptor().getClientHealthMonitor();
+ clientHealthMonitor.registerClient(clientProxyMembershipID);
+ clientHealthMonitor.addConnection(clientProxyMembershipID, this);
+
return true;
}
@Override
+ protected int getClientReadTimeout() {
+ return PoolFactory.DEFAULT_READ_TIMEOUT;
+ }
+
+ @Override
public boolean isClientServerConnection() {
return true;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
index 160e05b..7fc688c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
@@ -1264,7 +1264,7 @@ public abstract class ServerConnection implements Runnable {
this.requestSpecificTimeout = -1;
}
- int getClientReadTimeout() {
+ protected int getClientReadTimeout() {
if (this.requestSpecificTimeout == -1)
return this.handshake.getClientReadTimeout();
else
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitorJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitorJUnitTest.java
index 93b62d9..2937233 100755
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitorJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientHealthMonitorJUnitTest.java
@@ -14,6 +14,22 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
import org.apache.geode.Statistics;
import org.apache.geode.StatisticsType;
import org.apache.geode.cache.AttributesFactory;
@@ -30,25 +46,12 @@ import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.AvailablePort;
import org.apache.geode.internal.cache.EventID;
-import org.apache.geode.test.dunit.Wait;
-import org.apache.geode.test.dunit.WaitCriterion;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.junit.categories.ClientServerTest;
import org.apache.geode.test.junit.categories.IntegrationTest;
-import org.junit.After;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-import java.util.Properties;
-
-import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
-import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
/**
* This is a functional-test for <code>ClientHealthMonitor</code>.
- *
- *
*/
@Category({IntegrationTest.class, ClientServerTest.class})
public class ClientHealthMonitorJUnitTest {
@@ -59,16 +62,24 @@ public class ClientHealthMonitorJUnitTest {
return 0;
}
- /** connection proxy object for the client */
+ /**
+ * connection proxy object for the client
+ */
PoolImpl proxy = null;
- /** the distributed system instance for the test */
+ /**
+ * the distributed system instance for the test
+ */
DistributedSystem system;
- /** the cache instance for the test */
+ /**
+ * the cache instance for the test
+ */
Cache cache;
- /** name of the region created */
+ /**
+ * name of the region created
+ */
final String regionName = "region1";
private static int PORT;
@@ -81,13 +92,17 @@ public class ClientHealthMonitorJUnitTest {
{
removeExceptions();
- this.cache.close();
- this.system.disconnect();
+ if (this.cache != null) {
+ this.cache.close();
+ }
+ if (this.system != null) {
+ this.system.disconnect();
+ }
+ ClientHealthMonitor.shutdownInstance();
}
/**
* Initializes proxy object and creates region for client
- *
*/
private void createProxyAndRegionForClient() {
try {
@@ -104,11 +119,10 @@ public class ClientHealthMonitorJUnitTest {
}
}
- private final static int TIME_BETWEEN_PINGS = 2500;
+ private final static int TIME_BETWEEN_PINGS = 50;
/**
* Creates and starts the server instance
- *
*/
private int createServer() {
CacheServer server = null;
@@ -133,6 +147,36 @@ public class ClientHealthMonitorJUnitTest {
return server.getPort();
}
+ @Test
+ public void settingMonitorIntervalViaProperty() {
+ int monitorInterval = 10;
+ System.setProperty(ClientHealthMonitor.CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY,
+ String.valueOf(monitorInterval));
+
+ assertEquals(monitorInterval,
+ ClientHealthMonitor
+ .getInstance(mock(InternalCache.class), 0, mock(CacheClientNotifierStats.class))
+ .getMonitorInterval());
+ }
+
+ @Test
+ public void monitorIntervalDefaultsWhenNotSet() {
+ assertNotNull(ClientHealthMonitor
+ .getInstance(mock(InternalCache.class), 0, mock(CacheClientNotifierStats.class))
+ .getMonitorInterval());
+ }
+
+ @Test
+ public void monitorIntervalDefaultsWhenInvalidValue() {
+ String monitorInterval = "this isn't a number";
+ System.setProperty(ClientHealthMonitor.CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY,
+ monitorInterval);
+
+ assertNotNull(ClientHealthMonitor
+ .getInstance(mock(InternalCache.class), 0, mock(CacheClientNotifierStats.class))
+ .getMonitorInterval());
+ }
+
/**
* This test performs the following:<br>
* 1)create server<br>
@@ -145,53 +189,55 @@ public class ClientHealthMonitorJUnitTest {
*/
@Test
public void testDeadClientRemovalByServer() throws Exception {
+ System.setProperty(ClientHealthMonitor.CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY,
+ String.valueOf("100"));
PORT = createServer();
createProxyAndRegionForClient();
- // String connection2String = null;
- StatisticsType st = this.system.findType("CacheServerStats");
- final Statistics s = this.system.findStatisticsByType(st)[0];
- assertEquals(0, s.getInt("currentClients"));
- assertEquals(0, s.getInt("currentClientConnections"));
- this.system.getLogWriter().info("beforeAcquireConnection clients=" + s.getInt("currentClients")
- + " cnxs=" + s.getInt("currentClientConnections"));
+ StatisticsType statisticsType = this.system.findType("CacheServerStats");
+ final Statistics statistics = this.system.findStatisticsByType(statisticsType)[0];
+ assertEquals(0, statistics.getInt("currentClients"));
+ assertEquals(0, statistics.getInt("currentClientConnections"));
+ this.system.getLogWriter()
+ .info("beforeAcquireConnection clients=" + statistics.getInt("currentClients") + " cnxs="
+ + statistics.getInt("currentClientConnections"));
Connection connection1 = proxy.acquireConnection();
- this.system.getLogWriter().info("afterAcquireConnection clients=" + s.getInt("currentClients")
- + " cnxs=" + s.getInt("currentClientConnections"));
+ this.system.getLogWriter()
+ .info("afterAcquireConnection clients=" + statistics.getInt("currentClients") + " cnxs="
+ + statistics.getInt("currentClientConnections"));
this.system.getLogWriter().info("acquired connection " + connection1);
- WaitCriterion ev = new WaitCriterion() {
- public boolean done() {
- return s.getInt("currentClients") != 0;
- }
-
- public String description() {
- return null;
- }
+
+ int pollInterval = 20;
+ int maximumTimeBetweenPings = ClientHealthMonitor.getInstance().getMaximumTimeBetweenPings();
+
+ long monitorInterval = ClientHealthMonitor.getInstance().getMonitorInterval();
+ long connectTimeout = maximumTimeBetweenPings + monitorInterval + pollInterval;
+ Runnable verifyConnectedClients = () -> {
+ assertNotEquals(0, statistics.getInt("currentClients"));
};
- Wait.waitForCriterion(ev, 20 * 1000, 200, true);
- assertEquals(1, s.getInt("currentClients"));
- assertEquals(1, s.getInt("currentClientConnections"));
+ waitAndVerify(0, pollInterval, connectTimeout, verifyConnectedClients);
+
+ assertEquals(1, statistics.getInt("currentClients"));
+ assertEquals(1, statistics.getInt("currentClientConnections"));
// String connection1String = connection1.toString();
ServerRegionProxy srp = new ServerRegionProxy("region1", proxy);
srp.putOnForTestsOnly(connection1, "key-1", "value-1", new EventID(new byte[] {1}, 1, 1), null);
this.system.getLogWriter().info("did put 1");
// proxy.testfinalizeServerConnectionMonitor();
- ev = new WaitCriterion() {
- public boolean done() {
- return s.getInt("currentClients") == 0;
- }
-
- public String description() {
- return null;
- }
+
+ long disconnectTimeout = maximumTimeBetweenPings + monitorInterval + pollInterval;
+
+ Runnable verifyNoConnectedClients = () -> {
+ assertEquals(0, statistics.getInt("currentClients"));
};
- Wait.waitForCriterion(ev, TIME_BETWEEN_PINGS * 5, 200, true);
+ long disconnectDelay = monitorInterval + 1;
+ waitAndVerify(disconnectDelay, pollInterval, disconnectTimeout, verifyNoConnectedClients);
{
- this.system.getLogWriter().info("currentClients=" + s.getInt("currentClients")
- + " currentClientConnections=" + s.getInt("currentClientConnections"));
- assertEquals(0, s.getInt("currentClients"));
- assertEquals(0, s.getInt("currentClientConnections"));
+ this.system.getLogWriter().info("currentClients=" + statistics.getInt("currentClients")
+ + " currentClientConnections=" + statistics.getInt("currentClientConnections"));
+ assertEquals(0, statistics.getInt("currentClients"));
+ assertEquals(0, statistics.getInt("currentClientConnections"));
}
addExceptions();
// the connection should now fail since the server timed it out
@@ -200,38 +246,13 @@ public class ClientHealthMonitorJUnitTest {
fail("expected EOF");
} catch (ServerConnectivityException expected) {
}
- // The rest of this test no longer works.
- // connection1.finalizeConnection();
- // proxy.release();
-
- // connection1 = proxy.acquireConnection();
- // connection2String = connection1.toString();
- // this.system.getLogWriter().info("connection is now " + connection2String);
-
- // if (connection1String.equals(connection2String)) {
- // fail("New connection object was not obtained");
- // }
- // connection1.putObject("region1", "key-1", "value-2", new EventID(new byte[] {1},1,3), null);
- // this.system.getLogWriter().info("did put 2");
- // assertIndexDetailsEquals(1, s.getInt("currentClients"));
- // assertIndexDetailsEquals(1, s.getInt("currentClientConnections"));
-
- // // now lets see what happens when we close our connection
- // // note we use a nasty close which just closes the socket instead
- // // of sending a nice message to the server telling him we are going away
- // ((ConnectionImpl)connection1).finalizeConnection();
- // {
- // int retry = (TIME_BETWEEN_PINGS*5) / 100;
- // while (s.getInt("currentClients") > 0 && retry-- > 0) {
- // Thread.sleep(100);
- // }
- // this.system.getLogWriter().info("currentClients="
- // + s.getInt("currentClients")
- // + " currentClientConnections="
- // + s.getInt("currentClientConnections"));
- // assertIndexDetailsEquals(0, s.getInt("currentClients"));
- // assertIndexDetailsEquals(0, s.getInt("currentClientConnections"));
- // }
+ }
+
+ private void waitAndVerify(long pollDelay, int pollInterval, long timeout,
+ Runnable verifyNoConnectedClients) {
+ Awaitility.await().atMost(timeout, TimeUnit.MILLISECONDS)
+ .pollDelay(pollDelay, TimeUnit.MILLISECONDS)
+ .pollInterval(pollInterval, TimeUnit.MILLISECONDS).until(verifyNoConnectedClients);
}
public void addExceptions() throws Exception {
diff --git a/geode-protobuf/src/test/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnectionTest.java b/geode-protobuf/src/test/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnectionTest.java
index d52223c..4ec96e4 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnectionTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/internal/cache/tier/sockets/GenericProtocolServerConnectionTest.java
@@ -14,11 +14,25 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
@@ -28,28 +42,17 @@ import org.apache.geode.protocol.protobuf.statistics.NoOpProtobufStatistics;
import org.apache.geode.security.server.NoOpAuthenticator;
import org.apache.geode.test.junit.categories.UnitTest;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.mockito.Mockito;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.Socket;
-import java.net.UnknownHostException;
-
@Category(UnitTest.class)
public class GenericProtocolServerConnectionTest {
+
+ private ClientHealthMonitor clientHealthMonitorMock;
+
@Test
public void testProcessFlag() throws IOException {
- try {
- System.setProperty("geode.feature-protobuf-protocol", "true");
- ServerConnection serverConnection = IOExceptionThrowingServerConnection();
- Assert.assertTrue(serverConnection.processMessages);
- serverConnection.doOneMessage();
- Assert.assertTrue(!serverConnection.processMessages);
- } finally {
- System.clearProperty("geode.feature-protobuf-protocol");
- }
+ ServerConnection serverConnection = IOExceptionThrowingServerConnection();
+ Assert.assertTrue(serverConnection.processMessages);
+ serverConnection.doOneMessage();
+ Assert.assertTrue(!serverConnection.processMessages);
}
@Test
@@ -57,33 +60,86 @@ public class GenericProtocolServerConnectionTest {
Socket socketMock = mock(Socket.class);
when(socketMock.getInetAddress()).thenReturn(InetAddress.getByName("localhost"));
+ AcceptorImpl acceptorStub = mock(AcceptorImpl.class);
ClientProtocolMessageHandler mockHandler = mock(ClientProtocolMessageHandler.class);
- when(mockHandler.getStatistics()).thenReturn(new NoOpProtobufStatistics());
GenericProtocolServerConnection genericProtocolServerConnection =
- getGenericProtocolServerConnection(socketMock, mockHandler);
+ getServerConnection(socketMock, mockHandler, acceptorStub);
genericProtocolServerConnection.emergencyClose();
Mockito.verify(socketMock).close();
}
- private static ServerConnection IOExceptionThrowingServerConnection() throws IOException {
- Socket socketMock = mock(Socket.class);
- when(socketMock.getInetAddress()).thenReturn(InetAddress.getByName("localhost"));
+ @Test
+ public void testClientHealthMonitorRegistration() throws UnknownHostException {
+ AcceptorImpl acceptorStub = mock(AcceptorImpl.class);
ClientProtocolMessageHandler clientProtocolMock = mock(ClientProtocolMessageHandler.class);
+
+ ServerConnection serverConnection = getServerConnection(clientProtocolMock, acceptorStub);
+
+ ArgumentCaptor<ClientProxyMembershipID> registerCpmidArgumentCaptor =
+ ArgumentCaptor.forClass(ClientProxyMembershipID.class);
+
+ ArgumentCaptor<ClientProxyMembershipID> addConnectionCpmidArgumentCaptor =
+ ArgumentCaptor.forClass(ClientProxyMembershipID.class);
+
+ verify(clientHealthMonitorMock).addConnection(addConnectionCpmidArgumentCaptor.capture(),
+ eq(serverConnection));
+ verify(clientHealthMonitorMock).registerClient(registerCpmidArgumentCaptor.capture());
+ assertEquals("identity(localhost<ec>:0,connection=1",
+ registerCpmidArgumentCaptor.getValue().toString());
+ assertEquals("identity(localhost<ec>:0,connection=1",
+ addConnectionCpmidArgumentCaptor.getValue().toString());
+ }
+
+ @Test
+ public void testDoOneMessageNotifiesClientHealthMonitor() throws UnknownHostException {
+ AcceptorImpl acceptorStub = mock(AcceptorImpl.class);
+ ClientProtocolMessageHandler clientProtocolMock = mock(ClientProtocolMessageHandler.class);
+
+ ServerConnection serverConnection = getServerConnection(clientProtocolMock, acceptorStub);
+ serverConnection.doOneMessage();
+
+ ArgumentCaptor<ClientProxyMembershipID> clientProxyMembershipIDArgumentCaptor =
+ ArgumentCaptor.forClass(ClientProxyMembershipID.class);
+ verify(clientHealthMonitorMock).receivedPing(clientProxyMembershipIDArgumentCaptor.capture());
+ assertEquals("identity(localhost<ec>:0,connection=1",
+ clientProxyMembershipIDArgumentCaptor.getValue().toString());
+ }
+
+ private GenericProtocolServerConnection IOExceptionThrowingServerConnection() throws IOException {
+ ClientProtocolMessageHandler clientProtocolMock = mock(ClientProtocolMessageHandler.class);
ClientProtocolStatistics statisticsMock = mock(ClientProtocolStatistics.class);
when(clientProtocolMock.getStatistics()).thenReturn(statisticsMock);
doThrow(new IOException()).when(clientProtocolMock).receiveMessage(any(), any(), any());
- return getGenericProtocolServerConnection(socketMock, clientProtocolMock);
+ return getServerConnection(clientProtocolMock, mock(AcceptorImpl.class));
}
- private static GenericProtocolServerConnection getGenericProtocolServerConnection(
- Socket socketMock, ClientProtocolMessageHandler clientProtocolMock) {
+ private GenericProtocolServerConnection getServerConnection(Socket socketMock,
+ ClientProtocolMessageHandler clientProtocolMock, AcceptorImpl acceptorStub)
+ throws UnknownHostException {
+ clientHealthMonitorMock = mock(ClientHealthMonitor.class);
+ when(acceptorStub.getClientHealthMonitor()).thenReturn(clientHealthMonitorMock);
+ InetSocketAddress inetSocketAddressStub = InetSocketAddress.createUnresolved("localhost", 9071);
+ InetAddress inetAddressStub = mock(InetAddress.class);
+ when(socketMock.getInetAddress()).thenReturn(InetAddress.getByName("localhost"));
+ when(socketMock.getRemoteSocketAddress()).thenReturn(inetSocketAddressStub);
+ when(socketMock.getInetAddress()).thenReturn(inetAddressStub);
+
+ when(clientProtocolMock.getStatistics()).thenReturn(new NoOpProtobufStatistics());
+
return new GenericProtocolServerConnection(socketMock, mock(InternalCache.class),
mock(CachedRegionHelper.class), mock(CacheServerStats.class), 0, 0, "",
- CommunicationMode.ProtobufClientServerProtocol.getModeNumber(), mock(AcceptorImpl.class),
+ CommunicationMode.ProtobufClientServerProtocol.getModeNumber(), acceptorStub,
clientProtocolMock, mock(SecurityService.class), new NoOpAuthenticator());
}
+
+ private GenericProtocolServerConnection getServerConnection(
+ ClientProtocolMessageHandler clientProtocolMock, AcceptorImpl acceptorStub)
+ throws UnknownHostException {
+ Socket socketMock = mock(Socket.class);
+ return getServerConnection(socketMock, clientProtocolMock, acceptorStub);
+ }
}
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionJUnitTest.java
new file mode 100644
index 0000000..35bd26f
--- /dev/null
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionJUnitTest.java
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache 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://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.protocol.acceptance;
+
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENABLED_COMPONENTS;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_PASSWORD;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_TYPE;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_REQUIRE_AUTHENTICATION;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE;
+import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase.disconnectAllFromDS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.internal.AvailablePortHelper;
+import org.apache.geode.internal.admin.SSLConfig;
+import org.apache.geode.internal.cache.CacheServerImpl;
+import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
+import org.apache.geode.internal.net.SocketCreator;
+import org.apache.geode.internal.net.SocketCreatorFactory;
+import org.apache.geode.internal.protocol.protobuf.BasicTypes;
+import org.apache.geode.internal.protocol.protobuf.ClientProtocol;
+import org.apache.geode.internal.protocol.protobuf.RegionAPI;
+import org.apache.geode.internal.statistics.StatArchiveReader;
+import org.apache.geode.protocol.MessageUtil;
+import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
+import org.apache.geode.protocol.protobuf.ProtobufSerializationService;
+import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
+import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
+import org.apache.geode.serialization.SerializationService;
+import org.apache.geode.serialization.exception.UnsupportedEncodingTypeException;
+import org.apache.geode.serialization.registry.exception.CodecAlreadyRegisteredForTypeException;
+import org.apache.geode.serialization.registry.exception.CodecNotRegisteredForTypeException;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+import org.apache.geode.util.test.TestUtil;
+
+/**
+ * Test that using the magic byte to indicate intend ot use ProtoBuf messages works
+ */
+@Category(IntegrationTest.class)
+@RunWith(value = Parameterized.class)
+public class CacheConnectionJUnitTest {
+ private final String TEST_KEY = "testKey";
+ private final String TEST_VALUE = "testValue";
+ private final String TEST_REGION = "testRegion";
+ private final int TEST_PUT_CORRELATION_ID = 574;
+ private final int TEST_GET_CORRELATION_ID = 68451;
+
+ private final String DEFAULT_STORE = "default.keystore";
+ private final String SSL_PROTOCOLS = "any";
+ private final String SSL_CIPHERS = "any";
+
+
+ private Cache cache;
+ private int cacheServerPort;
+ private SerializationService serializationService;
+ private Socket socket;
+ private OutputStream outputStream;
+
+ @Parameterized.Parameter()
+ public boolean useSSL;
+
+ @Parameterized.Parameters(name = "use ssl {0}")
+ public static Collection<Boolean> data() {
+ return Arrays.asList(false, true);
+ }
+
+ @Rule
+ public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Rule
+ public TestName testName = new TestName();
+ private File statisticsArchiveFile;
+
+ @Before
+ public void setup() throws Exception {
+ Properties properties = new Properties();
+ if (useSSL) {
+ updatePropertiesForSSLCache(properties);
+ }
+
+ CacheFactory cacheFactory = new CacheFactory(properties);
+ cacheFactory.set(ConfigurationProperties.MCAST_PORT, "0");
+ cacheFactory.set(ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION, "false");
+ cacheFactory.set(ConfigurationProperties.USE_CLUSTER_CONFIGURATION, "false");
+ statisticsArchiveFile =
+ new File(getClass().getSimpleName() + "_" + testName.getMethodName() + ".gfs");
+ cacheFactory.set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE,
+ statisticsArchiveFile.getName());
+ cache = cacheFactory.create();
+
+ CacheServer cacheServer = cache.addCacheServer();
+ cacheServerPort = AvailablePortHelper.getRandomAvailableTCPPort();
+ cacheServer.setPort(cacheServerPort);
+ cacheServer.start();
+
+ RegionFactory<Object, Object> regionFactory = cache.createRegionFactory();
+ regionFactory.create(TEST_REGION);
+
+ System.setProperty("geode.feature-protobuf-protocol", "true");
+
+ if (useSSL) {
+ socket = getSSLSocket();
+ } else {
+ socket = new Socket("localhost", cacheServerPort);
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(socket::isConnected);
+ outputStream = socket.getOutputStream();
+ outputStream.write(110);
+
+ serializationService = new ProtobufSerializationService();
+ }
+
+ @After
+ public void cleanup() throws IOException {
+ cache.close();
+ socket.close();
+ SocketCreatorFactory.close();
+ }
+
+ @Test
+ public void testNewProtocolHeaderLeadsToNewProtocolServerConnection() throws Exception {
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ ClientProtocol.Message putMessage =
+ MessageUtil.makePutRequestMessage(serializationService, TEST_KEY, TEST_VALUE, TEST_REGION,
+ ProtobufUtilities.createMessageHeader(TEST_PUT_CORRELATION_ID));
+ protobufProtocolSerializer.serialize(putMessage, outputStream);
+ validatePutResponse(socket, protobufProtocolSerializer);
+
+ ClientProtocol.Message getMessage = MessageUtil.makeGetRequestMessage(serializationService,
+ TEST_KEY, TEST_REGION, ProtobufUtilities.createMessageHeader(TEST_GET_CORRELATION_ID));
+ protobufProtocolSerializer.serialize(getMessage, outputStream);
+ validateGetResponse(socket, protobufProtocolSerializer, TEST_VALUE);
+
+ long startTime = System.currentTimeMillis();
+ Thread.sleep(3000);
+
+ long endTime = System.currentTimeMillis();
+
+ disconnectAllFromDS();
+
+ StatArchiveReader.ValueFilter filter = new StatArchiveReader.ValueFilter() {
+ @Override
+ public boolean archiveMatches(File archive) {
+ return true;
+ }
+
+ @Override
+ public boolean typeMatches(String type) {
+ return type.equals("ProtobufServerStats");
+ }
+
+ @Override
+ public boolean statMatches(String statName) {
+ return true;
+ }
+
+ @Override
+ public boolean instanceMatches(String textId, long numericId) {
+ return true;
+ }
+ };
+
+ StatArchiveReader reader = new StatArchiveReader(new File[] {statisticsArchiveFile},
+ new StatArchiveReader.ValueFilter[] {filter}, true);
+ List resourceInstList = reader.getResourceInstList();
+
+ assertEquals(1, resourceInstList.size());
+ StatArchiveReader.ResourceInst resourceInst =
+ (StatArchiveReader.ResourceInst) resourceInstList.iterator().next();
+ StatArchiveReader.StatValue statValue =
+ resourceInst.getStatValue("currentClientConnections").createTrimmed(startTime, endTime);
+ assertEquals(2.0, statValue.getSnapshotsMinimum(), 0.01);
+ }
+
+ @Test
+ public void testConnectionCountIsProperlyDecremented() throws Exception {
+ CacheServer cacheServer = this.cache.getCacheServers().stream().findFirst().get();
+ AcceptorImpl acceptor = ((CacheServerImpl) cacheServer).getAcceptor();
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> acceptor.getClientServerCnxCount() == 1);
+
+ // make a request to the server
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ ClientProtocol.Message getMessage = MessageUtil.makeGetRequestMessage(serializationService,
+ TEST_KEY, TEST_REGION, ProtobufUtilities.createMessageHeader(TEST_GET_CORRELATION_ID));
+ protobufProtocolSerializer.serialize(getMessage, outputStream);
+
+ // make sure socket is still open
+ assertFalse(socket.isClosed());
+ socket.close();
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> acceptor.getClientServerCnxCount() == 0);
+ }
+
+ private void validatePutResponse(Socket socket,
+ ProtobufProtocolSerializer protobufProtocolSerializer) throws Exception {
+ ClientProtocol.Response response =
+ deserializeResponse(socket, protobufProtocolSerializer, TEST_PUT_CORRELATION_ID);
+ assertEquals(ClientProtocol.Response.ResponseAPICase.PUTRESPONSE,
+ response.getResponseAPICase());
+ }
+
+ private void validateGetResponse(Socket socket,
+ ProtobufProtocolSerializer protobufProtocolSerializer, Object expectedValue)
+ throws InvalidProtocolMessageException, IOException, UnsupportedEncodingTypeException,
+ CodecNotRegisteredForTypeException, CodecAlreadyRegisteredForTypeException {
+ ClientProtocol.Response response =
+ deserializeResponse(socket, protobufProtocolSerializer, TEST_GET_CORRELATION_ID);
+
+ assertEquals(ClientProtocol.Response.ResponseAPICase.GETRESPONSE,
+ response.getResponseAPICase());
+ RegionAPI.GetResponse getResponse = response.getGetResponse();
+ BasicTypes.EncodedValue result = getResponse.getResult();
+ assertEquals(BasicTypes.EncodedValue.ValueCase.STRINGRESULT, result.getValueCase());
+ assertEquals(expectedValue, result.getStringResult());
+ }
+
+ private ClientProtocol.Response deserializeResponse(Socket socket,
+ ProtobufProtocolSerializer protobufProtocolSerializer, int expectedCorrelationId)
+ throws InvalidProtocolMessageException, IOException {
+ ClientProtocol.Message message =
+ protobufProtocolSerializer.deserialize(socket.getInputStream());
+ assertEquals(expectedCorrelationId, message.getMessageHeader().getCorrelationId());
+ assertEquals(ClientProtocol.Message.MessageTypeCase.RESPONSE, message.getMessageTypeCase());
+ return message.getResponse();
+ }
+
+ private void updatePropertiesForSSLCache(Properties properties) {
+ String keyStore = TestUtil.getResourcePath(CacheConnectionJUnitTest.class, DEFAULT_STORE);
+ String trustStore = TestUtil.getResourcePath(CacheConnectionJUnitTest.class, DEFAULT_STORE);
+
+ properties.put(SSL_ENABLED_COMPONENTS, "server");
+ properties.put(ConfigurationProperties.SSL_PROTOCOLS, SSL_PROTOCOLS);
+ properties.put(ConfigurationProperties.SSL_CIPHERS, SSL_CIPHERS);
+ properties.put(SSL_REQUIRE_AUTHENTICATION, String.valueOf(true));
+
+ properties.put(SSL_KEYSTORE_TYPE, "jks");
+ properties.put(SSL_KEYSTORE, keyStore);
+ properties.put(SSL_KEYSTORE_PASSWORD, "password");
+ properties.put(SSL_TRUSTSTORE, trustStore);
+ properties.put(SSL_TRUSTSTORE_PASSWORD, "password");
+ }
+
+ private Socket getSSLSocket() throws IOException {
+ String keyStorePath = TestUtil.getResourcePath(CacheConnectionJUnitTest.class, DEFAULT_STORE);
+ String trustStorePath = TestUtil.getResourcePath(CacheConnectionJUnitTest.class, DEFAULT_STORE);
+
+ SSLConfig sslConfig = new SSLConfig();
+ sslConfig.setEnabled(true);
+ sslConfig.setCiphers(SSL_CIPHERS);
+ sslConfig.setProtocols(SSL_PROTOCOLS);
+ sslConfig.setRequireAuth(true);
+ sslConfig.setKeystoreType("jks");
+ sslConfig.setKeystore(keyStorePath);
+ sslConfig.setKeystorePassword("password");
+ sslConfig.setTruststore(trustStorePath);
+ sslConfig.setKeystorePassword("password");
+
+ SocketCreator socketCreator = new SocketCreator(sslConfig);
+ return socketCreator.connectForClient("localhost", cacheServerPort, 5000);
+ }
+
+
+}
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionTimeoutJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionTimeoutJUnitTest.java
new file mode 100644
index 0000000..cda7e7b
--- /dev/null
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheConnectionTimeoutJUnitTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache 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://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.protocol.acceptance;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.internal.AvailablePortHelper;
+import org.apache.geode.internal.cache.tier.sockets.ClientHealthMonitor;
+import org.apache.geode.internal.net.SocketCreatorFactory;
+import org.apache.geode.internal.protocol.protobuf.ClientProtocol;
+import org.apache.geode.protocol.MessageUtil;
+import org.apache.geode.protocol.protobuf.ProtobufSerializationService;
+import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
+import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
+import org.apache.geode.serialization.SerializationService;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+
+/**
+ * Test the new protocol correctly times out connections
+ */
+@Category(IntegrationTest.class)
+public class CacheConnectionTimeoutJUnitTest {
+ private final String TEST_KEY = "testKey";
+ private final String TEST_VALUE = "testValue";
+ private final String TEST_REGION = "testRegion";
+ private final int TEST_PUT_CORRELATION_ID = 574;
+
+ private Cache cache;
+ private int cacheServerPort;
+ private SerializationService serializationService;
+ private Socket socket;
+ private OutputStream outputStream;
+
+ @Rule
+ public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Rule
+ public TestName testName = new TestName();
+
+ @Before
+ public void setup() throws Exception {
+ Properties properties = new Properties();
+
+ CacheFactory cacheFactory = new CacheFactory(properties);
+ cacheFactory.set(ConfigurationProperties.MCAST_PORT, "0");
+ cacheFactory.set(ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION, "false");
+ cacheFactory.set(ConfigurationProperties.USE_CLUSTER_CONFIGURATION, "false");
+
+ cache = cacheFactory.create();
+
+ CacheServer cacheServer = cache.addCacheServer();
+ cacheServerPort = AvailablePortHelper.getRandomAvailableTCPPort();
+ cacheServer.setPort(cacheServerPort);
+ cacheServer.setMaximumTimeBetweenPings(100);
+ cacheServer.start();
+
+ RegionFactory<Object, Object> regionFactory = cache.createRegionFactory();
+ regionFactory.create(TEST_REGION);
+
+ System.setProperty("geode.feature-protobuf-protocol", "true");
+ System.setProperty(ClientHealthMonitor.CLIENT_HEALTH_MONITOR_INTERVAL_PROPERTY, "100");
+
+ socket = new Socket("localhost", cacheServerPort);
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(socket::isConnected);
+ outputStream = socket.getOutputStream();
+ outputStream.write(110);
+
+ serializationService = new ProtobufSerializationService();
+ }
+
+ @After
+ public void cleanup() throws IOException {
+ cache.close();
+ socket.close();
+ SocketCreatorFactory.close();
+ }
+
+ @Test
+ public void testUnresponsiveClientsGetDisconnected() throws Exception {
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ ClientProtocol.Message putMessage =
+ MessageUtil.makePutRequestMessage(serializationService, TEST_KEY, TEST_VALUE, TEST_REGION,
+ ProtobufUtilities.createMessageHeader(TEST_PUT_CORRELATION_ID));
+
+ int pollInterval = 20;
+ int maximumTimeBetweenPings = ClientHealthMonitor.getInstance().getMaximumTimeBetweenPings();
+ long monitorInterval = ClientHealthMonitor.getInstance().getMonitorInterval();
+ long timeout = maximumTimeBetweenPings + monitorInterval + pollInterval;
+
+ // wait for client to get disconnected
+ Awaitility.await().atMost(timeout, TimeUnit.MILLISECONDS)
+ .pollInterval(pollInterval, TimeUnit.MILLISECONDS)
+ .pollDelay(maximumTimeBetweenPings + monitorInterval, TimeUnit.MILLISECONDS).until(() -> {
+ try {
+ /*
+ * send a PUT message
+ *
+ * Note: The `await` will run this at an interval larger than the maximum timeout
+ * allowed between pings. This is so that we have better validation that we are actually
+ * timing out connections after `maximumTimeBetweenPings` and not some other larger time
+ * that's smaller than `timeout`
+ */
+ protobufProtocolSerializer.serialize(putMessage, outputStream);
+ assertEquals(-1, socket.getInputStream().read());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ });
+ }
+
+ @Test
+ public void testResponsiveClientsStaysConnected() throws Exception {
+ ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
+ ClientProtocol.Message putMessage =
+ MessageUtil.makePutRequestMessage(serializationService, TEST_KEY, TEST_VALUE, TEST_REGION,
+ ProtobufUtilities.createMessageHeader(TEST_PUT_CORRELATION_ID));
+
+ int timeout = 1500;
+ int interval = 100;
+ for (int i = 0; i < timeout; i += interval) {
+ // send a PUT message
+ protobufProtocolSerializer.serialize(putMessage, outputStream);
+ assertNotEquals(-1, socket.getInputStream().read());
+ Thread.sleep(interval);
+ }
+ }
+}
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripCacheConnectionJUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheOperationsJUnitTest.java
similarity index 71%
rename from geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripCacheConnectionJUnitTest.java
rename to geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheOperationsJUnitTest.java
index 4a6b44a..127ef2e 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripCacheConnectionJUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/CacheOperationsJUnitTest.java
@@ -13,7 +13,7 @@
* the License.
*/
-package org.apache.geode.protocol;
+package org.apache.geode.protocol.acceptance;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENABLED_COMPONENTS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE;
@@ -22,18 +22,15 @@ import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_
import static org.apache.geode.distributed.ConfigurationProperties.SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_PASSWORD;
-import static org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase.disconnectAllFromDS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
-import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Collection;
import java.util.HashSet;
-import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -58,16 +55,13 @@ import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.admin.SSLConfig;
-import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.tier.CommunicationMode;
-import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
-import org.apache.geode.internal.cache.tier.sockets.GenericProtocolServerConnection;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.internal.protocol.protobuf.BasicTypes;
import org.apache.geode.internal.protocol.protobuf.ClientProtocol;
import org.apache.geode.internal.protocol.protobuf.RegionAPI;
-import org.apache.geode.internal.statistics.StatArchiveReader;
+import org.apache.geode.protocol.MessageUtil;
import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
import org.apache.geode.protocol.protobuf.ProtobufSerializationService;
import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
@@ -81,13 +75,11 @@ import org.apache.geode.test.junit.categories.IntegrationTest;
import org.apache.geode.util.test.TestUtil;
/**
- * Test that switching on the header byte makes instances of
- * {@link GenericProtocolServerConnection}.
+ * Test operations using ProtoBuf
*/
@Category(IntegrationTest.class)
-public class RoundTripCacheConnectionJUnitTest {
+public class CacheOperationsJUnitTest {
private final String TEST_KEY = "testKey";
- private final String TEST_VALUE = "testValue";
private final String TEST_REGION = "testRegion";
private final int TEST_PUT_CORRELATION_ID = 574;
private final int TEST_GET_CORRELATION_ID = 68451;
@@ -115,7 +107,6 @@ public class RoundTripCacheConnectionJUnitTest {
@Rule
public TestName testName = new TestName();
- private File statisticsArchiveFile;
@Before
public void setup() throws Exception {
@@ -131,12 +122,6 @@ public class RoundTripCacheConnectionJUnitTest {
cacheFactory.set(ConfigurationProperties.MCAST_PORT, "0");
cacheFactory.set(ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION, "false");
cacheFactory.set(ConfigurationProperties.USE_CLUSTER_CONFIGURATION, "false");
- cacheFactory.set(ConfigurationProperties.STATISTIC_SAMPLING_ENABLED, "true");
- cacheFactory.set(ConfigurationProperties.STATISTIC_SAMPLE_RATE, "100");
- statisticsArchiveFile =
- new File(getClass().getSimpleName() + "_" + testName.getMethodName() + ".gfs");
- cacheFactory.set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE,
- statisticsArchiveFile.getName());
cache = cacheFactory.create();
CacheServer cacheServer = cache.addCacheServer();
@@ -169,21 +154,6 @@ public class RoundTripCacheConnectionJUnitTest {
}
@Test
- public void testNewProtocolHeaderLeadsToNewProtocolServerConnection() throws Exception {
- ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
- ClientProtocol.Message putMessage =
- MessageUtil.makePutRequestMessage(serializationService, TEST_KEY, TEST_VALUE, TEST_REGION,
- ProtobufUtilities.createMessageHeader(TEST_PUT_CORRELATION_ID));
- protobufProtocolSerializer.serialize(putMessage, outputStream);
- validatePutResponse(socket, protobufProtocolSerializer);
-
- ClientProtocol.Message getMessage = MessageUtil.makeGetRequestMessage(serializationService,
- TEST_KEY, TEST_REGION, ProtobufUtilities.createMessageHeader(TEST_GET_CORRELATION_ID));
- protobufProtocolSerializer.serialize(getMessage, outputStream);
- validateGetResponse(socket, protobufProtocolSerializer, TEST_VALUE);
- }
-
- @Test
public void testNewProtocolWithMultikeyOperations() throws Exception {
System.setProperty("geode.feature-protobuf-protocol", "true");
@@ -219,51 +189,6 @@ public class RoundTripCacheConnectionJUnitTest {
ProtobufUtilities.createProtobufRequestWithGetAllRequest(getAllRequest));
protobufProtocolSerializer.serialize(getAllMessage, outputStream);
validateGetAllResponse(socket, protobufProtocolSerializer);
- long startTime = System.currentTimeMillis();
- Thread.sleep(3000);
-
- long endTime = System.currentTimeMillis();
-
- disconnectAllFromDS();
-
- StatArchiveReader.ValueFilter filter = new StatArchiveReader.ValueFilter() {
- @Override
- public boolean archiveMatches(File archive) {
- return true;
- }
-
- @Override
- public boolean typeMatches(String type) {
- return type.equals("ProtobufServerStats");
- }
-
- @Override
- public boolean statMatches(String statName) {
- return true;
- }
-
- @Override
- public boolean instanceMatches(String textId, long numericId) {
- return true;
- }
- };
-
- StatArchiveReader reader = new StatArchiveReader(new File[] {statisticsArchiveFile},
- new StatArchiveReader.ValueFilter[] {filter}, true);
- List resourceInstList = reader.getResourceInstList();
- // for (Object inst : resourceInstList) {
- // StatArchiveReader.ResourceInst ri = (StatArchiveReader.ResourceInst) inst;
- // String resourceName = ri.getName();
- // String resourceTypeName = ri.getType().getName();
- // System.out.println("===> resource name: " + resourceName + "; type name: " +
- // resourceTypeName);
- // }
- assertEquals(1, resourceInstList.size());
- StatArchiveReader.ResourceInst resourceInst =
- (StatArchiveReader.ResourceInst) resourceInstList.iterator().next();
- StatArchiveReader.StatValue statValue =
- resourceInst.getStatValue("currentClientConnections").createTrimmed(startTime, endTime);
- assertEquals(2.0, statValue.getSnapshotsMinimum(), 0.01);
}
@Test
@@ -314,7 +239,7 @@ public class RoundTripCacheConnectionJUnitTest {
}
@Test
- public void testNullResponse() throws Exception {
+ public void testResponseToGetWithNoData() throws Exception {
// Get request without any data set must return a null
ProtobufProtocolSerializer protobufProtocolSerializer = new ProtobufProtocolSerializer();
ClientProtocol.Message getMessage = MessageUtil.makeGetRequestMessage(serializationService,
@@ -331,85 +256,6 @@ public class RoundTripCacheConnectionJUnitTest {
}
@Test
- public void testConnectionCountIsProperlyDecremented() throws Exception {
- CacheServer cacheServer = this.cache.getCacheServers().stream().findFirst().get();
- AcceptorImpl acceptor = ((CacheServerImpl) cacheServer).getAcceptor();
- Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> {
- return acceptor.getClientServerCnxCount() == 1;
- });
- // run another test that creates a connection to the server
- testNewProtocolGetRegionNamesCallSucceeds();
- assertFalse(socket.isClosed());
- socket.close();
- Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> {
- return acceptor.getClientServerCnxCount() == 0;
- });
- }
-
- @Test
- public void testNewProtocolRespectsMaxConnectionLimit() throws IOException, InterruptedException {
- cache.getDistributedSystem().disconnect();
-
- CacheFactory cacheFactory = new CacheFactory();
- cacheFactory.set(ConfigurationProperties.LOCATORS, "");
- cacheFactory.set(ConfigurationProperties.MCAST_PORT, "0");
- cacheFactory.set(ConfigurationProperties.STATISTIC_SAMPLING_ENABLED, "true");
- cacheFactory.set(ConfigurationProperties.STATISTIC_SAMPLE_RATE, "100");
- cacheFactory.set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE,
- getClass().getSimpleName() + "_" + testName.getMethodName() + ".gfs");
- cache = cacheFactory.create();
-
- CacheServer cacheServer = cache.addCacheServer();
- final int cacheServerPort = AvailablePortHelper.getRandomAvailableTCPPort();
- cacheServer.setPort(cacheServerPort);
- cacheServer.setMaxConnections(16);
- cacheServer.setMaxThreads(16);
- cacheServer.start();
-
- AcceptorImpl acceptor = ((CacheServerImpl) cacheServer).getAcceptor();
-
- // Start 16 sockets, which is exactly the maximum that the server will support.
- Socket[] sockets = new Socket[16];
- for (int i = 0; i < 16; i++) {
- Socket socket = new Socket("localhost", cacheServerPort);
- sockets[i] = socket;
- Awaitility.await().atMost(5, TimeUnit.SECONDS).until(socket::isConnected);
- socket.getOutputStream()
- .write(CommunicationMode.ProtobufClientServerProtocol.getModeNumber());
- }
-
- // try to start a new socket, expecting it to be disconnected.
- try (Socket socket = new Socket("localhost", cacheServerPort)) {
- Awaitility.await().atMost(5, TimeUnit.SECONDS).until(socket::isConnected);
- socket.getOutputStream()
- .write(CommunicationMode.ProtobufClientServerProtocol.getModeNumber());
- assertEquals(-1, socket.getInputStream().read()); // EOF implies disconnected.
- }
-
- Thread.sleep(15000);
- for (Socket currentSocket : sockets) {
- currentSocket.close();
- }
-
- // Once all connections are closed, the acceptor should have a connection count of 0.
- Awaitility.await().atMost(5, TimeUnit.SECONDS)
- .until(() -> acceptor.getClientServerCnxCount() == 0);
-
- // Try to start 16 new connections, again at the limit.
- for (int i = 0; i < 16; i++) {
- Socket socket = new Socket("localhost", cacheServerPort);
- sockets[i] = socket;
- Awaitility.await().atMost(5, TimeUnit.SECONDS).until(socket::isConnected);
- socket.getOutputStream()
- .write(CommunicationMode.ProtobufClientServerProtocol.getModeNumber());
- }
-
- for (Socket currentSocket : sockets) {
- currentSocket.close();
- }
- }
-
- @Test
public void testNewProtocolGetRegionNamesCallSucceeds() throws Exception {
int correlationId = TEST_GET_CORRELATION_ID; // reuse this value for this test
@@ -425,12 +271,7 @@ public class RoundTripCacheConnectionJUnitTest {
}
@Test
- public void useSSL_testNewProtocolHeaderLeadsToNewProtocolServerConnection() throws Exception {
- testNewProtocolHeaderLeadsToNewProtocolServerConnection();
- }
-
- @Test
- public void testNewProtocolGetRegionCallSucceeds() throws Exception {
+ public void testNewProtocolGetRegionCall() throws Exception {
System.setProperty("geode.feature-protobuf-protocol", "true");
Socket socket = new Socket("localhost", cacheServerPort);
@@ -460,14 +301,6 @@ public class RoundTripCacheConnectionJUnitTest {
assertEquals(Scope.DISTRIBUTED_NO_ACK, Scope.fromString(region.getScope()));
}
- private void validatePutResponse(Socket socket,
- ProtobufProtocolSerializer protobufProtocolSerializer) throws Exception {
- ClientProtocol.Response response =
- deserializeResponse(socket, protobufProtocolSerializer, TEST_PUT_CORRELATION_ID);
- assertEquals(ClientProtocol.Response.ResponseAPICase.PUTRESPONSE,
- response.getResponseAPICase());
- }
-
private void validateGetResponse(Socket socket,
ProtobufProtocolSerializer protobufProtocolSerializer, Object expectedValue)
throws InvalidProtocolMessageException, IOException, UnsupportedEncodingTypeException,
@@ -561,10 +394,8 @@ public class RoundTripCacheConnectionJUnitTest {
}
private void updatePropertiesForSSLCache(Properties properties) {
- String keyStore =
- TestUtil.getResourcePath(RoundTripCacheConnectionJUnitTest.class, DEFAULT_STORE);
- String trustStore =
- TestUtil.getResourcePath(RoundTripCacheConnectionJUnitTest.class, DEFAULT_STORE);
+ String keyStore = TestUtil.getResourcePath(CacheOperationsJUnitTest.class, DEFAULT_STORE);
+ String trustStore = TestUtil.getResourcePath(CacheOperationsJUnitTest.class, DEFAULT_STORE);
properties.put(SSL_ENABLED_COMPONENTS, "server");
properties.put(ConfigurationProperties.SSL_PROTOCOLS, SSL_PROTOCOLS);
@@ -579,10 +410,8 @@ public class RoundTripCacheConnectionJUnitTest {
}
private Socket getSSLSocket() throws IOException {
- String keyStorePath =
- TestUtil.getResourcePath(RoundTripCacheConnectionJUnitTest.class, DEFAULT_STORE);
- String trustStorePath =
- TestUtil.getResourcePath(RoundTripCacheConnectionJUnitTest.class, DEFAULT_STORE);
+ String keyStorePath = TestUtil.getResourcePath(CacheOperationsJUnitTest.class, DEFAULT_STORE);
+ String trustStorePath = TestUtil.getResourcePath(CacheOperationsJUnitTest.class, DEFAULT_STORE);
SSLConfig sslConfig = new SSLConfig();
sslConfig.setEnabled(true);
diff --git a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionDUnitTest.java b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/LocatorConnectionDUnitTest.java
similarity index 93%
rename from geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionDUnitTest.java
rename to geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/LocatorConnectionDUnitTest.java
index 9603caf..b4d3981 100644
--- a/geode-protobuf/src/test/java/org/apache/geode/protocol/RoundTripLocatorConnectionDUnitTest.java
+++ b/geode-protobuf/src/test/java/org/apache/geode/protocol/acceptance/LocatorConnectionDUnitTest.java
@@ -13,7 +13,7 @@
* the License.
*/
-package org.apache.geode.protocol;
+package org.apache.geode.protocol.acceptance;
import static org.apache.geode.internal.cache.tier.CommunicationMode.ProtobufClientServerProtocol;
import static org.junit.Assert.assertEquals;
@@ -22,6 +22,7 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.Socket;
+import java.util.Properties;
import org.junit.Before;
import org.junit.Rule;
@@ -31,13 +32,11 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.Locator;
-import org.apache.geode.internal.Config;
import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
import org.apache.geode.internal.protocol.protobuf.ClientProtocol;
-import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
import org.apache.geode.internal.protocol.protobuf.ServerAPI;
+import org.apache.geode.protocol.exception.InvalidProtocolMessageException;
+import org.apache.geode.protocol.protobuf.ProtocolErrorCode;
import org.apache.geode.protocol.protobuf.serializer.ProtobufProtocolSerializer;
import org.apache.geode.protocol.protobuf.utilities.ProtobufRequestUtilities;
import org.apache.geode.protocol.protobuf.utilities.ProtobufUtilities;
@@ -45,19 +44,12 @@ import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.DistributedTest;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.contrib.java.lang.system.RestoreSystemProperties;
-import org.junit.experimental.categories.Category;
-
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.net.Socket;
-import java.util.Properties;
+/*
+ * Test sending ProtoBuf messages to the locator
+ */
@Category(DistributedTest.class)
-public class RoundTripLocatorConnectionDUnitTest extends JUnit4CacheTestCase {
+public class LocatorConnectionDUnitTest extends JUnit4CacheTestCase {
private Socket socket;
diff --git a/geode-protobuf/src/test/resources/org/apache/geode/protocol/default.keystore b/geode-protobuf/src/test/resources/org/apache/geode/protocol/acceptance/default.keystore
similarity index 100%
rename from geode-protobuf/src/test/resources/org/apache/geode/protocol/default.keystore
rename to geode-protobuf/src/test/resources/org/apache/geode/protocol/acceptance/default.keystore
--
To stop receiving notification emails like this one, please contact
['"commits@geode.apache.org" <co...@geode.apache.org>'].