GEODE-9277: Redis FLUSHALL should only remove local keys (apache#6481)

- This changes the `FLUSHALL` command from a distributed command to a command
  which only removes keys on the node to which the client is connected. This is
  consistent with redis cluster.
- In DUnit tests, now use `RedisClusterStartupRule.flushAll()` to remove all
  keys in the cluster.
- This commit also disables 4 session-related tests due to issues with retrying
  session commands.

boms/geode-all-bom/src/test/resources/expected-pom.xml

@@ -187,6 +187,16 @@
         <artifactId>classgraph</artifactId>
         <version>4.8.104</version>
       </dependency>
+      <dependency>
+        <groupId>io.github.resilience4j</groupId>
+        <artifactId>resilience4j-retry</artifactId>
+        <version>1.7.0</version>
+      </dependency>
+      <dependency>
+        <groupId>io.lettuce</groupId>
+        <artifactId>lettuce-core</artifactId>
+        <version>6.1.1.RELEASE</version>
+      </dependency>
       <dependency>
         <groupId>io.micrometer</groupId>
         <artifactId>micrometer-core</artifactId>
@@ -452,11 +462,6 @@
         <artifactId>jedis</artifactId>
         <version>3.5.2</version>
       </dependency>
-      <dependency>
-        <groupId>io.lettuce</groupId>
-        <artifactId>lettuce-core</artifactId>
-        <version>6.1.1.RELEASE</version>
-      </dependency>
       <dependency>
         <groupId>xerces</groupId>
         <artifactId>xercesImpl</artifactId>

buildSrc/src/main/groovy/org/apache/geode/gradle/plugins/DependencyConstraints.groovy

@@ -112,6 +112,8 @@ class DependencyConstraints implements Plugin<Project> {
         api(group: 'commons-validator', name: 'commons-validator', version: get('commons-validator.version'))
         // Careful when upgrading this dependency: see GEODE-7370 and GEODE-8150.
         api(group: 'io.github.classgraph', name: 'classgraph', version: '4.8.104')
+        api(group: 'io.github.resilience4j', name: 'resilience4j-retry', version: '1.7.0')
+        api(group: 'io.lettuce', name: 'lettuce-core', version: '6.1.1.RELEASE')
         api(group: 'io.micrometer', name: 'micrometer-core', version: get('micrometer.version'))
         api(group: 'io.netty', name: 'netty-all', version: '4.1.59.Final')
         api(group: 'io.swagger', name: 'swagger-annotations', version: '1.6.2')
@@ -165,7 +167,6 @@ class DependencyConstraints implements Plugin<Project> {
         api(group: 'org.testcontainers', name: 'testcontainers', version: '1.15.3')
         api(group: 'pl.pragmatists', name: 'JUnitParams', version: '1.1.0')
         api(group: 'redis.clients', name: 'jedis', version: '3.5.2')
-        api(group: 'io.lettuce', name: 'lettuce-core', version: '6.1.1.RELEASE')
         api(group: 'xerces', name: 'xercesImpl', version: '2.12.0')
       }
     }

geode-apis-compatible-with-redis/build.gradle

@@ -81,12 +81,14 @@ dependencies {
   acceptanceTestImplementation('org.apache.logging.log4j:log4j-core')
   // This only exists for debugging PubSubNativeRedisAcceptanceTest
   acceptanceTestImplementation('org.buildobjects:jproc:2.6.0')
+  acceptanceTestImplementation('io.github.resilience4j:resilience4j-retry')
 
   distributedTestImplementation('org.apache.logging.log4j:log4j-core')
   distributedTestImplementation(project(':geode-dunit'))
   distributedTestImplementation(sourceSets.commonTest.output)
   distributedTestImplementation('redis.clients:jedis')
   distributedTestImplementation('io.lettuce:lettuce-core')
+  distributedTestImplementation('io.github.resilience4j:resilience4j-retry')
 
   distributedTestImplementation('javax.servlet:javax.servlet-api')
   distributedTestImplementation('org.springframework.boot:spring-boot-starter-jetty')

geode-apis-compatible-with-redis/src/acceptanceTest/java/session/NativeRedisSessionAcceptanceTest.java

@@ -16,11 +16,13 @@
 
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.apache.geode.redis.NativeRedisClusterTestRule;
 import org.apache.geode.redis.session.RedisSessionDUnitTest;
 
+@Ignore("GEODE-9341")
 public class NativeRedisSessionAcceptanceTest extends RedisSessionDUnitTest {
 
   @ClassRule
@@ -30,16 +32,21 @@ public class NativeRedisSessionAcceptanceTest extends RedisSessionDUnitTest {
   public static void setup() {
     setupAppPorts();
     setupNativeRedis();
-    setupClient();
     startSpringApp(APP1, DEFAULT_SESSION_TIMEOUT, ports.get(SERVER1));
     startSpringApp(APP2, DEFAULT_SESSION_TIMEOUT, ports.get(SERVER1));
+    setupRetry();
   }
 
   protected static void setupNativeRedis() {
     ports.put(SERVER1, redis.getExposedPorts().get(0));
     ports.put(SERVER2, redis.getExposedPorts().get(0));
   }
 
+  @Override
+  protected void flushAll() {
+    redis.flushAll();
+  }
+
   @Test
   public void should_getSessionFromServer1_whenServer2GoesDown() {
     // Cannot crash docker-based redis cluster instance (yet).

geode-apis-compatible-with-redis/src/acceptanceTest/java/session/NativeRedisSessionExpirationAcceptanceTest.java

@@ -16,11 +16,13 @@
 
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.apache.geode.redis.NativeRedisClusterTestRule;
 import org.apache.geode.redis.session.SessionExpirationDUnitTest;
 
+@Ignore("GEODE-9341")
 public class NativeRedisSessionExpirationAcceptanceTest extends SessionExpirationDUnitTest {
   @ClassRule
   public static NativeRedisClusterTestRule redis = new NativeRedisClusterTestRule();
@@ -29,9 +31,14 @@ public class NativeRedisSessionExpirationAcceptanceTest extends SessionExpiratio
   public static void setup() {
     setupAppPorts();
     setupNativeRedis();
-    setupClient();
     startSpringApp(APP1, SHORT_SESSION_TIMEOUT, ports.get(SERVER1));
     startSpringApp(APP2, SHORT_SESSION_TIMEOUT, ports.get(SERVER1));
+    setupRetry();
+  }
+
+  @Override
+  protected void flushAll() {
+    redis.flushAll();
   }
 
   protected static void setupNativeRedis() {

geode-apis-compatible-with-redis/src/commonTest/java/org/apache/geode/redis/RedisIntegrationTest.java

@@ -26,8 +26,18 @@ public interface RedisIntegrationTest {
   int REDIS_CLIENT_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
 
   default void flushAll() {
+    ClusterNodes nodes;
     try (Jedis jedis = new Jedis("localhost", getPort(), REDIS_CLIENT_TIMEOUT)) {
-      jedis.flushAll();
+      nodes = ClusterNodes.parseClusterNodes(jedis.clusterNodes());
+    }
+
+    for (ClusterNode node : nodes.getNodes()) {
+      if (!node.primary) {
+        continue;
+      }
+      try (Jedis jedis = new Jedis(node.ipAddress, (int) node.port, REDIS_CLIENT_TIMEOUT)) {
+        jedis.flushAll();
+      }
     }
   }
 }

geode-apis-compatible-with-redis/src/commonTest/java/org/apache/geode/test/dunit/rules/RedisClusterStartupRule.java

@@ -22,14 +22,21 @@
 
 import java.util.Properties;
 
+import redis.clients.jedis.Jedis;
+
+import org.apache.geode.redis.ClusterNode;
+import org.apache.geode.redis.ClusterNodes;
 import org.apache.geode.redis.internal.GeodeRedisServer;
 import org.apache.geode.redis.internal.GeodeRedisService;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.junit.rules.ServerStarterRule;
 
 public class RedisClusterStartupRule extends ClusterStartupRule {
 
-  private static final String BIND_ADDRESS = "127.0.0.1";
-  public static final String DEFAULT_MAX_WAIT_TIME_RECONNECT = "15000";
+  public static final int REDIS_CLIENT_TIMEOUT =
+      Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
+
+  public static final String BIND_ADDRESS = "127.0.0.1";
 
   public RedisClusterStartupRule() {
     super();
@@ -102,4 +109,24 @@ public Long getDataStoreBytesInUseForDataRegion(MemberVM vm) {
       return service.getDataStoreBytesInUseForDataRegion();
     });
   }
+
+  public void flushAll() {
+    flushAll(getRedisPort(1));
+  }
+
+  private void flushAll(int redisPort) {
+    ClusterNodes nodes;
+    try (Jedis jedis = new Jedis("localhost", redisPort, REDIS_CLIENT_TIMEOUT)) {
+      nodes = ClusterNodes.parseClusterNodes(jedis.clusterNodes());
+    }
+
+    for (ClusterNode node : nodes.getNodes()) {
+      if (!node.primary) {
+        continue;
+      }
+      try (Jedis jedis = new Jedis(node.ipAddress, (int) node.port, REDIS_CLIENT_TIMEOUT)) {
+        jedis.flushAll();
+      }
+    }
+  }
 }

geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java

@@ -22,7 +22,6 @@
 import static org.assertj.core.api.AssertionsForClassTypes.catchThrowable;
 
 import java.util.Arrays;
-import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.AfterClass;
@@ -72,9 +71,8 @@ public static void classSetup() {
 
     MemberVM locator = clusterStartUp.startLocatorVM(0);
 
-    Properties serverProperties = new Properties();
-    server1 = clusterStartUp.startRedisVM(1, serverProperties, locator.getPort());
-    server2 = clusterStartUp.startRedisVM(2, serverProperties, locator.getPort());
+    server1 = clusterStartUp.startRedisVM(1, locator.getPort());
+    server2 = clusterStartUp.startRedisVM(2, locator.getPort());
 
     server1.getVM().invoke(() -> RedisClusterStartupRule.getCache().getResourceManager()
         .setCriticalHeapPercentage(5.0F));
@@ -90,7 +88,7 @@ public static void classSetup() {
 
   @Before
   public void testSetup() {
-    jedis1.flushAll();
+    clusterStartUp.flushAll();
   }
 
   @AfterClass

geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/RedisDistDUnitTest.java

@@ -101,8 +101,7 @@ public void run() {
 
   @After
   public void cleanup() {
-    Jedis jedis = new Jedis(LOCALHOST, server1Port, JEDIS_TIMEOUT);
-    jedis.flushAll();
+    cluster.flushAll();
   }
 
   @Test

geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/internal/cluster/RedisPartitionResolverDUnitTest.java

@@ -53,19 +53,21 @@ public class RedisPartitionResolverDUnitTest {
   private static MemberVM server2;
   private static MemberVM server3;
 
+
   @BeforeClass
   public static void classSetup() {
     MemberVM locator = cluster.startLocatorVM(0);
     server1 = cluster.startRedisVM(1, locator.getPort());
     server2 = cluster.startRedisVM(2, locator.getPort());
     server3 = cluster.startRedisVM(3, locator.getPort());
 
-    jedis1 = new Jedis(LOCAL_HOST, cluster.getRedisPort(1), JEDIS_TIMEOUT);
+    int redisServerPort1 = cluster.getRedisPort(1);
+    jedis1 = new Jedis(LOCAL_HOST, redisServerPort1, JEDIS_TIMEOUT);
   }
 
   @Before
   public void testSetup() {
-    jedis1.flushAll();
+    cluster.flushAll();
   }
 
   @Test

geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/internal/data/DeltaDUnitTest.java

@@ -15,15 +15,12 @@
 
 package org.apache.geode.redis.internal.data;
 
-import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
-import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.DEFAULT_MAX_WAIT_TIME_RECONNECT;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.Random;
 
 import org.junit.AfterClass;
@@ -55,27 +52,20 @@ public class DeltaDUnitTest {
   private static Jedis jedis1;
   private static Jedis jedis2;
 
-  private static Properties locatorProperties;
-
   private static MemberVM locator;
   private static MemberVM server1;
   private static MemberVM server2;
 
-  private static int redisServerPort1;
-  private static int redisServerPort2;
   private static Random random;
 
   @BeforeClass
   public static void classSetup() {
-    locatorProperties = new Properties();
-    locatorProperties.setProperty(MAX_WAIT_TIME_RECONNECT, DEFAULT_MAX_WAIT_TIME_RECONNECT);
-
-    locator = clusterStartUp.startLocatorVM(0, locatorProperties);
+    locator = clusterStartUp.startLocatorVM(0);
     server1 = clusterStartUp.startRedisVM(1, locator.getPort());
     server2 = clusterStartUp.startRedisVM(2, locator.getPort());
 
-    redisServerPort1 = clusterStartUp.getRedisPort(1);
-    redisServerPort2 = clusterStartUp.getRedisPort(2);
+    int redisServerPort1 = clusterStartUp.getRedisPort(1);
+    int redisServerPort2 = clusterStartUp.getRedisPort(2);
 
     jedis1 = new Jedis(LOCAL_HOST, redisServerPort1, JEDIS_TIMEOUT);
     jedis2 = new Jedis(LOCAL_HOST, redisServerPort2, JEDIS_TIMEOUT);
@@ -84,7 +74,7 @@ public static void classSetup() {
 
   @Before
   public void testSetup() {
-    jedis1.flushAll();
+    clusterStartUp.flushAll();
   }
 
   @AfterClass

geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/internal/data/PartitionedRegionStatsUpdateTest.java

@@ -17,12 +17,8 @@
 
 package org.apache.geode.redis.internal.data;
 
-import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
-import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.DEFAULT_MAX_WAIT_TIME_RECONNECT;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Properties;
-
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -47,18 +43,15 @@ public class PartitionedRegionStatsUpdateTest {
 
   private static final int JEDIS_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
   private static final String LOCAL_HOST = "127.0.0.1";
-  public static final String STRING_KEY = "string key";
-  public static final String SET_KEY = "set key";
-  public static final String HASH_KEY = "hash key";
-  public static final String LONG_APPEND_VALUE = String.valueOf(Integer.MAX_VALUE);
-  public static final String FIELD = "field";
+  private static final String STRING_KEY = "string key";
+  private static final String SET_KEY = "set key";
+  private static final String HASH_KEY = "hash key";
+  private static final String LONG_APPEND_VALUE = String.valueOf(Integer.MAX_VALUE);
+  private static final String FIELD = "field";
 
   @BeforeClass
   public static void classSetup() {
-    Properties locatorProperties = new Properties();
-    locatorProperties.setProperty(MAX_WAIT_TIME_RECONNECT, DEFAULT_MAX_WAIT_TIME_RECONNECT);
-
-    MemberVM locator = clusterStartUpRule.startLocatorVM(0, locatorProperties);
+    MemberVM locator = clusterStartUpRule.startLocatorVM(0);
     int locatorPort = locator.getPort();
 
     server1 = clusterStartUpRule.startRedisVM(1, locatorPort);
@@ -72,7 +65,7 @@ public static void classSetup() {
 
   @Before
   public void setup() {
-    jedis1.flushAll();
+    clusterStartUpRule.flushAll();
   }
 
   @Test
@@ -133,7 +126,7 @@ public void should_resetMemoryUsage_givenFlushAllCommand() {
 
     jedis1.set(STRING_KEY, "value");
 
-    jedis1.flushAll();
+    clusterStartUpRule.flushAll();
 
     long finalDataStoreBytesInUse = clusterStartUpRule.getDataStoreBytesInUseForDataRegion(server1);