You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by jj...@apache.org on 2020/05/08 13:12:48 UTC
[geode] branch develop updated: GEODE-7565: Allow gateway receivers
with same host and port (2nd try)
This is an automated email from the ASF dual-hosted git repository.
jjramos 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 be87788 GEODE-7565: Allow gateway receivers with same host and port (2nd try)
be87788 is described below
commit be877883f3e123ba3c00af7d01464c2c80785522
Author: Alberto Bustamante Reyes <al...@users.noreply.github.com>
AuthorDate: Fri May 8 15:11:48 2020 +0200
GEODE-7565: Allow gateway receivers with same host and port (2nd try)
This reverts the revert of the original commit, adds extra tests and
also fixes the originally introduced issues.
----
There was a problem with Geode WAN replication when GW receivers are
configured with the same hostname-for-senders and port on all servers.
The reason for such a setup is deploying Geode cluster on a Kubernetes
cluster where all GW receivers are reachable from the outside world on
the same VIP and port.
The problem experienced is that shutting down one server is stopping
replication to this cluster until the server is up again. This is
because Geode incorrectly assumes there are no more alive servers when
just one of them is down, because since they share hostname-for-senders
and port, they are treated as one same server.
With these changes locator is able to distinguish the different
receivers using the same hostname and port so replication is not
impacted when one server is stopped.
---
.../cache/tier/sockets/PingOpDistributedTest.java | 202 ++++++
.../LocatorLoadSnapshotIntegrationTest.java | 14 +-
.../codeAnalysis/sanctionedDataSerializables.txt | 4 +
.../client/internal/CacheServerLoadMessage.java | 4 +-
.../cache/client/internal/LiveServerPinger.java | 2 +-
.../apache/geode/cache/client/internal/PingOp.java | 20 +-
.../geode/cache/client/internal/ServerProxy.java | 10 -
.../distributed/internal/LocatorLoadSnapshot.java | 168 +++--
.../internal/ServerLocationAndMemberId.java | 65 ++
.../geode/distributed/internal/ServerLocator.java | 14 +-
.../org/apache/geode/internal/DSFIDFactory.java | 2 +
.../internal/cache/DistributedPingMessage.java | 86 +++
.../apache/geode/internal/cache/GridAdvisor.java | 14 +-
.../internal/cache/tier/sockets/command/Ping.java | 68 ++-
.../internal/LocatorLoadSnapshotJUnitTest.java | 675 +++++++++++++++++++--
.../internal/ServerLocationAndMemberIdTest.java | 69 +++
.../internal/cache/DistributedPingMessageTest.java | 85 +++
.../cache/tier/sockets/command/PingTest.java | 145 +++++
.../sockets/ClientServerMiscDUnitTestBase.java | 19 +
.../serialization/DataSerializableFixedID.java | 2 +
20 files changed, 1532 insertions(+), 136 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/PingOpDistributedTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/PingOpDistributedTest.java
new file mode 100644
index 0000000..3b41e75
--- /dev/null
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/PingOpDistributedTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.internal.cache.tier.sockets;
+
+import static java.util.Arrays.asList;
+import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPortsForDUnitSite;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.PoolFactory;
+import org.apache.geode.cache.client.PoolManager;
+import org.apache.geode.cache.client.ServerOperationException;
+import org.apache.geode.cache.client.internal.PingOp;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.internal.ServerLocation;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.ClientCacheRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+public class PingOpDistributedTest implements Serializable {
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public ClientCacheRule clientCacheRule = new ClientCacheRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule(2);
+
+ @Rule
+ public SerializableTemporaryFolder folder = new SerializableTemporaryFolder();
+
+ private VM client;
+ private VM server1, server2;
+ private int server1Port, server2Port;
+
+
+ private void initServer(int serverPort) throws IOException {
+ cacheRule.createCache();
+ CacheServer cacheServer = cacheRule.getCache().addCacheServer();
+ cacheServer.setPort(serverPort);
+
+ // "Disable" the auto-ping for the duration of this test.
+ cacheServer.setMaximumTimeBetweenPings((int) GeodeAwaitility.getTimeout().toMillis());
+ cacheServer.start();
+ }
+
+ private void initClient(String poolName, List<Integer> serverPorts) {
+ final ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
+ clientCacheFactory.create();
+
+ PoolFactory poolFactory = PoolManager.createFactory();
+ serverPorts.forEach(serverPort -> poolFactory.addServer("localhost", serverPort));
+
+ // "Disable" the auto-ping for the duration of this test.
+ poolFactory.setPingInterval((int) GeodeAwaitility.getTimeout().toMillis());
+ poolFactory.create(poolName);
+ }
+
+ @Before
+ public void setUp() throws IOException {
+ int[] ports = getRandomAvailableTCPPortsForDUnitSite(2);
+
+ client = getVM(0);
+ server1 = getVM(1);
+ server2 = getVM(2);
+ server1Port = ports[0];
+ server2Port = ports[1];
+ server1.invoke(() -> initServer(server1Port));
+ server2.invoke(() -> initServer(server2Port));
+ }
+
+ void parametrizedSetUp(String poolName, List<Integer> serverPorts) {
+ client.invoke(() -> initClient(poolName, serverPorts));
+ }
+
+ public void executePing(String poolName, int serverPort,
+ InternalDistributedMember distributedMember) {
+ PoolImpl poolImpl = (PoolImpl) PoolManager.find(poolName);
+ PingOp.execute(poolImpl, new ServerLocation("localhost", serverPort), distributedMember);
+ }
+
+ public Long getSingleHeartBeat() {
+ ClientHealthMonitor chm = ClientHealthMonitor.getInstance();
+ if (chm.getClientHeartbeats().size() == 0) {
+ return 0L;
+ }
+ assertThat(chm.getClientHeartbeats()).isNotEmpty().hasSize(1);
+
+ return chm.getClientHeartbeats().entrySet().iterator().next().getValue();
+ }
+
+ @Test
+ public void regularPingFlow() {
+ final String poolName = testName.getMethodName();
+ parametrizedSetUp(poolName, Collections.singletonList(server1Port));
+ InternalDistributedMember distributedMember1 = (InternalDistributedMember) server1
+ .invoke(() -> cacheRule.getCache().getDistributedSystem().getDistributedMember());
+
+ client.invoke(() -> executePing(poolName, server1Port, distributedMember1));
+ Long firstHeartbeat = server1.invoke(this::getSingleHeartBeat);
+
+ client.invoke(() -> executePing(poolName, server1Port, distributedMember1));
+ Long secondHeartbeat = server1.invoke(this::getSingleHeartBeat);
+
+ assertThat(secondHeartbeat).isGreaterThan(firstHeartbeat);
+
+ }
+
+ @Test
+ public void memberShouldNotRedirectPingMessageWhenClientCachedViewIdIsWrong() {
+ final String poolName = testName.getMethodName();
+ parametrizedSetUp(poolName, Collections.singletonList(server1Port));
+ InternalDistributedMember distributedMember1 = (InternalDistributedMember) server1
+ .invoke(() -> cacheRule.getCache().getDistributedSystem().getDistributedMember());
+
+ client.invoke(() -> {
+ PoolImpl poolImpl = (PoolImpl) PoolManager.find(poolName);
+ distributedMember1.setVmViewId(distributedMember1.getVmViewId() + 1);
+ assertThatThrownBy(() -> {
+ PingOp.execute(poolImpl, new ServerLocation("localhost", server1Port), distributedMember1);
+ }).isInstanceOf(ServerOperationException.class).hasMessageContaining("has different viewId:");
+ });
+ }
+
+ @Test
+ public void pingReturnsErrorIfTheTargetServerIsNotAMember() {
+ final String poolName = testName.getMethodName();
+ parametrizedSetUp(poolName, Collections.singletonList(server1Port));
+ int notUsedPort = getRandomAvailableTCPPortsForDUnitSite(1)[0];
+ InternalDistributedMember fakeDistributedMember =
+ new InternalDistributedMember("localhost", notUsedPort);
+ client.invoke(() -> {
+ PoolImpl poolImpl = (PoolImpl) PoolManager.find(poolName);
+ assertThatThrownBy(() -> {
+ PingOp.execute(poolImpl, new ServerLocation("localhost", server1Port),
+ fakeDistributedMember);
+ }).isInstanceOf(ServerOperationException.class)
+ .hasMessageContaining("Unable to ping non-member");
+ });
+ }
+
+ @Test
+ public void memberShouldCorrectlyRedirectPingMessage() {
+ final String poolName = testName.getMethodName();
+ parametrizedSetUp(poolName, asList(server1Port, server2Port));
+ InternalDistributedMember distributedMember1 = (InternalDistributedMember) server1
+ .invoke(() -> cacheRule.getCache().getDistributedSystem().getDistributedMember());
+ InternalDistributedMember distributedMember2 = (InternalDistributedMember) server2
+ .invoke(() -> cacheRule.getCache().getDistributedSystem().getDistributedMember());
+
+ // Run two correct pings to ensure both ClientHealthMonitors have a valid value
+ client.invoke(() -> executePing(poolName, server1Port, distributedMember1));
+ client.invoke(() -> executePing(poolName, server2Port, distributedMember2));
+
+ Long firstHeartbeatServer1 = server1.invoke(this::getSingleHeartBeat);
+ Long firstHeartbeatServer2 = server2.invoke(this::getSingleHeartBeat);
+
+ // Ping to be forwarded from server1 to server2
+ client.invoke(() -> executePing(poolName, server1Port, distributedMember2));
+
+ Long secondHeartbeatServer1 = server1.invoke(this::getSingleHeartBeat);
+ Long secondHeartbeatServer2 = server2.invoke(this::getSingleHeartBeat);
+
+ // Heartbeat in server2 changed as the ping was forwarded from server1 to him.
+ assertThat(secondHeartbeatServer1).isEqualTo(firstHeartbeatServer1);
+ assertThat(secondHeartbeatServer2).isGreaterThan(firstHeartbeatServer2);
+ }
+}
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotIntegrationTest.java
index 61bdcec..1d5aa76 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotIntegrationTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotIntegrationTest.java
@@ -27,6 +27,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.server.ServerLoad;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.test.junit.categories.MembershipTest;
/**
@@ -46,19 +47,26 @@ public class LocatorLoadSnapshotIntegrationTest {
final int NUM_REQUESTS = 10000;
int ALLOWED_THRESHOLD = 50; // We should never be off by more than
// the number of concurrent threads.
+ final int LOAD_POLL_INTERVAL = 30000;
final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
final ServerLocation l1 = new ServerLocation("localhost", 1);
final ServerLocation l2 = new ServerLocation("localhost", 2);
final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
int initialLoad1 = (int) (Math.random() * (NUM_REQUESTS / 2));
int initialLoad2 = (int) (Math.random() * (NUM_REQUESTS / 2));
int initialLoad3 = (int) (Math.random() * (NUM_REQUESTS / 2));
- sn.addServer(l1, new String[0], new ServerLoad(initialLoad1, 1, 0, 1));
- sn.addServer(l2, new String[0], new ServerLoad(initialLoad2, 1, 0, 1));
- sn.addServer(l3, new String[0], new ServerLoad(initialLoad3, 1, 0, 1));
+ sn.addServer(l1, uniqueId1, new String[0], new ServerLoad(initialLoad1, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[0], new ServerLoad(initialLoad2, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l3, uniqueId3, new String[0], new ServerLoad(initialLoad3, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
final Map loadCounts = new HashMap();
loadCounts.put(l1, new AtomicInteger(initialLoad1));
diff --git a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
index 09ad609..a2ccc26 100644
--- a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
+++ b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
@@ -939,6 +939,10 @@ org/apache/geode/internal/cache/DistributedClearOperation$ClearRegionWithContext
fromData,15
toData,15
+org/apache/geode/internal/cache/DistributedPingMessage,2
+fromData,26
+toData,23
+
org/apache/geode/internal/cache/DistributedPutAllOperation$EntryVersionsList,2
fromData,272
toData,292
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CacheServerLoadMessage.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CacheServerLoadMessage.java
index ad8412d..3646486 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/CacheServerLoadMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/CacheServerLoadMessage.java
@@ -67,13 +67,11 @@ public class CacheServerLoadMessage extends SerialDistributionMessage {
InternalLocator l = (InternalLocator) locators.get(i);
ServerLocator serverLocator = l.getServerLocatorAdvisee();
if (serverLocator != null) {
- serverLocator.updateLoad(location, load, this.clientIds);
+ serverLocator.updateLoad(location, this.getSender().getUniqueId(), load, this.clientIds);
}
}
}
-
-
@Override
public int getDSFID() {
return CACHE_SERVER_LOAD_MESSAGE;
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/LiveServerPinger.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/LiveServerPinger.java
index 0b13f1b..623d976 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/LiveServerPinger.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/LiveServerPinger.java
@@ -87,7 +87,7 @@ public class LiveServerPinger extends EndpointListenerAdapter {
public void run2() {
if (endpoint.timeToPing(pingIntervalNanos)) {
try {
- PingOp.execute(pool, endpoint.getLocation());
+ PingOp.execute(pool, endpoint.getLocation(), endpoint.getMemberId());
} catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Error occurred while pinging server: {} - {}", endpoint.getLocation(),
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
index 6429098..4140fe8 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PingOp.java
@@ -14,6 +14,7 @@
*/
package org.apache.geode.cache.client.internal;
+import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.Message;
@@ -29,11 +30,12 @@ public class PingOp {
* Ping the specified server to see if it is still alive
*
* @param pool the pool to use to communicate with the server.
- * @param server the server to do the execution on
+ * @param serverLocation the server to do the execution on
*/
- public static void execute(ExecutablePool pool, ServerLocation server) {
- AbstractOp op = new PingOpImpl();
- pool.executeOn(server, op, false, false);
+ public static void execute(ExecutablePool pool, ServerLocation serverLocation,
+ DistributedMember serverID) {
+ AbstractOp op = new PingOpImpl(serverID);
+ pool.executeOn(serverLocation, op, false, false);
}
private PingOp() {
@@ -42,13 +44,14 @@ public class PingOp {
static class PingOpImpl extends AbstractOp {
- private long startTime;
+ private final DistributedMember serverID;
/**
* @throws org.apache.geode.SerializationException if serialization fails
*/
- PingOpImpl() {
+ PingOpImpl(DistributedMember serverID) {
super(MessageType.PING, 0);
+ this.serverID = serverID;
}
@Override
@@ -65,8 +68,9 @@ public class PingOp {
@Override
protected void sendMessage(Connection cnx) throws Exception {
getMessage().clearMessageHasSecurePartFlag();
- this.startTime = System.currentTimeMillis();
- getMessage().send(false);
+ getMessage().setNumberOfParts(1);
+ getMessage().addObjPart(serverID);
+ getMessage().send(true);
Message.MESSAGE_TYPE.set(MessageType.PING);
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ServerProxy.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ServerProxy.java
index 675a55e5..5f9c7b9 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ServerProxy.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ServerProxy.java
@@ -16,7 +16,6 @@ package org.apache.geode.cache.client.internal;
import org.apache.geode.cache.query.SelectResults;
-import org.apache.geode.distributed.internal.ServerLocation;
/**
* Used to send operations from a client to a server.
@@ -50,15 +49,6 @@ public class ServerProxy {
}
/**
- * Ping the specified server to see if it is still alive
- *
- * @param server the server to do the execution on
- */
- public void ping(ServerLocation server) {
- PingOp.execute(this.pool, server);
- }
-
- /**
* Does a query on a server
*
* @param queryPredicate A query language boolean query predicate
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorLoadSnapshot.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorLoadSnapshot.java
index f0ba83b..7da092f 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorLoadSnapshot.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorLoadSnapshot.java
@@ -30,6 +30,8 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.geode.InternalGemFireException;
+import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.server.ServerLoad;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -50,7 +52,8 @@ public class LocatorLoadSnapshot {
private final Map<ServerLocation, String[]> serverGroupMap = new HashMap<>();
- private final Map<String, Map<ServerLocation, LoadHolder>> connectionLoadMap = new HashMap<>();
+ private final Map<String, Map<ServerLocationAndMemberId, LoadHolder>> connectionLoadMap =
+ new HashMap<>();
private final Map<String, Map<ServerLocation, LoadHolder>> queueLoadMap = new HashMap<>();
@@ -84,29 +87,27 @@ public class LocatorLoadSnapshot {
}
}
- public void addServer(ServerLocation location, String[] groups, ServerLoad initialLoad) {
- addServer(location, groups, initialLoad, 30000);
- }
-
/**
* Add a new server to the load snapshot.
*/
- public synchronized void addServer(ServerLocation location, String[] groups,
+ public synchronized void addServer(ServerLocation location, String memberId, String[] groups,
ServerLoad initialLoad, long loadPollInterval) {
serverGroupMap.put(location, groups);
- LoadHolder connectionLoad = new LoadHolder(location, initialLoad.getConnectionLoad(),
- initialLoad.getLoadPerConnection(), loadPollInterval);
- addGroups(connectionLoadMap, groups, connectionLoad);
- LoadHolder queueLoad = new LoadHolder(location, initialLoad.getSubscriptionConnectionLoad(),
+ LoadHolder connectionLoad =
+ new LoadHolder(location, initialLoad.getConnectionLoad(),
+ initialLoad.getLoadPerConnection(), loadPollInterval);
+ addGroups(connectionLoadMap, groups, connectionLoad, memberId);
+ LoadHolder queueLoad = new LoadHolder(location,
+ initialLoad.getSubscriptionConnectionLoad(),
initialLoad.getLoadPerSubscriptionConnection(), loadPollInterval);
addGroups(queueLoadMap, groups, queueLoad);
- updateLoad(location, initialLoad);
+ updateLoad(location, memberId, initialLoad);
}
/**
* Remove a server from the load snapshot.
*/
- public synchronized void removeServer(ServerLocation location) {
+ public synchronized void removeServer(ServerLocation location, String memberId) {
String[] groups = serverGroupMap.remove(location);
/*
* Adding null check for #41522 - we were getting a remove from a BridgeServer that was shutting
@@ -114,20 +115,21 @@ public class LocatorLoadSnapshot {
* be a race from profile add / remove from different channels.
*/
if (groups != null) {
- removeFromMap(connectionLoadMap, groups, location);
+ removeFromMap(connectionLoadMap, groups, location, memberId);
removeFromMap(queueLoadMap, groups, location);
}
}
- public void updateLoad(ServerLocation location, ServerLoad newLoad) {
- updateLoad(location, newLoad, null);
+ public void updateLoad(ServerLocation location, String memberId, ServerLoad newLoad) {
+ updateLoad(location, memberId, newLoad, null);
}
/**
* Update the load information for a server that was previously added.
*/
- synchronized void updateLoad(ServerLocation location, ServerLoad newLoad,
+ synchronized void updateLoad(ServerLocation location, String memberId, ServerLoad newLoad,
List<ClientProxyMembershipID> clientIds) {
+
String[] groups = serverGroupMap.get(location);
// the server was asynchronously removed, so don't do anything.
if (groups == null) {
@@ -140,7 +142,7 @@ public class LocatorLoadSnapshot {
}
}
- updateMap(connectionLoadMap, location, newLoad.getConnectionLoad(),
+ updateMap(connectionLoadMap, location, memberId, newLoad.getConnectionLoad(),
newLoad.getLoadPerConnection());
updateMap(queueLoadMap, location, newLoad.getSubscriptionConnectionLoad(),
newLoad.getLoadPerSubscriptionConnection());
@@ -151,15 +153,15 @@ public class LocatorLoadSnapshot {
group = null;
}
- Map<ServerLocation, LoadHolder> groupServers = connectionLoadMap.get(group);
+ Map<ServerLocationAndMemberId, LoadHolder> groupServers = connectionLoadMap.get(group);
return isBalanced(groupServers);
}
- private synchronized boolean isBalanced(Map<ServerLocation, LoadHolder> groupServers) {
+ private synchronized boolean isBalanced(Map<ServerLocationAndMemberId, LoadHolder> groupServers) {
return isBalanced(groupServers, false);
}
- private synchronized boolean isBalanced(Map<ServerLocation, LoadHolder> groupServers,
+ private synchronized boolean isBalanced(Map<ServerLocationAndMemberId, LoadHolder> groupServers,
boolean withThresholdCheck) {
if (groupServers == null || groupServers.isEmpty()) {
return true;
@@ -169,7 +171,7 @@ public class LocatorLoadSnapshot {
float largestLoadPerConnection = Float.MIN_VALUE;
float worstLoad = Float.MIN_VALUE;
- for (Entry<ServerLocation, LoadHolder> loadHolderEntry : groupServers.entrySet()) {
+ for (Entry<ServerLocationAndMemberId, LoadHolder> loadHolderEntry : groupServers.entrySet()) {
LoadHolder nextLoadReference = loadHolderEntry.getValue();
float nextLoad = nextLoadReference.getLoad();
float nextLoadPerConnection = nextLoadReference.getLoadPerConnection();
@@ -246,7 +248,7 @@ public class LocatorLoadSnapshot {
group = null;
}
- Map<ServerLocation, LoadHolder> groupServers = connectionLoadMap.get(group);
+ Map<ServerLocationAndMemberId, LoadHolder> groupServers = connectionLoadMap.get(group);
if (groupServers == null || groupServers.isEmpty()) {
return null;
}
@@ -266,12 +268,15 @@ public class LocatorLoadSnapshot {
if ("".equals(group)) {
group = null;
}
- Map<ServerLocation, LoadHolder> groupServers = connectionLoadMap.get(group);
+ Map<ServerLocationAndMemberId, LoadHolder> groupServers = connectionLoadMap.get(group);
if (groupServers == null || groupServers.isEmpty()) {
return null;
}
-
- return new ArrayList<>(groupServers.keySet());
+ ArrayList result = new ArrayList<>();
+ for (ServerLocationAndMemberId locationAndMemberId : groupServers.keySet()) {
+ result.add(locationAndMemberId.getServerLocation());
+ }
+ return result;
}
public void shutDown() {
@@ -279,7 +284,7 @@ public class LocatorLoadSnapshot {
}
/**
- * Pick the least loaded server in the given group if currentServer is the most loaded server. n
+ * Pick the least loaded server in the given group if currentServer is the most loaded server.
*
* @param group the group, or null or "" if the client has no server group.
* @param excludedServers a list of servers to exclude as choices
@@ -292,7 +297,7 @@ public class LocatorLoadSnapshot {
group = null;
}
- Map<ServerLocation, LoadHolder> groupServers = connectionLoadMap.get(group);
+ Map<ServerLocationAndMemberId, LoadHolder> groupServers = connectionLoadMap.get(group);
if (groupServers == null || groupServers.isEmpty()) {
return null;
}
@@ -391,13 +396,13 @@ public class LocatorLoadSnapshot {
* each server.
*/
public synchronized Map<ServerLocation, ServerLoad> getLoadMap() {
- Map<ServerLocation, LoadHolder> connectionMap = connectionLoadMap.get(null);
+ Map<ServerLocationAndMemberId, LoadHolder> connectionMap = connectionLoadMap.get(null);
Map<ServerLocation, LoadHolder> queueMap = queueLoadMap.get(null);
Map<ServerLocation, ServerLoad> result = new HashMap<>();
- for (Entry<ServerLocation, LoadHolder> entry : connectionMap
+ for (Entry<ServerLocationAndMemberId, LoadHolder> entry : connectionMap
.entrySet()) {
- ServerLocation location = entry.getKey();
+ ServerLocation location = entry.getKey().getServerLocation();
LoadHolder connectionLoad = entry.getValue();
LoadHolder queueLoad = queueMap.get(location);
// was asynchronously removed
@@ -412,7 +417,8 @@ public class LocatorLoadSnapshot {
return result;
}
- private void addGroups(Map<String, Map<ServerLocation, LoadHolder>> map, String[] groups,
+ @VisibleForTesting
+ void addGroups(Map<String, Map<ServerLocation, LoadHolder>> map, String[] groups,
LoadHolder holder) {
for (String group : groups) {
Map<ServerLocation, LoadHolder> groupMap = map.computeIfAbsent(group, k -> new HashMap<>());
@@ -425,7 +431,25 @@ public class LocatorLoadSnapshot {
}
}
- private void removeFromMap(Map<String, Map<ServerLocation, LoadHolder>> map, String[] groups,
+ @VisibleForTesting
+ void addGroups(Map<String, Map<ServerLocationAndMemberId, LoadHolder>> map,
+ String[] groups,
+ LoadHolder holder, String memberId) {
+ for (String group : groups) {
+ Map<ServerLocationAndMemberId, LoadHolder> groupMap =
+ map.computeIfAbsent(group, k -> new HashMap<>());
+ groupMap.put(new ServerLocationAndMemberId(holder.getLocation(), memberId), holder);
+ }
+ // Special case for GatewayReceiver where we don't put those serverlocation against holder
+ if (!(groups.length > 0 && groups[0].equals(GatewayReceiver.RECEIVER_GROUP))) {
+ Map<ServerLocationAndMemberId, LoadHolder> groupMap =
+ map.computeIfAbsent(null, k -> new HashMap<>());
+ groupMap.put(new ServerLocationAndMemberId(holder.getLocation(), memberId), holder);
+ }
+ }
+
+ @VisibleForTesting
+ void removeFromMap(Map<String, Map<ServerLocation, LoadHolder>> map, String[] groups,
ServerLocation location) {
for (String group : groups) {
Map<ServerLocation, LoadHolder> groupMap = map.get(group);
@@ -440,9 +464,42 @@ public class LocatorLoadSnapshot {
groupMap.remove(location);
}
- private void updateMap(Map map, ServerLocation location, float load, float loadPerConnection) {
+ @VisibleForTesting
+ void removeFromMap(Map<String, Map<ServerLocationAndMemberId, LoadHolder>> map,
+ String[] groups,
+ ServerLocation location, String memberId) {
+ ServerLocationAndMemberId locationAndMemberId =
+ new ServerLocationAndMemberId(location, memberId);
+ for (String group : groups) {
+ Map<ServerLocationAndMemberId, LoadHolder> groupMap = map.get(group);
+ if (groupMap != null) {
+ groupMap.remove(locationAndMemberId);
+ if (groupMap.size() == 0) {
+ map.remove(group);
+ }
+ }
+ }
+ Map groupMap = map.get(null);
+ groupMap.remove(locationAndMemberId);
+ }
+
+ @VisibleForTesting
+ void updateMap(Map map, ServerLocation location, float load, float loadPerConnection) {
+ updateMap(map, location, "", load, loadPerConnection);
+ }
+
+ @VisibleForTesting
+ void updateMap(Map map, ServerLocation location, String memberId, float load,
+ float loadPerConnection) {
Map groupMap = (Map) map.get(null);
- LoadHolder holder = (LoadHolder) groupMap.get(location);
+ LoadHolder holder;
+ if (memberId.equals("")) {
+ holder = (LoadHolder) groupMap.get(location);
+ } else {
+ ServerLocationAndMemberId locationAndMemberId =
+ new ServerLocationAndMemberId(location, memberId);
+ holder = (LoadHolder) groupMap.get(locationAndMemberId);
+ }
if (holder != null) {
holder.setLoad(load, loadPerConnection);
}
@@ -455,9 +512,15 @@ public class LocatorLoadSnapshot {
* @param count how many you want. a negative number means all of them in order of best to worst
* @return a list of best...worst server LoadHolders
*/
- private List<LoadHolder> findBestServers(Map<ServerLocation, LoadHolder> groupServers,
+ @VisibleForTesting
+ List<LoadHolder> findBestServers(
+ Map<?, LoadHolder> groupServers,
Set<ServerLocation> excludedServers, int count) {
+ if (count == 0) {
+ return new ArrayList<>();
+ }
+
TreeSet<LoadHolder> bestEntries = new TreeSet<>((l1, l2) -> {
int difference = Float.compare(l1.getLoad(), l2.getLoad());
if (difference != 0) {
@@ -471,8 +534,17 @@ public class LocatorLoadSnapshot {
boolean retainAll = (count < 0);
float lastBestLoad = Float.MAX_VALUE;
- for (Map.Entry<ServerLocation, LoadHolder> loadEntry : groupServers.entrySet()) {
- ServerLocation location = loadEntry.getKey();
+ for (Map.Entry<?, LoadHolder> loadEntry : groupServers.entrySet()) {
+ ServerLocation location;
+ Object key = loadEntry.getKey();
+ if (key instanceof ServerLocationAndMemberId) {
+ location = ((ServerLocationAndMemberId) key).getServerLocation();
+ } else if (key instanceof ServerLocation) {
+ location = ((ServerLocation) key);
+ } else {
+ throw new InternalGemFireException(
+ "findBestServers method was called with incorrect type parameters.");
+ }
if (excludedServers.contains(location)) {
continue;
}
@@ -496,14 +568,23 @@ public class LocatorLoadSnapshot {
/**
* If it is most loaded then return its LoadHolder; otherwise return null;
*/
- private LoadHolder isCurrentServerMostLoaded(ServerLocation currentServer,
- Map<ServerLocation, LoadHolder> groupServers) {
- final LoadHolder currentLH = groupServers.get(currentServer);
+ @VisibleForTesting
+ LoadHolder isCurrentServerMostLoaded(ServerLocation currentServer,
+ Map<ServerLocationAndMemberId, LoadHolder> groupServers) {
+
+ // Check if there are keys in the map that contains currentServer.
+ LoadHolder currentLH = null;
+ for (ServerLocationAndMemberId locationAndMemberId : groupServers.keySet()) {
+ if (currentServer.equals(locationAndMemberId.getServerLocation())) {
+ currentLH = groupServers.get(locationAndMemberId);
+ break;
+ }
+ }
if (currentLH == null)
return null;
final float currentLoad = currentLH.getLoad();
- for (Map.Entry<ServerLocation, LoadHolder> loadEntry : groupServers.entrySet()) {
- ServerLocation location = loadEntry.getKey();
+ for (Map.Entry<ServerLocationAndMemberId, LoadHolder> loadEntry : groupServers.entrySet()) {
+ ServerLocation location = loadEntry.getKey().getServerLocation();
if (location.equals(currentServer)) {
continue;
}
@@ -622,7 +703,8 @@ public class LocatorLoadSnapshot {
}
}
- private static class LoadHolder {
+ @VisibleForTesting
+ static class LoadHolder {
private float load;
private float loadPerConnection;
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocationAndMemberId.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocationAndMemberId.java
new file mode 100644
index 0000000..40ef4b6
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocationAndMemberId.java
@@ -0,0 +1,65 @@
+/*
+ * 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.distributed.internal;
+
+public class ServerLocationAndMemberId {
+
+ private final ServerLocation serverLocation;
+ private final String memberId;
+
+ public ServerLocationAndMemberId(ServerLocation serverLocation, String memberId) {
+ this.serverLocation = serverLocation;
+ this.memberId = memberId;
+ }
+
+ public ServerLocation getServerLocation() {
+ return this.serverLocation;
+ }
+
+ public String getMemberId() {
+ return this.memberId;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (!(obj instanceof ServerLocationAndMemberId))
+ return false;
+ final ServerLocationAndMemberId other = (ServerLocationAndMemberId) obj;
+
+ if (!this.serverLocation.equals(other.getServerLocation())) {
+ return false;
+ }
+
+ return this.memberId.equals(other.getMemberId());
+ }
+
+ @Override
+ public String toString() {
+ return serverLocation.toString() + "@" + memberId;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result =
+ prime * result + serverLocation.hashCode() + memberId.hashCode();
+ return result;
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
index 075489f..4442073 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ServerLocator.java
@@ -405,7 +405,9 @@ public class ServerLocator implements TcpHandler, RestartHandler, DistributionAd
CacheServerProfile bp = (CacheServerProfile) profile;
ServerLocation location = buildServerLocation(bp);
String[] groups = bp.getGroups();
- loadSnapshot.addServer(location, groups, bp.getInitialLoad(), bp.getLoadPollInterval());
+ loadSnapshot.addServer(
+ location, bp.getDistributedMember().getUniqueId(), groups,
+ bp.getInitialLoad(), bp.getLoadPollInterval());
if (logger.isDebugEnabled()) {
logger.debug("ServerLocator: Received load from a new server {}, {}", location,
bp.getInitialLoad());
@@ -423,7 +425,7 @@ public class ServerLocator implements TcpHandler, RestartHandler, DistributionAd
CacheServerProfile bp = (CacheServerProfile) profile;
// InternalDistributedMember id = bp.getDistributedMember();
ServerLocation location = buildServerLocation(bp);
- loadSnapshot.removeServer(location);
+ loadSnapshot.removeServer(location, bp.getDistributedMember().getUniqueId());
if (logger.isDebugEnabled()) {
logger.debug("ServerLocator: server departed {}", location);
}
@@ -441,12 +443,14 @@ public class ServerLocator implements TcpHandler, RestartHandler, DistributionAd
.warning("ServerLocator - unexpected profile update.");
}
- public void updateLoad(ServerLocation location, ServerLoad load, List clientIds) {
+ public void updateLoad(ServerLocation location, String memberId, ServerLoad load,
+ List clientIds) {
if (getLogWriter().fineEnabled()) {
getLogWriter()
- .fine("ServerLocator: Received a load update from " + location + ", " + load);
+ .fine("ServerLocator: Received a load update from " + location + " at " + memberId + " , "
+ + load);
}
- loadSnapshot.updateLoad(location, load, clientIds);
+ loadSnapshot.updateLoad(location, memberId, load, clientIds);
this.stats.incServerLoadUpdates();
logServers();
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
index d322d7b..7990323 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
@@ -203,6 +203,7 @@ import org.apache.geode.internal.cache.DistTXPrecommitMessage.DistTxPrecommitRes
import org.apache.geode.internal.cache.DistTXRollbackMessage;
import org.apache.geode.internal.cache.DistributedClearOperation.ClearRegionMessage;
import org.apache.geode.internal.cache.DistributedClearOperation.ClearRegionWithContextMessage;
+import org.apache.geode.internal.cache.DistributedPingMessage;
import org.apache.geode.internal.cache.DistributedPutAllOperation.EntryVersionsList;
import org.apache.geode.internal.cache.DistributedPutAllOperation.PutAllMessage;
import org.apache.geode.internal.cache.DistributedRegionFunctionStreamingMessage;
@@ -978,6 +979,7 @@ public class DSFIDFactory implements DataSerializableFixedID {
GatewaySenderQueueEntrySynchronizationOperation.GatewaySenderQueueEntrySynchronizationEntry.class);
serializer.registerDSFID(ABORT_BACKUP_REQUEST, AbortBackupRequest.class);
serializer.registerDSFID(HOST_AND_PORT, HostAndPort.class);
+ serializer.registerDSFID(DISTRIBUTED_PING_MESSAGE, DistributedPingMessage.class);
}
/**
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPingMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPingMessage.java
new file mode 100644
index 0000000..b601730
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPingMessage.java
@@ -0,0 +1,86 @@
+/*
+ * 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.internal.cache;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.ClusterDistributionManager;
+import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.tier.sockets.ClientHealthMonitor;
+import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.internal.serialization.Version;
+
+public class DistributedPingMessage extends HighPriorityDistributionMessage {
+
+ private ClientProxyMembershipID proxyID;
+
+ public DistributedPingMessage() {
+ // no-arg constructor for serialization
+ }
+
+ public DistributedPingMessage(DistributedMember targetServer, ClientProxyMembershipID proxyID) {
+ super();
+ this.proxyID = proxyID;
+ setRecipient((InternalDistributedMember) targetServer);
+ }
+
+ @Override
+ protected void process(ClusterDistributionManager dm) {
+ ClientHealthMonitor chm = ClientHealthMonitor.getInstance();
+ if (chm != null) {
+ chm.receivedPing(proxyID);
+ }
+ }
+
+ @Override
+ public void toData(DataOutput out, SerializationContext context) throws IOException {
+ super.toData(out, context);
+ context.getSerializer().writeObject(proxyID, out);
+ }
+
+ @Override
+ public void fromData(DataInput in, DeserializationContext context)
+ throws IOException, ClassNotFoundException {
+ super.fromData(in, context);
+ proxyID = context.getDeserializer().readObject(in);
+ }
+
+ @Override
+ public int getDSFID() {
+ return DISTRIBUTED_PING_MESSAGE;
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + "; proxyId=" + proxyID;
+ }
+
+ @VisibleForTesting
+ protected ClientProxyMembershipID getProxyID() {
+ return this.proxyID;
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GridAdvisor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GridAdvisor.java
index 98faad2..cf1566e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GridAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GridAdvisor.java
@@ -418,18 +418,28 @@ public abstract class GridAdvisor extends DistributionAdvisor {
public int hashCode() {
final String thisHost = this.gp.getHost();
final int thisPort = this.gp.getPort();
- return thisHost != null ? (thisHost.hashCode() ^ thisPort) : thisPort;
+ final String thisMemberId = this.getMemberId().getUniqueId();
+ final int thisMemberIdHashCode = (thisMemberId != null) ? thisMemberId.hashCode() : 0;
+ return thisHost != null ? (thisHost.hashCode() ^ thisPort) + thisMemberIdHashCode
+ : thisPort + thisMemberIdHashCode;
}
@Override
public boolean equals(Object obj) {
if (obj instanceof GridProfileId) {
final GridProfileId other = (GridProfileId) obj;
+
if (this.gp.getPort() == other.gp.getPort()) {
final String thisHost = this.gp.getHost();
final String otherHost = other.gp.getHost();
if (thisHost != null) {
- return thisHost.equals(otherHost);
+ if (thisHost.equals(otherHost)) {
+ if (this.getMemberId() != null) {
+ return this.getMemberId().getUniqueId().equals(other.getMemberId().getUniqueId());
+ } else {
+ return other.getMemberId() == null;
+ }
+ }
} else {
return (otherHost == null);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java
index ea90f14..3dfaaef 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/Ping.java
@@ -17,8 +17,11 @@ package org.apache.geode.internal.cache.tier.sockets.command;
import java.io.IOException;
import org.apache.geode.annotations.Immutable;
+import org.apache.geode.cache.client.ServerOperationException;
+import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionStats;
-import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.DistributedPingMessage;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.BaseCommand;
@@ -38,6 +41,10 @@ public class Ping extends BaseCommand {
private Ping() {}
+ protected ClientHealthMonitor getClientHealthMonitor() {
+ return ClientHealthMonitor.getInstance();
+ }
+
@Override
public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection,
final SecurityService securityService, long start) throws IOException {
@@ -47,10 +54,37 @@ public class Ping extends BaseCommand {
clientMessage.getTransactionId(), serverConnection.getSocketString(),
(DistributionStats.getStatTime() - start));
}
- ClientHealthMonitor chm = ClientHealthMonitor.getInstance();
- if (chm != null)
+ if (clientMessage.getNumberOfParts() > 0) {
+ try {
+ InternalDistributedMember targetServer =
+ (InternalDistributedMember) clientMessage.getPart(0).getObject();
+ InternalDistributedMember myID = serverConnection.getCache().getMyId();
+ if (!myID.equals(targetServer)) {
+ if (myID.compareTo(targetServer, true, false) == 0) {
+ String errorMessage =
+ "Target server " + targetServer + " has different viewId: " + myID;
+ logger.warn(errorMessage);
+ writeException(clientMessage, new ServerOperationException(errorMessage), false,
+ serverConnection);
+ serverConnection.setAsTrue(RESPONDED);
+ return;
+ }
+
+ pingCorrectServer(clientMessage, targetServer, serverConnection);
+ serverConnection.setAsTrue(RESPONDED);
+ return;
+ }
+ } catch (ClassNotFoundException e) {
+ logger.warn("Unable to deserialize message from " + serverConnection.getProxyID());
+ writeException(clientMessage, e, false, serverConnection);
+ serverConnection.setAsTrue(RESPONDED);
+ return;
+ }
+ }
+ ClientHealthMonitor chm = getClientHealthMonitor();
+ if (chm != null) {
chm.receivedPing(serverConnection.getProxyID());
- CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+ }
writeReply(clientMessage, serverConnection);
serverConnection.setAsTrue(RESPONDED);
@@ -60,6 +94,32 @@ public class Ping extends BaseCommand {
}
}
+ /**
+ * Process a ping request that was sent to the wrong server
+ */
+ protected void pingCorrectServer(Message clientMessage, DistributedMember targetServer,
+ ServerConnection serverConnection)
+ throws IOException {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Received a Ping request from {} intended for {}. Forwarding the ping...",
+ serverConnection.getProxyID(), targetServer);
+ }
+ if (!serverConnection.getCache().getDistributionManager().isCurrentMember(targetServer)) {
+ String errorMessage = "Unable to ping non-member " + targetServer + " for client "
+ + serverConnection.getProxyID();
+ logger.warn(errorMessage);
+ writeException(clientMessage, new ServerOperationException(errorMessage), false,
+ serverConnection);
+ serverConnection.setAsTrue(RESPONDED);
+ } else {
+ // send a ping message to the server. This is a one-way message that doesn't send a reply
+ final DistributedPingMessage distributedPingMessage =
+ new DistributedPingMessage(targetServer, serverConnection.getProxyID());
+ serverConnection.getCache().getDistributionManager().putOutgoing(distributedPingMessage);
+ writeReply(clientMessage, serverConnection);
+ }
+ }
+
@Override
protected void writeReply(Message origMsg, ServerConnection serverConnection) throws IOException {
Message replyMsg = serverConnection.getReplyMessage();
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotJUnitTest.java
index 5a38b28..8f6ad81 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/LocatorLoadSnapshotJUnitTest.java
@@ -24,12 +24,14 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.server.ServerLoad;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.test.junit.categories.MembershipTest;
/**
@@ -39,12 +41,14 @@ import org.apache.geode.test.junit.categories.MembershipTest;
@Category({MembershipTest.class})
public class LocatorLoadSnapshotJUnitTest {
+ final int LOAD_POLL_INTERVAL = 30000;
+
/**
* Test to make sure than an empty snapshot returns the correct values.
*/
@Test
public void testEmptySnapshot() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
assertNull(sn.getServerForConnection("group", Collections.EMPTY_SET));
assertNull(sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(Collections.EMPTY_LIST, sn.getServersForQueue(null, Collections.EMPTY_SET, 5));
@@ -57,13 +61,15 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testTwoServers() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- ServerLoad ld1 = new ServerLoad(3, 1, 1.01f, 1);
- ServerLoad ld2 = new ServerLoad(5, .2f, 1f, .2f);
- sn.addServer(l1, new String[0], ld1);
- sn.addServer(l2, new String[0], ld2);
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLoad ld1 = new ServerLoad(3, 1, 1.01f, 1);
+ final ServerLoad ld2 = new ServerLoad(5, .2f, 1f, .2f);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[0], ld1, LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[0], ld2, LOAD_POLL_INTERVAL);
HashMap expectedLoad = new HashMap();
expectedLoad.put(l1, ld1);
@@ -100,15 +106,18 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testUpdateLoad() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- sn.addServer(l1, new String[0], new ServerLoad(1, 1, 1, 1));
- sn.addServer(l2, new String[0], new ServerLoad(100, .2f, 1, .2f));
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[0], new ServerLoad(1, 1, 1, 1), LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[0], new ServerLoad(100, .2f, 1, .2f),
+ LOAD_POLL_INTERVAL);
assertEquals(l1, sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(l1, sn.getServerForConnection(null, Collections.EMPTY_SET));
- sn.updateLoad(l1, new ServerLoad(200, 1, 1, 1));
+ sn.updateLoad(l1, uniqueId1, new ServerLoad(200, 1, 1, 1));
assertEquals(l2, sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(l2, sn.getServerForConnection(null, Collections.EMPTY_SET));
}
@@ -119,16 +128,19 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testRemoveServer() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- sn.addServer(l1, new String[0], new ServerLoad(1, 1, 1, 1));
- sn.addServer(l2, new String[0], new ServerLoad(100, .2f, 10, .2f));
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[0], new ServerLoad(1, 1, 1, 1), LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[0], new ServerLoad(100, .2f, 10, .2f),
+ LOAD_POLL_INTERVAL);
assertEquals(l1, sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(Arrays.asList(new ServerLocation[] {l1, l2}),
sn.getServersForQueue(null, Collections.EMPTY_SET, -1));
- sn.removeServer(l1);
+ sn.removeServer(l1, uniqueId1);
assertEquals(l2, sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(Collections.singletonList(l2),
sn.getServersForQueue(null, Collections.EMPTY_SET, -1));
@@ -139,17 +151,21 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testGroups() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- sn.addServer(l1, new String[] {"a", "b"}, new ServerLoad(1, 1, 1, 1));
- sn.addServer(l2, new String[] {"b", "c"}, new ServerLoad(1, 1, 1, 1));
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[] {"a", "b"}, new ServerLoad(1, 1, 1, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[] {"b", "c"}, new ServerLoad(1, 1, 1, 1),
+ LOAD_POLL_INTERVAL);
assertNotNull(sn.getServerForConnection(null, Collections.EMPTY_SET));
assertEquals(l1, sn.getServerForConnection("a", Collections.EMPTY_SET));
assertEquals(l2, sn.getServerForConnection("c", Collections.EMPTY_SET));
- sn.updateLoad(l1, new ServerLoad(10, 1, 1, 1));
+ sn.updateLoad(l1, uniqueId1, new ServerLoad(10, 1, 1, 1));
assertEquals(l2, sn.getServerForConnection("b", Collections.EMPTY_SET));
- sn.updateLoad(l2, new ServerLoad(100, 1, 1, 1));
+ sn.updateLoad(l2, uniqueId2, new ServerLoad(100, 1, 1, 1));
assertEquals(l1, sn.getServerForConnection("b", Collections.EMPTY_SET));
assertEquals(Arrays.asList(new ServerLocation[] {l1}),
sn.getServersForQueue("a", Collections.EMPTY_SET, -1));
@@ -162,7 +178,7 @@ public class LocatorLoadSnapshotJUnitTest {
assertEquals(Arrays.asList(new ServerLocation[] {l1, l2}),
sn.getServersForQueue("b", Collections.EMPTY_SET, 5));
- sn.removeServer(l1);
+ sn.removeServer(l1, uniqueId1);
assertEquals(l2, sn.getServerForConnection("b", Collections.EMPTY_SET));
assertEquals(l2, sn.getServerForConnection("b", Collections.EMPTY_SET));
assertNull(sn.getServerForConnection("a", Collections.EMPTY_SET));
@@ -178,13 +194,18 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testIntersectingGroups() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- ServerLocation l3 = new ServerLocation("localhost", 3);
- sn.addServer(l1, new String[] {"a",}, new ServerLoad(0, 1, 0, 1));
- sn.addServer(l2, new String[] {"a", "b"}, new ServerLoad(0, 1, 0, 1));
- sn.addServer(l3, new String[] {"b"}, new ServerLoad(0, 1, 0, 1));
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[] {"a",}, new ServerLoad(0, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[] {"a", "b"}, new ServerLoad(0, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l3, uniqueId3, new String[] {"b"}, new ServerLoad(0, 1, 0, 1), LOAD_POLL_INTERVAL);
// Test with interleaving requests for either group
for (int i = 0; i < 60; i++) {
@@ -201,9 +222,9 @@ public class LocatorLoadSnapshotJUnitTest {
expected.put(l3, expectedLoad);
assertEquals(expected, sn.getLoadMap());
- sn.updateLoad(l1, new ServerLoad(0, 1, 0, 1));
- sn.updateLoad(l2, new ServerLoad(0, 1, 0, 1));
- sn.updateLoad(l3, new ServerLoad(0, 1, 0, 1));
+ sn.updateLoad(l1, uniqueId1, new ServerLoad(0, 1, 0, 1));
+ sn.updateLoad(l2, uniqueId2, new ServerLoad(0, 1, 0, 1));
+ sn.updateLoad(l3, uniqueId3, new ServerLoad(0, 1, 0, 1));
// Now do the same test, but make all the requests for one group first,
@@ -242,11 +263,13 @@ public class LocatorLoadSnapshotJUnitTest {
*/
@Test
public void testExcludes() {
- LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
- ServerLocation l1 = new ServerLocation("localhost", 1);
- ServerLocation l2 = new ServerLocation("localhost", 2);
- sn.addServer(l1, new String[0], new ServerLoad(1, 1, 1, 1));
- sn.addServer(l2, new String[0], new ServerLoad(100, 1, 100, 1));
+ final LocatorLoadSnapshot sn = new LocatorLoadSnapshot();
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[0], new ServerLoad(1, 1, 1, 1), LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[0], new ServerLoad(100, 1, 100, 1), LOAD_POLL_INTERVAL);
HashSet excludeAll = new HashSet();
excludeAll.add(l1);
@@ -271,21 +294,24 @@ public class LocatorLoadSnapshotJUnitTest {
final ServerLocation l1 = new ServerLocation("localhost", 1);
final ServerLocation l2 = new ServerLocation("localhost", 2);
final ServerLocation l3 = new ServerLocation("localhost", 3);
-
- sn.addServer(l1, new String[] {"a"}, new ServerLoad(0, 1, 0, 1));
- sn.addServer(l2, new String[] {"a", "b"}, new ServerLoad(0, 1, 0, 1));
- sn.addServer(l3, new String[] {"b"}, new ServerLoad(0, 1, 0, 1));
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+ sn.addServer(l1, uniqueId1, new String[] {"a"}, new ServerLoad(0, 1, 0, 1), LOAD_POLL_INTERVAL);
+ sn.addServer(l2, uniqueId2, new String[] {"a", "b"}, new ServerLoad(0, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ sn.addServer(l3, uniqueId3, new String[] {"b"}, new ServerLoad(0, 1, 0, 1), LOAD_POLL_INTERVAL);
assertTrue(sn.hasBalancedConnections(null));
assertTrue(sn.hasBalancedConnections("a"));
assertTrue(sn.hasBalancedConnections("b"));
- sn.updateLoad(l1, new ServerLoad(1, 1, 0, 1));
+ sn.updateLoad(l1, uniqueId1, new ServerLoad(1, 1, 0, 1));
assertTrue(sn.hasBalancedConnections(null));
assertTrue(sn.hasBalancedConnections("a"));
assertTrue(sn.hasBalancedConnections("b"));
- sn.updateLoad(l2, new ServerLoad(2, 1, 0, 1));
+ sn.updateLoad(l2, uniqueId2, new ServerLoad(2, 1, 0, 1));
assertFalse(sn.hasBalancedConnections(null));
assertTrue(sn.hasBalancedConnections("a"));
assertFalse(sn.hasBalancedConnections("b"));
@@ -298,15 +324,22 @@ public class LocatorLoadSnapshotJUnitTest {
final ServerLocation l1 = new ServerLocation("localhost", 1);
final ServerLocation l2 = new ServerLocation("localhost", 2);
final ServerLocation l3 = new ServerLocation("localhost", 3);
-
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
float defaultLoadImbalanceThreshold = LocatorLoadSnapshot.DEFAULT_LOAD_IMBALANCE_THRESHOLD;
float l1ConnectionLoad = 50 + defaultLoadImbalanceThreshold;
float l2ConnectionLoad = 50;
float l3ConnectionLoad = 50 - defaultLoadImbalanceThreshold;
- loadSnapshot.addServer(l1, new String[] {"a"}, new ServerLoad(l1ConnectionLoad, 1, 0, 1));
- loadSnapshot.addServer(l2, new String[] {"a", "b"}, new ServerLoad(l2ConnectionLoad, 1, 0, 1));
- loadSnapshot.addServer(l3, new String[] {"b"}, new ServerLoad(l3ConnectionLoad, 1, 0, 1));
+ loadSnapshot.addServer(l1, uniqueId1, new String[] {"a"},
+ new ServerLoad(l1ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l2, uniqueId2, new String[] {"a", "b"},
+ new ServerLoad(l2ConnectionLoad, 1, 0, 1), LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l3, uniqueId3, new String[] {"b"},
+ new ServerLoad(l3ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
// a new server should be selected until the load-imbalance-threshold is reached
ServerLocation newServer = null;
@@ -340,15 +373,22 @@ public class LocatorLoadSnapshotJUnitTest {
final ServerLocation l1 = new ServerLocation("localhost", 1);
final ServerLocation l2 = new ServerLocation("localhost", 2);
final ServerLocation l3 = new ServerLocation("localhost", 3);
-
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
float defaultLoadImbalanceThreshold = LocatorLoadSnapshot.DEFAULT_LOAD_IMBALANCE_THRESHOLD;
float l1ConnectionLoad = 50 + defaultLoadImbalanceThreshold - 1;
float l2ConnectionLoad = 50;
float l3ConnectionLoad = 50 + (defaultLoadImbalanceThreshold / 2);
- loadSnapshot.addServer(l1, new String[] {"a"}, new ServerLoad(l1ConnectionLoad, 1, 0, 1));
- loadSnapshot.addServer(l2, new String[] {"a", "b"}, new ServerLoad(l2ConnectionLoad, 1, 0, 1));
- loadSnapshot.addServer(l3, new String[] {"b"}, new ServerLoad(l3ConnectionLoad, 1, 0, 1));
+ loadSnapshot.addServer(l1, uniqueId1, new String[] {"a"},
+ new ServerLoad(l1ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l2, uniqueId2, new String[] {"a", "b"},
+ new ServerLoad(l2ConnectionLoad, 1, 0, 1), LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l3, uniqueId3, new String[] {"b"},
+ new ServerLoad(l3ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
ServerLocation newServer =
loadSnapshot.getReplacementServerForConnection(l1, "", Collections.EMPTY_SET);
@@ -357,4 +397,525 @@ public class LocatorLoadSnapshotJUnitTest {
ServerLoad l1Load = loadMap.get(l1);
assertEquals(l1ConnectionLoad, l1Load.getConnectionLoad(), 0.01);
}
+
+ @Test
+ public void testisCurrentServerMostLoaded() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(l1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(l2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(l3, uniqueId3);
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 40;
+ float l3ConnectionLoad = 30;
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(l1, l1ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(l2, l2ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(l3, l3ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ assertEquals(loadHolder1, loadSnapshot.isCurrentServerMostLoaded(l1, groupServers));
+ assertNull(loadSnapshot.isCurrentServerMostLoaded(l2, groupServers));
+ assertNull(loadSnapshot.isCurrentServerMostLoaded(l3, groupServers));
+ }
+
+ @Test
+ public void testGetReplacementServerForConnection() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 40;
+ float l3ConnectionLoad = 30;
+
+ loadSnapshot.addServer(l1, uniqueId1, new String[] {"a"},
+ new ServerLoad(l1ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l2, uniqueId2, new String[] {"a", "b"},
+ new ServerLoad(l2ConnectionLoad, 1, 0, 1), LOAD_POLL_INTERVAL);
+ loadSnapshot.addServer(l3, uniqueId3, new String[] {"b"},
+ new ServerLoad(l3ConnectionLoad, 1, 0, 1),
+ LOAD_POLL_INTERVAL);
+
+ ServerLocation newServer1 =
+ loadSnapshot.getReplacementServerForConnection(l1, "", Collections.EMPTY_SET);
+ assertEquals(l3, newServer1);
+ ServerLocation newServer2 =
+ loadSnapshot.getReplacementServerForConnection(l1, "a", Collections.EMPTY_SET);
+ assertEquals(l2, newServer2);
+ ServerLocation newServer3 =
+ loadSnapshot.getReplacementServerForConnection(l3, "b", Collections.EMPTY_SET);
+ assertEquals(l3, newServer3);
+ ServerLocation newServer4 =
+ loadSnapshot.getReplacementServerForConnection(l2, "b", Collections.EMPTY_SET);
+ assertEquals(l3, newServer4);
+ }
+
+ @Test
+ public void testFindBestServersReturnOneServer() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(l1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(l2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(l3, uniqueId3);
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 30;
+ float l3ConnectionLoad = 40;
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(l1, l1ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(l2, l2ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(l3, l3ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ List<LocatorLoadSnapshot.LoadHolder> result =
+ loadSnapshot.findBestServers(groupServers, Collections.EMPTY_SET, 1);
+ assertEquals(1, result.size());
+ assertEquals(loadHolder2, result.get(0));
+ }
+
+ @Test
+ public void testFindBestServersReturnMoreThanOneServer() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(l1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(l2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(l3, uniqueId3);
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 30;
+ float l3ConnectionLoad = 40;
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(l1, l1ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(l2, l2ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(l3, l3ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ List<LocatorLoadSnapshot.LoadHolder> result =
+ loadSnapshot.findBestServers(groupServers, Collections.EMPTY_SET, 2);
+ assertEquals(2, result.size());
+ assertEquals(loadHolder2, result.get(0));
+ assertEquals(loadHolder3, result.get(1));
+
+ result = loadSnapshot.findBestServers(groupServers, Collections.EMPTY_SET, 0);
+ assertEquals(0, result.size());
+ }
+
+ @Test
+ public void testFindBestServersCalledWithZeroCount() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(l1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(l2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(l3, uniqueId3);
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 30;
+ float l3ConnectionLoad = 40;
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(l1, l1ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(l2, l2ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(l3, l3ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ List<LocatorLoadSnapshot.LoadHolder> result =
+ loadSnapshot.findBestServers(groupServers, Collections.EMPTY_SET, 0);
+ assertEquals(0, result.size());
+ }
+
+ @Test
+ public void testFindBestServersCalledWithNegativeCount() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation l1 = new ServerLocation("localhost", 1);
+ final ServerLocation l2 = new ServerLocation("localhost", 2);
+ final ServerLocation l3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(l1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(l2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(l3, uniqueId3);
+
+ float l1ConnectionLoad = 50;
+ float l2ConnectionLoad = 30;
+ float l3ConnectionLoad = 40;
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(l1, l1ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(l2, l2ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(l3, l3ConnectionLoad, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ List<LocatorLoadSnapshot.LoadHolder> result =
+ loadSnapshot.findBestServers(groupServers, Collections.EMPTY_SET, -1);
+ assertEquals(3, result.size());
+ assertEquals(loadHolder2, result.get(0));
+ assertEquals(loadHolder3, result.get(1));
+ assertEquals(loadHolder1, result.get(2));
+ }
+
+ @Test
+ public void updateMapWithServerLocationAndMemberId() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation serverLocation = new ServerLocation("localhost", 1);
+ final String uniqueId = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final ServerLocationAndMemberId sli = new ServerLocationAndMemberId(serverLocation, uniqueId);
+ LocatorLoadSnapshot.LoadHolder loadHolder =
+ new LocatorLoadSnapshot.LoadHolder(serverLocation, 50, 1, LOAD_POLL_INTERVAL);
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(sli, loadHolder);
+ Map<String, Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.updateMap(map, serverLocation, uniqueId, 60, 2);
+
+ LocatorLoadSnapshot.LoadHolder expectedLoadHolder =
+ new LocatorLoadSnapshot.LoadHolder(serverLocation, 60, 2, LOAD_POLL_INTERVAL);
+
+ assertEquals(expectedLoadHolder.getLoad(), groupServers.get(sli).getLoad(), 0);
+ assertEquals(expectedLoadHolder.getLoadPerConnection(),
+ groupServers.get(sli).getLoadPerConnection(), 0);
+ }
+
+ @Test
+ public void updateMapWithServerLocationAndMemberIdKeyNotFound() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation serverLocation = new ServerLocation("localhost", 1);
+ final String uniqueId = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final ServerLocationAndMemberId sli = new ServerLocationAndMemberId(serverLocation, uniqueId);
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ Map<String, Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.updateMap(map, serverLocation, uniqueId, 50, 1);
+
+ assertNull(groupServers.get(sli));
+ }
+
+ @Test
+ public void updateMapWithServerLocation() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation serverLocation = new ServerLocation("localhost", 1);
+ LocatorLoadSnapshot.LoadHolder loadHolder =
+ new LocatorLoadSnapshot.LoadHolder(serverLocation, 50, 1, LOAD_POLL_INTERVAL);
+ Map<ServerLocation, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(serverLocation, loadHolder);
+ Map<String, Map<ServerLocation, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.updateMap(map, serverLocation, 60, 2);
+
+ LocatorLoadSnapshot.LoadHolder expectedLoadHolder =
+ new LocatorLoadSnapshot.LoadHolder(serverLocation, 60, 2, LOAD_POLL_INTERVAL);
+
+ assertEquals(expectedLoadHolder.getLoad(), groupServers.get(serverLocation).getLoad(), 0);
+ assertEquals(expectedLoadHolder.getLoadPerConnection(),
+ groupServers.get(serverLocation).getLoadPerConnection(), 0);
+ }
+
+ @Test
+ public void updateMapWithServerLocationKeyNotFound() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation serverLocation = new ServerLocation("localhost", 1);
+ Map<ServerLocation, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ Map<String, Map<ServerLocation, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.updateMap(map, serverLocation, 50, 1);
+
+ assertNull(groupServers.get(serverLocation));
+ }
+
+ @Test
+ public void testRemoveFromMapWithServerLocationAndMemberId() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(sl1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(sl2, uniqueId2);
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ Map<String, Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.removeFromMap(map, new String[] {""}, sl1, uniqueId1);
+
+ assertEquals(1, groupServers.size());
+ assertNull(groupServers.get(sli1));
+ }
+
+ @Test
+ public void testRemoveFromMapWithServerLocationAndMemberIdAndGroups() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+ final ServerLocation sl3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(sl1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(sl2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(sl3, uniqueId3);
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(sli1, loadHolder1);
+ groupServers.put(sli2, loadHolder2);
+ groupServers.put(sli3, loadHolder3);
+
+ Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder> groupAServers = new HashMap<>();
+ groupAServers.put(sli1, loadHolder1);
+ groupAServers.put(sli2, loadHolder2);
+
+ Map<String, Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+ map.put("a", groupAServers);
+
+ loadSnapshot.removeFromMap(map, new String[] {"a"}, sl1, uniqueId1);
+
+ assertEquals(2, groupServers.size());
+ assertNull(groupServers.get(sli1));
+
+ assertEquals(1, groupAServers.size());
+ assertNull(groupAServers.get(sli1));
+ }
+
+ @Test
+ public void testRemoveFromMapWithServerLocation() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocation, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(sl1, loadHolder1);
+ groupServers.put(sl2, loadHolder2);
+ Map<String, Map<ServerLocation, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+
+ loadSnapshot.removeFromMap(map, new String[] {""}, sl1);
+
+ assertEquals(1, groupServers.size());
+ assertNull(groupServers.get(sl1));
+ }
+
+ @Test
+ public void testRemoveFromMapWithServerLocationAndGroups() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+ final ServerLocation sl3 = new ServerLocation("localhost", 3);
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(sl3, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<ServerLocation, LocatorLoadSnapshot.LoadHolder> groupServers = new HashMap<>();
+ groupServers.put(sl1, loadHolder1);
+ groupServers.put(sl2, loadHolder2);
+ groupServers.put(sl3, loadHolder3);
+
+ Map<ServerLocation, LocatorLoadSnapshot.LoadHolder> groupAServers = new HashMap<>();
+ groupAServers.put(sl1, loadHolder1);
+ groupAServers.put(sl2, loadHolder2);
+
+ Map<String, Map<ServerLocation, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+ map.put(null, groupServers);
+ map.put("a", groupAServers);
+
+ loadSnapshot.removeFromMap(map, new String[] {"a"}, sl1);
+
+ assertEquals(2, groupServers.size());
+ assertNull(groupServers.get(sl1));
+
+ assertEquals(1, groupAServers.size());
+ assertNull(groupAServers.get(sl1));
+ }
+
+ @Test
+ public void testAddGroupsWithServerLocationAndMemberId() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+ final ServerLocation sl3 = new ServerLocation("localhost", 3);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+ final String uniqueId3 = new InternalDistributedMember("localhost", 3).getUniqueId();
+ final ServerLocationAndMemberId sli1 = new ServerLocationAndMemberId(sl1, uniqueId1);
+ final ServerLocationAndMemberId sli2 = new ServerLocationAndMemberId(sl2, uniqueId2);
+ final ServerLocationAndMemberId sli3 = new ServerLocationAndMemberId(sl3, uniqueId3);
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(sl3, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<String, Map<ServerLocationAndMemberId, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+
+ loadSnapshot.addGroups(map, new String[] {"a"}, loadHolder1, uniqueId1);
+ loadSnapshot.addGroups(map, new String[] {"a", "b"}, loadHolder2, uniqueId2);
+ loadSnapshot.addGroups(map, new String[] {}, loadHolder3, uniqueId3);
+
+ assertEquals(3, map.get(null).size());
+ assertEquals(loadHolder1, map.get(null).get(sli1));
+ assertEquals(loadHolder2, map.get(null).get(sli2));
+ assertEquals(loadHolder3, map.get(null).get(sli3));
+
+ assertEquals(2, map.get("a").size());
+ assertEquals(loadHolder1, map.get("a").get(sli1));
+ assertEquals(loadHolder2, map.get("a").get(sli2));
+
+ assertEquals(1, map.get("b").size());
+ assertEquals(loadHolder2, map.get("b").get(sli2));
+ }
+
+ @Test
+ public void testAddGroupsWithServerLocation() {
+ final LocatorLoadSnapshot loadSnapshot = new LocatorLoadSnapshot();
+
+ final ServerLocation sl1 = new ServerLocation("localhost", 1);
+ final ServerLocation sl2 = new ServerLocation("localhost", 2);
+ final ServerLocation sl3 = new ServerLocation("localhost", 3);
+
+ LocatorLoadSnapshot.LoadHolder loadHolder1 =
+ new LocatorLoadSnapshot.LoadHolder(sl1, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder2 =
+ new LocatorLoadSnapshot.LoadHolder(sl2, 50, 1, LOAD_POLL_INTERVAL);
+ LocatorLoadSnapshot.LoadHolder loadHolder3 =
+ new LocatorLoadSnapshot.LoadHolder(sl3, 50, 1, LOAD_POLL_INTERVAL);
+
+ Map<String, Map<ServerLocation, LocatorLoadSnapshot.LoadHolder>> map =
+ new HashMap<>();
+
+ loadSnapshot.addGroups(map, new String[] {"a"}, loadHolder1);
+ loadSnapshot.addGroups(map, new String[] {"a", "b"}, loadHolder2);
+ loadSnapshot.addGroups(map, new String[] {}, loadHolder3);
+
+ assertEquals(3, map.get(null).size());
+ assertEquals(loadHolder1, map.get(null).get(sl1));
+ assertEquals(loadHolder2, map.get(null).get(sl2));
+ assertEquals(loadHolder3, map.get(null).get(sl3));
+
+ assertEquals(2, map.get("a").size());
+ assertEquals(loadHolder1, map.get("a").get(sl1));
+ assertEquals(loadHolder2, map.get("a").get(sl2));
+
+ assertEquals(1, map.get("b").size());
+ assertEquals(loadHolder2, map.get("b").get(sl2));
+ }
}
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/ServerLocationAndMemberIdTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/ServerLocationAndMemberIdTest.java
new file mode 100644
index 0000000..955874d
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/ServerLocationAndMemberIdTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.distributed.internal;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.junit.Test;
+
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+
+public class ServerLocationAndMemberIdTest {
+
+ @Test
+ public void givenTwoObjectsWithSameHostAndPortAndId_whenCompared_thenAreEquals() {
+ final ServerLocation serverLocation1 = new ServerLocation("localhost", 1);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+
+ ServerLocationAndMemberId serverLocationAndMemberId1 =
+ new ServerLocationAndMemberId(serverLocation1, uniqueId1);
+ ServerLocationAndMemberId serverLocationAndMemberId2 =
+ new ServerLocationAndMemberId(serverLocation1, uniqueId1);
+ assertEquals(serverLocationAndMemberId1, serverLocationAndMemberId2);
+ }
+
+ @Test
+ public void givenTwoObjectsWithSameHostAndPortButDifferentViewId_whenCompared_thenAreNotEquals() {
+
+ final ServerLocation serverLocation1 = new ServerLocation("localhost", 1);
+ InternalDistributedMember idmWithView1 = new InternalDistributedMember("localhost", 1);
+ idmWithView1.setVmViewId(1);
+ InternalDistributedMember idmWithView2 = new InternalDistributedMember("localhost", 1);
+ idmWithView2.setVmViewId(2);
+
+ ServerLocationAndMemberId serverLocationAndMemberId1 =
+ new ServerLocationAndMemberId(serverLocation1, idmWithView1.getUniqueId());
+ ServerLocationAndMemberId serverLocationAndMemberId2 =
+ new ServerLocationAndMemberId(serverLocation1, idmWithView2.getUniqueId());
+
+ assertNotEquals(serverLocationAndMemberId1, serverLocationAndMemberId2);
+ }
+
+ @Test
+ public void givenTwoObjectsWithDifferentHostPortAndId_whenCompared_thenAreNotEquals() {
+ final ServerLocation serverLocation1 = new ServerLocation("localhost", 1);
+ final ServerLocation serverLocation2 = new ServerLocation("localhost", 2);
+ final String uniqueId1 = new InternalDistributedMember("localhost", 1).getUniqueId();
+ final String uniqueId2 = new InternalDistributedMember("localhost", 2).getUniqueId();
+
+ ServerLocationAndMemberId serverLocationAndMemberId1 =
+ new ServerLocationAndMemberId(serverLocation1, uniqueId1);
+ ServerLocationAndMemberId serverLocationAndMemberId2 =
+ new ServerLocationAndMemberId(serverLocation2, uniqueId2);
+
+ assertNotEquals(serverLocationAndMemberId1, serverLocationAndMemberId2);
+ }
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedPingMessageTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedPingMessageTest.java
new file mode 100644
index 0000000..9b4cd08
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedPingMessageTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.internal.cache;
+
+import static org.apache.geode.internal.serialization.DataSerializableFixedID.DISTRIBUTED_PING_MESSAGE;
+import static org.hamcrest.text.MatchesPattern.matchesPattern;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
+
+public class DistributedPingMessageTest {
+
+ @Test
+ public void testToString() {
+ final InternalDistributedMember dummyDistributedMember =
+ new InternalDistributedMember("localhost", 1);
+ final InternalDistributedMember clientDistributedMember =
+ new InternalDistributedMember("localhost", 2);
+ final ClientProxyMembershipID proxyID = new ClientProxyMembershipID(clientDistributedMember);
+ final DistributedPingMessage message =
+ new DistributedPingMessage(dummyDistributedMember, proxyID);
+ assertThat(message.toString(),
+ matchesPattern("DistributedPingMessage@.*; proxyId=identity\\(localhost<.*>:2.*"));
+ }
+
+ @Test
+ public void testGetDSFID() {
+ assertEquals(new DistributedPingMessage().getDSFID(), DISTRIBUTED_PING_MESSAGE);
+ }
+
+ @Test
+ public void testGetSerializationVersions() {
+ assertNull(new DistributedPingMessage().getSerializationVersions());
+ }
+
+ @Test
+ public void testToDataAndFromData() throws IOException, ClassNotFoundException {
+ final InternalDistributedMember dummyDistributedMember =
+ new InternalDistributedMember("localhost", 1);
+ final InternalDistributedMember clientDistributedMember =
+ new InternalDistributedMember("localhost", 2);
+ final ClientProxyMembershipID proxyID = new ClientProxyMembershipID(clientDistributedMember);
+ final DistributedPingMessage before =
+ new DistributedPingMessage(dummyDistributedMember, proxyID);
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(1024);
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ before.toData(dataOutputStream,
+ InternalDataSerializer.createSerializationContext(dataOutputStream));
+ dataOutputStream.close();
+
+ final DistributedPingMessage after = new DistributedPingMessage();
+ ByteArrayInputStream byteArrayInputStream =
+ new ByteArrayInputStream(byteArrayOutputStream.toByteArray());
+ DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream);
+ after.fromData(dataInputStream,
+ InternalDataSerializer.createDeserializationContext(dataInputStream));
+
+ assertEquals(before.getProxyID(), after.getProxyID());
+ }
+
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/PingTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/PingTest.java
new file mode 100644
index 0000000..4b68bd2
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/PingTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.internal.cache.tier.sockets.command;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.distributed.internal.DistributionManager;
+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.sockets.ClientHealthMonitor;
+import org.apache.geode.internal.cache.tier.sockets.Message;
+import org.apache.geode.internal.cache.tier.sockets.Part;
+import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
+
+public class PingTest {
+
+ @Mock
+ private Message message;
+ @Mock
+ private ServerConnection serverConnection;
+ @Mock
+ private Part targetServerPart;
+ @Mock
+ private InternalCache cache;
+ @Mock
+ private DistributionManager distributionManager;
+ @Mock
+ private Message errorResponseMessage;
+ @Mock
+ private ClientHealthMonitor clientHealthMonitor;
+
+ private InternalDistributedMember myID;
+ private InternalDistributedMember targetServer;
+
+ @Mock
+ private Message replyMessage;
+
+ @Spy
+ @InjectMocks
+ private Ping ping;
+
+ @Before
+ public void setUp() throws Exception {
+ this.ping = (Ping) Ping.getCommand();
+ MockitoAnnotations.initMocks(this);
+
+ when(this.ping.getClientHealthMonitor()).thenReturn(clientHealthMonitor);
+ when(this.message.getNumberOfParts()).thenReturn(1);
+ when(this.message.getPart(eq(0))).thenReturn(this.targetServerPart);
+ when(this.serverConnection.getCache()).thenReturn(this.cache);
+ when(this.serverConnection.getReplyMessage()).thenReturn(this.replyMessage);
+ when(this.cache.getCancelCriterion()).thenReturn(mock(CancelCriterion.class));
+ }
+
+ @Test
+ public void testTargetServerAndCurrentServerIDAreEquals() throws Exception {
+
+ myID = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ when(this.cache.getMyId()).thenReturn(this.myID);
+ targetServer = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ when(this.targetServerPart.getObject()).thenReturn(targetServer);
+ this.ping.cmdExecute(this.message, this.serverConnection, null, 0);
+
+ verify(this.replyMessage).send(this.serverConnection);
+ verify(ping, times(1)).getClientHealthMonitor();
+ }
+
+ @Test
+ public void testTargetServerAndCurrentServerIDAreDifferentAndTargetServerIsASystemMember()
+ throws Exception {
+
+ myID = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ when(this.cache.getMyId()).thenReturn(this.myID);
+ targetServer = new InternalDistributedMember(new ServerLocation("localhost", 2));
+ when(this.targetServerPart.getObject()).thenReturn(targetServer);
+ when(this.cache.getDistributionManager()).thenReturn(this.distributionManager);
+ when(this.distributionManager.isCurrentMember(targetServer)).thenReturn(true);
+
+ this.ping.cmdExecute(this.message, this.serverConnection, null, 0);
+
+ verify(this.replyMessage).send(this.serverConnection);
+ verify(ping, times(0)).getClientHealthMonitor();
+ }
+
+ @Test
+ public void testTargetServerAndCurrentServerIDAreDifferentAndTargetServerIsNotASystemMember()
+ throws Exception {
+
+ myID = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ when(this.cache.getMyId()).thenReturn(this.myID);
+ targetServer = new InternalDistributedMember(new ServerLocation("localhost", 2));
+ when(this.targetServerPart.getObject()).thenReturn(targetServer);
+ when(this.cache.getDistributionManager()).thenReturn(this.distributionManager);
+ when(this.distributionManager.isCurrentMember(targetServer)).thenReturn(false);
+ when(this.serverConnection.getErrorResponseMessage()).thenReturn(this.errorResponseMessage);
+
+ this.ping.cmdExecute(this.message, this.serverConnection, null, 0);
+
+ verify(this.errorResponseMessage, times(1)).send(this.serverConnection);
+ verify(this.replyMessage, times(0)).send(this.serverConnection);
+ verify(ping, times(0)).getClientHealthMonitor();
+ }
+
+ @Test
+ public void testTargetServerAndCurrentServerAreEqualsButWithDifferentViewId() throws Exception {
+ myID = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ myID.setVmViewId(1);
+ when(this.cache.getMyId()).thenReturn(this.myID);
+ targetServer = new InternalDistributedMember(new ServerLocation("localhost", 1));
+ targetServer.setVmViewId(2);
+ when(this.targetServerPart.getObject()).thenReturn(targetServer);
+ when(this.serverConnection.getErrorResponseMessage()).thenReturn(this.errorResponseMessage);
+
+ this.ping.cmdExecute(this.message, this.serverConnection, null, 0);
+
+ verify(this.errorResponseMessage, times(1)).send(this.serverConnection);
+ verify(this.replyMessage, times(0)).send(this.serverConnection);
+ verify(ping, times(0)).getClientHealthMonitor();
+ }
+}
diff --git a/geode-dunit/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientServerMiscDUnitTestBase.java b/geode-dunit/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientServerMiscDUnitTestBase.java
index af73fc7..41599a8 100755
--- a/geode-dunit/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientServerMiscDUnitTestBase.java
+++ b/geode-dunit/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientServerMiscDUnitTestBase.java
@@ -63,6 +63,7 @@ import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.Connection;
import org.apache.geode.cache.client.internal.Op;
+import org.apache.geode.cache.client.internal.PingOp;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.QueueConnectionImpl;
import org.apache.geode.cache.client.internal.RegisterInterestTracker;
@@ -72,6 +73,7 @@ import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
+import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.AvailablePort;
import org.apache.geode.internal.cache.CachePerfStats;
@@ -518,6 +520,23 @@ public class ClientServerMiscDUnitTestBase extends JUnit4CacheTestCase {
}
+ @Test
+ public void testPingWrongServer() throws Exception {
+ PORT1 = initServerCache(true);
+ initServerCache2();
+ InternalDistributedMember server2ID = server2.invoke("get ID", () -> cache.getMyId());
+ pool = (PoolImpl) createClientCache(NetworkUtils.getServerHostName(), PORT1);
+ // send the ping to server1 but use server2's identifier so the ping will be forwarded
+ PingOp.execute(pool, new ServerLocation(NetworkUtils.getServerHostName(), PORT1), server2ID);
+ ClientProxyMembershipID proxyID = server1.invoke(
+ () -> CacheClientNotifier.getInstance().getClientProxies().iterator().next().getProxyID());
+ // if the ping made it to server2 it will have the client's ID in its health monitor
+ server2.invoke(() -> {
+ assertThat(ClientHealthMonitor.getInstance().getClientHeartbeats().keySet().contains(proxyID))
+ .isTrue();
+ });
+ }
+
/**
* Test two regions: notify by subscription is true. Both the regions have registered interest in
* all the keys. Destroy region1 on the client. It should reach the server , kill the region on
diff --git a/geode-serialization/src/main/java/org/apache/geode/internal/serialization/DataSerializableFixedID.java b/geode-serialization/src/main/java/org/apache/geode/internal/serialization/DataSerializableFixedID.java
index 4ea6608..7a2cd57 100644
--- a/geode-serialization/src/main/java/org/apache/geode/internal/serialization/DataSerializableFixedID.java
+++ b/geode-serialization/src/main/java/org/apache/geode/internal/serialization/DataSerializableFixedID.java
@@ -57,6 +57,8 @@ public interface DataSerializableFixedID extends SerializationVersions, BasicSer
// NOTE, codes < -65536 will take 4 bytes to serialize
// NOTE, codes < -128 will take 2 bytes to serialize
+ short DISTRIBUTED_PING_MESSAGE = -162;
+
short REGION_REDUNDANCY_STATUS = -161;
short RESTORE_REDUNDANCY_RESULTS = -160;