GEODE-8333: Change Redis adapter threading model - fixes pubsub issues (apache#5488)

- Do not use Netty threads for the entire request lifecycle. Each
  instance of ExecutionHandlerContext (essentially each client
  connection) uses a command queue which is actioned by a single thread
  taken from the Geode 'waiting pool'.
- Every SUBSCRIBEed client is moved to a separate EventGroupLoop so that
  PUBLISHed messages are not sent back using the 'normal' Worker
  EventLoopGroup. This avoids a deadlock issue where PUBLISHed messages
  need to be sent using the same thread that the PUBLISH response needs
  to happen on.
- Fix issues with PubSub where switching the EventLoopGroup may fail
  (because a client has already closed the connection) thus resulting in
  hanging PUBLISHers waiting for a Subscription to be marked as
  'readyForPublish'.
- Consolidate various MockSubscriber classes

Author: jdeppe-pivotal

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

@@ -547,6 +547,12 @@
         <version>3.2.0</version>
         <scope>compile</scope>
       </dependency>
+      <dependency>
+        <groupId>io.lettuce</groupId>
+        <artifactId>lettuce-core</artifactId>
+        <version>5.2.1.RELEASE</version>
+        <scope>compile</scope>
+      </dependency>
       <dependency>
         <groupId>xerces</groupId>
         <artifactId>xercesImpl</artifactId>

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

@@ -169,6 +169,7 @@ class DependencyConstraints implements Plugin<Project> {
         api(group: 'org.testcontainers', name: 'testcontainers', version: '1.13.0')
         api(group: 'pl.pragmatists', name: 'JUnitParams', version: '1.1.0')
         api(group: 'redis.clients', name: 'jedis', version: '3.2.0')
+        api(group: 'io.lettuce', name: 'lettuce-core', version: '5.2.1.RELEASE')
         api(group: 'xerces', name: 'xercesImpl', version: '2.12.0')
       }
     }

geode-redis/build.gradle

@@ -45,11 +45,14 @@ dependencies {
 
   commonTestImplementation(project(':geode-junit'))
   commonTestImplementation(project(':geode-dunit'))
+  commonTestImplementation('redis.clients:jedis')
 
   integrationTestImplementation(project(':geode-dunit'))
   integrationTestImplementation(project(':geode-junit'))
   integrationTestImplementation(sourceSets.commonTest.output)
   integrationTestImplementation('redis.clients:jedis')
+  integrationTestImplementation('io.lettuce:lettuce-core')
+  integrationTestImplementation('org.apache.logging.log4j:log4j-core')
   integrationTestRuntime(project(':geode-log4j'))
 
   acceptanceTestImplementation(sourceSets.integrationTest.output)
@@ -65,6 +68,7 @@ dependencies {
   }
   acceptanceTestImplementation('org.springframework.boot:spring-boot-starter-data-redis')
   acceptanceTestImplementation('org.springframework.session:spring-session-data-redis')
+  acceptanceTestImplementation('org.apache.logging.log4j:log4j-core')
 
   distributedTestCompile('org.apache.logging.log4j:log4j-core')
   distributedTestImplementation(project(':geode-dunit'))

geode-redis/src/acceptanceTest/java/org/apache/geode/redis/internal/executor/pubsub/PubSubNativeRedisAcceptanceTest.java

@@ -42,8 +42,8 @@ public class PubSubNativeRedisAcceptanceTest extends PubSubIntegrationTest {
   public static void setUp() {
     redisContainer = new GenericContainer<>("redis:5.0.6").withExposedPorts(6379);
     redisContainer.start();
-    subscriber = new Jedis("localhost", redisContainer.getFirstMappedPort(), REDIS_CLIENT_TIMEOUT);
-    publisher = new Jedis("localhost", redisContainer.getFirstMappedPort(), REDIS_CLIENT_TIMEOUT);
+    subscriber = new Jedis("localhost", redisContainer.getFirstMappedPort(), JEDIS_TIMEOUT);
+    publisher = new Jedis("localhost", redisContainer.getFirstMappedPort(), JEDIS_TIMEOUT);
   }
 
   @AfterClass
@@ -55,4 +55,5 @@ public static void tearDown() {
   public int getPort() {
     return redisContainer.getFirstMappedPort();
   }
+
 }

geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockBinarySubscriber.java geode-redis/src/commonTest/java/org/apache/geode/redis/mocks/MockBinarySubscriber.java

@@ -11,7 +11,6 @@
  * 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.redis.mocks;
@@ -20,33 +19,112 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
+import redis.clients.jedis.Client;
 import redis.clients.jedis.JedisPubSub;
 
 public class MockSubscriber extends JedisPubSub {
+
+  private final CountDownLatch subscriptionLatch;
+  private final CountDownLatch unsubscriptionLatch;
   private final List<String> receivedMessages = Collections.synchronizedList(new ArrayList<>());
   private final List<String> receivedPMessages = Collections.synchronizedList(new ArrayList<>());
+  public final List<UnsubscribeInfo> unsubscribeInfos =
+      Collections.synchronizedList(new ArrayList<>());
+  public final List<UnsubscribeInfo> punsubscribeInfos =
+      Collections.synchronizedList(new ArrayList<>());
+  private String localSocketAddress;
+
+  public MockSubscriber() {
+    this(new CountDownLatch(1));
+  }
+
+  public MockSubscriber(CountDownLatch subscriptionLatch) {
+    this(subscriptionLatch, new CountDownLatch(1));
+  }
+
+  public MockSubscriber(CountDownLatch subscriptionLatch, CountDownLatch unsubscriptionLatch) {
+    this.subscriptionLatch = subscriptionLatch;
+    this.unsubscriptionLatch = unsubscriptionLatch;
+  }
+
+  @Override
+  public void proceed(Client client, String... channels) {
+    localSocketAddress = client.getSocket().getLocalSocketAddress().toString();
+    super.proceed(client, channels);
+  }
+
+  private void switchThreadName(String suffix) {
+    String threadName = Thread.currentThread().getName();
+    int suffixIndex = threadName.indexOf(" -- ");
+    if (suffixIndex >= 0) {
+      threadName = threadName.substring(0, suffixIndex);
+    }
+
+    threadName += " -- " + suffix + " [" + localSocketAddress + "]";
+    Thread.currentThread().setName(threadName);
+  }
 
   public List<String> getReceivedMessages() {
-    return new ArrayList<>(receivedMessages);
+    return receivedMessages;
   }
 
   public List<String> getReceivedPMessages() {
     return new ArrayList<>(receivedPMessages);
   }
 
-  public final List<UnsubscribeInfo> unsubscribeInfos =
-      Collections.synchronizedList(new ArrayList<>());
-  public final List<UnsubscribeInfo> punsubscribeInfos =
-      Collections.synchronizedList(new ArrayList<>());
+  @Override
+  public void onMessage(String channel, String message) {
+    switchThreadName(String.format("MESSAGE %s %s", channel, message));
+    receivedMessages.add(message);
+  }
+
+  @Override
+  public void onPMessage(String pattern, String channel, String message) {
+    switchThreadName(String.format("PMESSAGE %s %s %s", pattern, channel, message));
+    receivedPMessages.add(message);
+  }
+
+  @Override
+  public void onSubscribe(String channel, int subscribedChannels) {
+    switchThreadName(String.format("SUBSCRIBE %s", channel));
+    subscriptionLatch.countDown();
+  }
+
+  private static final int AWAIT_TIMEOUT_MILLIS = 30000;
+
+  public void awaitSubscribe(String channel) {
+    try {
+      if (!subscriptionLatch.await(AWAIT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
+        throw new RuntimeException("awaitSubscribe timed out for channel: " + channel);
+      }
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
 
   @Override
   public void onUnsubscribe(String channel, int subscribedChannels) {
+    switchThreadName(String.format("UNSUBSCRIBE %s %d", channel, subscribedChannels));
     unsubscribeInfos.add(new UnsubscribeInfo(channel, subscribedChannels));
+    unsubscriptionLatch.countDown();
+  }
+
+  public void awaitUnsubscribe(String channel) {
+    try {
+      if (!unsubscriptionLatch.await(AWAIT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
+        throw new RuntimeException("awaitUnsubscribe timed out for channel: " + channel);
+      }
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
   }
 
   @Override
   public void onPUnsubscribe(String pattern, int subscribedChannels) {
+    switchThreadName(String.format("PUNSUBSCRIBE %s %d", pattern, subscribedChannels));
     punsubscribeInfos.add(new UnsubscribeInfo(pattern, subscribedChannels));
   }
 
@@ -86,14 +164,4 @@ public String toString() {
     }
   }
 
-
-  @Override
-  public void onMessage(String channel, String message) {
-    receivedMessages.add(message);
-  }
-
-  @Override
-  public void onPMessage(String pattern, String channel, String message) {
-    receivedPMessages.add(message);
-  }
 }