You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ig...@apache.org on 2020/01/08 06:02:57 UTC
[geode] branch develop updated: GEODE-7210: Fix
RedundancyLevelPart1DUnitTest
This is an automated email from the ASF dual-hosted git repository.
igodwin 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 bc9860e GEODE-7210: Fix RedundancyLevelPart1DUnitTest
bc9860e is described below
commit bc9860eacf74d2c11134b5338cea2ad35e5200f0
Author: Ivan Godwin <ig...@pivotal.io>
AuthorDate: Tue Jan 7 22:02:26 2020 -0800
GEODE-7210: Fix RedundancyLevelPart1DUnitTest
Changes made to address flakiness:
- Inline super
- Implement Serializable
- Use Dunit rules DistributedRule and CacheRule
- Use AssertJ
- Use setup and teardown methods
- CountdownLatch rather than Thread.sleep
- No static methods
- Delete ignored tests
---
.../sockets/RedundancyLevelPart1DUnitTest.java | 554 ++++++++++++++-------
.../sockets/RedundancyLevelPart3DUnitTest.java | 2 -
2 files changed, 364 insertions(+), 192 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart1DUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart1DUnitTest.java
index 4ca1f4e..83c7a7c 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart1DUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart1DUnitTest.java
@@ -14,33 +14,138 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.apache.geode.internal.lang.SystemPropertyHelper.GEMFIRE_PREFIX;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.Disconnect.disconnectAllFromDS;
+import static org.apache.geode.test.dunit.VM.getController;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.apache.geode.test.dunit.VM.toArray;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.apache.geode.cache.InterestResultPolicy;
+import org.apache.geode.cache.NoSubscriptionServersAvailableException;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.client.PoolManager;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.client.internal.RegisterInterestTracker;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.internal.ServerLocation;
+import org.apache.geode.internal.cache.ClientServerObserver;
+import org.apache.geode.internal.cache.ClientServerObserverAdapter;
+import org.apache.geode.internal.cache.ClientServerObserverHolder;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalCacheServer;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.NetworkUtils;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
+import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;
/**
* Tests Redundancy Level Functionality
*/
@Category({ClientSubscriptionTest.class})
-public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
-
- @BeforeClass
- public static void caseSetUp() throws Exception {
- disconnectAllFromDS();
+public class RedundancyLevelPart1DUnitTest implements Serializable {
+
+ private static final long TIMEOUT_MILLIS = GeodeAwaitility.getTimeout().getValueInMS();
+ private static final String K1 = "k1";
+ private static final String K2 = "k2";
+ private static final String REGION_NAME = "RedundancyLevelTestBase_region";
+ private static final int CONNECTED_SERVERS = 3;
+ private static final int DEFAULT_SOCKET_READ_TIMEOUT = 3000;
+ private static final int DEFAULT_RETRY_INTERVAL = 10;
+ private static final int DEFAULT_CONNECTED_SERVER_COUNT = 4;
+
+ private static String server0;
+ private static String server1;
+ private static String server2;
+ private static String server3;
+ private static PoolImpl pool = null;
+ private static AtomicBoolean failOverDetectionByCCU = new AtomicBoolean(false);
+ private static ClientServerObserver clientServerObserver = null;
+ private static InternalCache cache;
+ private static String hostname;
+
+ private final transient CountDownLatch latch = new CountDownLatch(1);
+
+ private int port0;
+ private int port1;
+ private int port2;
+ private int port3;
+ private VM vm0;
+ private VM vm1;
+ private VM vm2;
+ private VM vm3;
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule();
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public DistributedRestoreSystemProperties restoreSystemProperties =
+ new DistributedRestoreSystemProperties();
+
+ @Before
+ public void setUp() {
+ hostname = NetworkUtils.getServerHostName();
+
+ vm0 = getVM(0);
+ vm1 = getVM(1);
+ vm2 = getVM(2);
+ vm3 = getVM(3);
+
+ IgnoredException.addIgnoredException("java.net.SocketException||java.net.ConnectException");
+
+ port0 = vm0.invoke(this::createServerCache);
+ port1 = vm1.invoke(this::createServerCache);
+ port2 = vm2.invoke(this::createServerCache);
+ port3 = vm3.invoke(this::createServerCache);
+
+ server0 = hostname + port0;
+ server1 = hostname + port1;
+ server2 = hostname + port2;
+ server3 = hostname + port3;
+
+ System.setProperty(GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints", "true");
}
- private void waitConnectedServers(final int expected) {
- await("Connected server count (" + pool.getConnectedServerCount() + ") never became "
- + expected)
- .until(() -> pool.getConnectedServerCount(), equalTo(expected));
+ @After
+ public void tearDown() {
+ latch.countDown();
+ if (!failOverDetectionByCCU.get()) {
+ ClientServerObserverHolder.setInstance(clientServerObserver);
+ }
+
+ for (VM vm : toArray(getController(), vm0, vm1, vm2, vm3)) {
+ vm.invoke(() -> cacheRule.closeAndNullCache());
+ }
+
+ disconnectAllFromDS();
}
/**
@@ -48,18 +153,14 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* well as the live server map
*/
@Test
- public void testRedundancyNotSpecifiedNonPrimaryServerFail() throws Exception {
+ public void testRedundancyNotSpecifiedNonPrimaryServerFail() {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 0);
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
- verifyConnectedAndRedundantServers(3, 0);
- verifyOrderOfEndpoints();
-
- await("pool still contains " + SERVER3)
- .until(() -> !pool.getCurrentServerNames().contains(SERVER3));
+ createClientCache(
+ 0, DEFAULT_SOCKET_READ_TIMEOUT, DEFAULT_RETRY_INTERVAL);
+ vm2.invoke(this::stopServer);
+ verifyConnectedAndRedundantServers(0);
+ await().untilAsserted(() -> assertThat(pool.getCurrentServerNames()).doesNotContain(server2));
}
/**
@@ -67,25 +168,20 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* removed from the live server map, added to dead server map.
*/
@Test
- public void testRedundancyNotSpecifiedPrimaryServerFails() throws Exception {
+ public void testRedundancyNotSpecifiedPrimaryServerFails() {
// Asif: Increased the socket read timeout to 3000 sec because the registering
// of keys was timing out sometimes causing fail over to EP4 causing
// below assertion to fail
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 0, 3000, 100);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- verifyOrderOfEndpoints();
- server0.invoke(RedundancyLevelTestBase::stopServer);
- verifyConnectedAndRedundantServers(3, 0);
- verifyOrderOfEndpoints();
-
- await("pool still contains " + SERVER1)
- .until(() -> !pool.getCurrentServerNames().contains(SERVER1));
+ createClientCache(0, 3000, 100);
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ vm0.invoke(this::stopServer);
+ verifyConnectedAndRedundantServers(0);
- assertThat(pool.getPrimaryName()).isNotEqualTo(SERVER1);
- assertThat(pool.getPrimaryName()).isEqualTo(SERVER2);
+ await().untilAsserted(() -> assertThat(pool.getCurrentServerNames()).doesNotContain(server0));
+ assertThat(pool.getPrimaryName()).isNotEqualTo(server0);
+ assertThat(pool.getPrimaryName()).isEqualTo(server1);
}
/**
@@ -94,83 +190,15 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* change the current failover set. Failover detection by LSM
*/
@Test
- public void testRedundancySpecifiedNonFailoverEPFails() throws Exception {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1);
- waitConnectedServers(4);
- assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
- verifyRedundantServersContain(SERVER2);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- }
-
- /**
- * Redundancy level specified & less than total Eps. If an EP dies & was not part of the fail over
- * list , then it should be removed from Live Server Map & added to dead server map. It should not
- * change the current failover set. Failover detection by CCU
- */
- @Ignore("TODO")
- @Test
- public void testRedundancySpecifiedNonFailoverEPFailsDetectionByCCU() throws Exception {
- FailOverDetectionByCCU = true;
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
- assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(pool.getRedundantNames().contains(SERVER4)).isTrue();
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
- verifyRedundantServersContain(SERVER4);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- }
-
- /**
- * Redundancy level specified & less than total Eps. If an EP dies & was not part of the fail over
- * list , then it should be removed from Live Server Map & added to dead server map. It should not
- * change the current failover set. Failover detection by Register Interest
- */
- @Ignore("TODO")
- @Test
- public void testRedundancySpecifiedNonFailoverEPFailsDetectionByRegisterInterest()
- throws Exception {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
- assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(pool.getRedundantNames().contains(SERVER4)).isTrue();
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
- createEntriesK1andK2();
- registerK1AndK2();
- verifyRedundantServersContain(SERVER4);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- }
-
- /**
- * Redundancy level specified & less than total Eps. If an EP dies & was not part of the fail over
- * list , then it should be removed from Live Server Map & added to dead server map. It should not
- * change the current failover set. Failover detection by Unregister Interest
- */
- @Ignore("TODO")
- @Test
- public void testRedundancySpecifiedNonFailoverEPFailsDetectionByUnregisterInterest()
- throws Exception {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
+ public void testRedundancySpecifiedNonFailoverEPFails() {
+ createClientCache(
+ 1, DEFAULT_SOCKET_READ_TIMEOUT, DEFAULT_RETRY_INTERVAL);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(pool.getRedundantNames().contains(SERVER4)).isTrue();
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
- unregisterInterest();
- verifyRedundantServersContain(SERVER4);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm2.invoke(this::stopServer);
+ verifyRedundantServersContain(server1);
+ verifyConnectedAndRedundantServers(1);
}
/**
@@ -179,18 +207,15 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* change the current failover set. Failover detection by Put operation.
*/
@Test
- public void testRedundancySpecifiedNonFailoverEPFailsDetectionByPut() throws Exception {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 500, 1000);
- waitConnectedServers(4);
+ public void testRedundancySpecifiedNonFailoverEPFailsDetectionByPut() {
+ createClientCache(1, 500, 1000);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::stopServer);
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm2.invoke(this::stopServer);
doPuts();
- verifyRedundantServersContain(SERVER2);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
+ verifyRedundantServersContain(server1);
+ verifyConnectedAndRedundantServers(1);
}
/**
@@ -199,20 +224,17 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* picked from the Live Server Map to compensate for the failure. Failure Detection by LSM.
*/
@Test
- public void testRedundancySpecifiedNonPrimaryEPFails() throws Exception {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1);
- waitConnectedServers(4);
+ public void testRedundancySpecifiedNonPrimaryEPFails() {
+ createClientCache(
+ 1, DEFAULT_SOCKET_READ_TIMEOUT, DEFAULT_RETRY_INTERVAL);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server1.invoke(RedundancyLevelTestBase::stopServer);
- verifyRedundantServersContain(SERVER3);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::verifyInterestRegistration);
-
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm1.invoke(this::stopServer);
+ verifyRedundantServersContain(server2);
+ verifyConnectedAndRedundantServers(1);
+ vm2.invoke(this::verifyInterestRegistration);
}
/**
@@ -221,21 +243,17 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* picked from the Live Server Map to compensate for the failure. Failure Detection by CCU.
*/
@Test
- public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByCCU() throws Exception {
- FailOverDetectionByCCU = true;
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
+ public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByCCU() {
+ failOverDetectionByCCU.set(true);
+ createClientCache(1, 250, 500);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server1.invoke(RedundancyLevelTestBase::stopServer);
- verifyRedundantServersContain(SERVER3);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::verifyInterestRegistration);
-
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm1.invoke(this::stopServer);
+ verifyRedundantServersContain(server2);
+ verifyConnectedAndRedundantServers(1);
+ vm2.invoke(this::verifyInterestRegistration);
}
/**
@@ -245,24 +263,19 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* Interest.
*/
@Test
- public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByRegisterInterest()
- throws Exception {
+ public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByRegisterInterest() {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
+ createClientCache(1, 250, 500);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server1.invoke(RedundancyLevelTestBase::stopServer);
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm1.invoke(this::stopServer);
createEntriesK1andK2();
registerK1AndK2();
- verifyRedundantServersContain(SERVER3);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::verifyInterestRegistration);
-
+ verifyRedundantServersContain(server2);
+ verifyConnectedAndRedundantServers(1);
+ vm2.invoke(this::verifyInterestRegistration);
}
/**
@@ -272,21 +285,17 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* Interest.
*/
@Test
- public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByUnregisterInterest()
- throws Exception {
+ public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByUnregisterInterest() {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
+ createClientCache(1, 250, 500);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server1.invoke(RedundancyLevelTestBase::stopServer);
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm1.invoke(this::stopServer);
unregisterInterest();
- verifyRedundantServersContain(SERVER3);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
+ verifyRedundantServersContain(server2);
+ verifyConnectedAndRedundantServers(1);
}
/**
@@ -296,25 +305,190 @@ public class RedundancyLevelPart1DUnitTest extends RedundancyLevelTestBase {
* operation.
*/
@Test
- public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByPut() throws Exception {
+ public void testRedundancySpecifiedNonPrimaryEPFailsDetectionByPut() {
- createClientCache(NetworkUtils.getServerHostName(), PORT1, PORT2, PORT3, PORT4,
- 1, 250, 500);
- waitConnectedServers(4);
+ createClientCache(1, 250, 500);
+ waitConnectedServers();
assertThat(pool.getRedundantNames().size()).isEqualTo(1);
- assertThat(SERVER1).isEqualTo(pool.getPrimaryName());
- assertThat(pool.getRedundantNames().contains(SERVER2)).isTrue();
- verifyOrderOfEndpoints();
- server1.invoke(RedundancyLevelTestBase::stopServer);
+ assertThat(server0).isEqualTo(pool.getPrimaryName());
+ assertThat(pool.getRedundantNames().contains(server1)).isTrue();
+ vm1.invoke(this::stopServer);
doPuts();
- System.out.println("server1=" + SERVER1);
- System.out.println("server2=" + SERVER2);
- System.out.println("server3=" + SERVER3);
- System.out.println("server4=" + SERVER4);
- verifyRedundantServersContain(SERVER3);
- verifyConnectedAndRedundantServers(3, 1);
- verifyOrderOfEndpoints();
- server2.invoke(RedundancyLevelTestBase::verifyInterestRegistration);
+ System.out.println("server1=" + server0);
+ System.out.println("server2=" + server1);
+ System.out.println("server3=" + server2);
+ System.out.println("server4=" + server3);
+ verifyRedundantServersContain(server2);
+ verifyConnectedAndRedundantServers(1);
+ vm2.invoke(this::verifyInterestRegistration);
+
+ }
+
+ private void doPuts() {
+ Region<String, String> region = cache.getRegion(REGION_NAME);
+ assertThat(region).isNotNull();
+ region.put(K1, K1);
+ region.put(K2, K2);
+ assertThat(region.get(K1)).isEqualTo(K1);
+ assertThat(region.get(K2)).isEqualTo(K2);
+ }
+
+ private void verifyRedundantServersContain(final String server) {
+ await().untilAsserted(() -> assertThat(pool.getRedundantNames()).contains(server));
+ }
+
+ private void verifyConnectedAndRedundantServers(final int redundantServers) {
+ if (redundantServers < 0) {
+ throw new IllegalArgumentException("can't test for < 0 redundant server via API");
+ }
+ await(
+ "Live server count didn't match expected and/or redundant server count didn't match expected in time")
+ .until(() -> {
+ try {
+ return pool.getConnectedServerCount() == CONNECTED_SERVERS
+ && pool.getRedundantNames().size() == redundantServers;
+ } catch (final NoSubscriptionServersAvailableException e) {
+ // when zero connected servers are actually available, we'll see this error
+ }
+ return false;
+ });
+ }
+
+ private void createEntriesK1andK2() {
+ Region<String, String> r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME);
+ assertThat(r1).isNotNull();
+ if (!r1.containsKey(K1)) {
+ r1.create(K1, K1);
+ }
+ if (!r1.containsKey(K2)) {
+ r1.create(K2, K2);
+ }
+ assertThat(r1.getEntry(K1).getValue()).isEqualTo(K1);
+ assertThat(r1.getEntry(K2).getValue()).isEqualTo(K2);
+ }
+
+ private void registerK1AndK2() {
+ Region<Object, Object> region = cache.getRegion(REGION_NAME);
+ assertThat(region).isNotNull();
+ List<String> list = new ArrayList<>();
+ list.add(K1);
+ list.add(K2);
+ region.registerInterest(list, InterestResultPolicy.KEYS_VALUES);
+ }
+
+ private static void unregisterInterest() {
+ Region<String, String> r = cache.getRegion(Region.SEPARATOR + REGION_NAME);
+ r.unregisterInterest("k1");
+ }
+
+ private void stopServer() {
+ Iterator<CacheServer> iterator = cache.getCacheServers().iterator();
+ if (iterator.hasNext()) {
+ CacheServer server = iterator.next();
+ server.stop();
+ }
+ }
+
+ private void createClientCache(int redundancy, int socketReadTimeout, int retryInterval) {
+ if (!failOverDetectionByCCU.get()) {
+ clientServerObserver =
+ ClientServerObserverHolder.setInstance(new ClientServerObserverAdapter() {
+ @Override
+ public void beforeFailoverByCacheClientUpdater(ServerLocation epFailed) {
+ try {
+ latch.await(TIMEOUT_MILLIS, MILLISECONDS);
+ } catch (InterruptedException ie) {
+ // expected - test will shut down the cache which will interrupt
+ // the CacheClientUpdater thread that invoked this method
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ }
+
+ Properties props = new Properties();
+ props.setProperty(MCAST_PORT, "0");
+ props.setProperty(LOCATORS, "");
+
+ cache = cacheRule.getOrCreateCache(props);
+
+ pool = (PoolImpl) PoolManager.createFactory()
+ .addServer(hostname, port0)
+ .addServer(hostname, port1)
+ .addServer(hostname, port2)
+ .addServer(hostname, port3)
+ .setSubscriptionEnabled(true)
+ .setReadTimeout(socketReadTimeout)
+ .setSocketBufferSize(32768)
+ .setMinConnections(8)
+ .setSubscriptionRedundancy(redundancy)
+ .setRetryAttempts(5)
+ .setPingInterval(retryInterval)
+ .create("DurableClientReconnectDUnitTestPool");
+
+ cache.createRegionFactory()
+ .setScope(Scope.DISTRIBUTED_ACK)
+ .setPoolName(pool.getName())
+ .create(REGION_NAME);
+
+ createEntriesK1andK2();
+ registerK1AndK2();
+ }
+
+ private int createServerCache() throws Exception {
+ cache = cacheRule.getOrCreateCache();
+
+ cache.createRegionFactory(RegionShortcut.REPLICATE).setEnableSubscriptionConflation(true)
+ .create(REGION_NAME);
+
+ CacheServer cacheServer = cache.addCacheServer();
+
+ cacheServer.setMaximumTimeBetweenPings(180000);
+ cacheServer.setPort(0);
+ cacheServer.start();
+
+ return cacheServer.getPort();
+ }
+
+ private void verifyInterestRegistration() {
+ await().untilAsserted(() -> assertThat(cache.getCacheServers()).hasSize(1));
+
+ InternalCacheServer cacheServer =
+ (InternalCacheServer) cache.getCacheServers().iterator().next();
+ assertThat(cacheServer).isNotNull();
+ assertThat(cacheServer.getAcceptor()).isNotNull();
+ assertThat(cacheServer.getAcceptor().getCacheClientNotifier()).isNotNull();
+
+ CacheClientNotifier cacheClientNotifier =
+ cacheServer.getAcceptor().getCacheClientNotifier();
+ await().untilAsserted(
+ () -> assertThat(cacheClientNotifier.getClientProxies().size()).isGreaterThan(0));
+
+ Iterator<CacheClientProxy> cacheClientProxyIterator =
+ cacheClientNotifier.getClientProxies().iterator();
+
+ assertThat(cacheClientProxyIterator.hasNext()).describedAs("A CacheClientProxy was expected")
+ .isTrue();
+ CacheClientProxy cacheClientProxy = cacheClientProxyIterator.next();
+
+ await().until(() -> {
+ Set<?> keysMap = cacheClientProxy.cils[RegisterInterestTracker.interestListIndex]
+ .getProfile(REGION_NAME).getKeysOfInterestFor(cacheClientProxy.getProxyID());
+ if (keysMap == null) {
+ return false;
+ }
+ return 2 == keysMap.size();
+ });
+
+ Set<?> keysMap = cacheClientProxy.cils[RegisterInterestTracker.interestListIndex]
+ .getProfile(REGION_NAME).getKeysOfInterestFor(cacheClientProxy.getProxyID());
+
+ assertThat(keysMap.contains(K1)).isTrue();
+ assertThat(keysMap.contains(K2)).isTrue();
+ }
+ private void waitConnectedServers() {
+ await().untilAsserted(
+ () -> assertThat(pool.getConnectedServerCount()).isEqualTo(DEFAULT_CONNECTED_SERVER_COUNT));
}
}
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart3DUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart3DUnitTest.java
index 472ca31..9ae2f9d 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart3DUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/RedundancyLevelPart3DUnitTest.java
@@ -342,12 +342,10 @@ public class RedundancyLevelPart3DUnitTest implements Serializable {
private void doPuts() {
Region<String, String> region = cache.getRegion(REGION_NAME);
assertThat(region).isNotNull();
- // for (int i = 0; i < 4; i++) {
region.put(K1, K1);
region.put(K2, K2);
assertThat(region.get(K1)).isEqualTo(K1);
assertThat(region.get(K2)).isEqualTo(K2);
- // }
}
private void verifyDispatcherIsAlive() {