[geode] branch develop updated: GEODE-7870: Fix PubSubIntegrationTest flakiness (#4946)

[je...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jensdeppe 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 d431f0d  GEODE-7870: Fix PubSubIntegrationTest flakiness (#4946)
d431f0d is described below

commit d431f0da3c66c49f90895710a0ce93ffc91a7e27
Author: Jens Deppe <jd...@pivotal.io>
AuthorDate: Tue Apr 14 05:34:32 2020 -0700

    GEODE-7870: Fix PubSubIntegrationTest flakiness (#4946)
    
    The complete fix also builds on the following prior fixes
    
    - GEODE-7943 add synchronization to Subscriptions class
    - GEODE-7946: Fix redis publish/subscribe leaking netty buffers
---
 .../java/org/apache/geode/redis/PubSubIntegrationTest.java    | 11 ++++++-----
 .../java/org/apache/geode/redis/mocks/MockSubscriber.java     |  9 +++++----
 2 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java b/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
index 1b70076..f50d74b 100644
--- a/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
+++ b/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
@@ -27,7 +27,6 @@ import java.util.concurrent.Callable;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import redis.clients.jedis.Jedis;
@@ -41,7 +40,6 @@ import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.junit.categories.RedisTest;
 
 @Category({RedisTest.class})
-@Ignore("GEODE-7870")
 public class PubSubIntegrationTest {
   static Jedis publisher;
   static Jedis subscriber;
@@ -53,7 +51,7 @@ public class PubSubIntegrationTest {
   @BeforeClass
   public static void setUp() {
     CacheFactory cf = new CacheFactory();
-    cf.set(LOG_LEVEL, "warn");
+    cf.set(LOG_LEVEL, "info");
     cf.set(MCAST_PORT, "0");
     cf.set(LOCATORS, "");
     cache = cf.create();
@@ -253,11 +251,14 @@ public class PubSubIntegrationTest {
 
     waitFor(() -> mockSubscriber.getSubscribedChannels() == 1);
 
-    Long result = publisher.publish("salutations", "hello");
+    String message = "hello-" + System.currentTimeMillis();
+
+    Long result = publisher.publish("salutations", message);
     assertThat(result).isEqualTo(1);
 
     assertThat(mockSubscriber.getReceivedMessages()).isEmpty();
-    assertThat(mockSubscriber.getReceivedPMessages()).containsExactly("hello");
+    GeodeAwaitility.await().until(() -> !mockSubscriber.getReceivedPMessages().isEmpty());
+    assertThat(mockSubscriber.getReceivedPMessages()).containsExactly(message);
 
     mockSubscriber.punsubscribe("sal*s");
     waitFor(() -> mockSubscriber.getSubscribedChannels() == 0);
diff --git a/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriber.java b/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriber.java
index b0d5326..5ba976e 100644
--- a/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriber.java
+++ b/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriber.java
@@ -17,20 +17,21 @@
 package org.apache.geode.redis.mocks;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 import redis.clients.jedis.JedisPubSub;
 
 public class MockSubscriber extends JedisPubSub {
-  private List<String> receivedMessages = new ArrayList<>();
-  private List<String> receivedPMessages = new ArrayList<>();
+  private final List<String> receivedMessages = Collections.synchronizedList(new ArrayList<>());
+  private final List<String> receivedPMessages = Collections.synchronizedList(new ArrayList<>());
 
   public List<String> getReceivedMessages() {
-    return receivedMessages;
+    return new ArrayList<>(receivedMessages);
   }
 
   public List<String> getReceivedPMessages() {
-    return receivedPMessages;
+    return new ArrayList<>(receivedPMessages);
   }
 
   @Override