GEODE-9985: add region redundancy to avoid data loss. (apache#7308)

hulishui · Feb 7, 2022 · e7094ee · e7094ee
1 parent 1da9039
commit e7094ee
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 5 deletions.
diff --git a/...ibutedTest/java/org/apache/geode/internal/cache/tier/sockets/SlowDispatcherDUnitTest.java b/...ibutedTest/java/org/apache/geode/internal/cache/tier/sockets/SlowDispatcherDUnitTest.java
@@ -29,14 +29,17 @@
 
 import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.InterestResultPolicy;
+import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.cache.client.ClientCache;
 import org.apache.geode.cache.client.ClientRegionShortcut;
 import org.apache.geode.cache.client.internal.Endpoint;
 import org.apache.geode.cache.client.internal.PoolImpl;
 import org.apache.geode.cache.util.CacheListenerAdapter;
+import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.security.UpdatableUserAuthInitialize;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.rules.ClientVM;
@@ -63,14 +66,21 @@ public void setup() {
     locator = cluster.startLocatorVM(0);
     int locatorPort = locator.getPort();
     SerializableFunction<ServerStarterRule> serverStartRule =
-        s -> s.withRegion(RegionShortcut.PARTITION, PARTITION_REGION)
-            .withSystemProperty("slowStartTimeForTesting", "10000")
+        s -> s.withSystemProperty("slowStartTimeForTesting", "10000")
             .withConnectionToLocator(locatorPort);
     server1 = cluster.startServerVM(1, serverStartRule);
     server2 = cluster.startServerVM(2, serverStartRule);
 
     MemberVM.invokeInEveryMember(() -> {
       CacheClientProxy.isSlowStartForTesting = true;
+      InternalCache cache = ClusterStartupRule.getCache();
+      RegionFactory<Object, Object> regionFactory =
+          cache.createRegionFactory(RegionShortcut.PARTITION);
+
+      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
+      pfact.setRedundantCopies(1);
+      regionFactory.setPartitionAttributes(pfact.create());
+      regionFactory.create(PARTITION_REGION);
     }, server1, server2);
   }
 
@@ -85,7 +95,7 @@ public void registeredInterest_durableClient_kill_primarySever_receives_marker()
     ClientVM clientVM = cluster.startClientVM(3,
         c -> c.withProperty(DURABLE_CLIENT_ID, "123456")
             .withCacheSetup(
-                a -> a.setPoolSubscriptionRedundancy(2).setPoolSubscriptionEnabled(true)
+                a -> a.setPoolSubscriptionRedundancy(1).setPoolSubscriptionEnabled(true)
                     .setPoolMinConnections(2))
             .withServerConnection(server1Port, server2Port));
 

diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionFactoryImpl.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionFactoryImpl.java
@@ -26,6 +26,7 @@
 import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.annotations.internal.MutableForTesting;
 import org.apache.geode.cache.GatewayConfigurationException;
+import org.apache.geode.cache.client.ServerConnectivityException;
 import org.apache.geode.cache.client.ServerRefusedConnectionException;
 import org.apache.geode.cache.client.internal.ServerDenyList.FailureTracker;
 import org.apache.geode.cache.wan.GatewaySender;
@@ -129,8 +130,9 @@ public Connection createClientToServerConnection(ServerLocation location, boolea
       testFailedConnectionToServer = true;
       throw src;
     } catch (Exception e) {
-      if (e.getMessage() != null && (e.getMessage().equals("Connection refused")
-          || e.getMessage().equals("Connection reset"))) {
+      String message = e.getMessage();
+      if (message != null && (message.contains("Connection refused")
+          || message.contains("Connection reset"))) {
         // this is the most common case, so don't print an exception
         if (logger.isDebugEnabled()) {
           logger.debug("Unable to connect to {}: connection refused", location);
@@ -164,6 +166,11 @@ void authenticateIfRequired(Connection conn) {
         return;
       }
       Long uniqueID = AuthenticateUserOp.executeOn(conn, pool);
+      // if connection failed, this would return null, instead of throwing a NPE, we should throw
+      // this
+      if (uniqueID == null) {
+        throw new ServerConnectivityException("Connection refused");
+      }
       server.setUserId(uniqueID);
     }