Put message persisters behind feature flags.

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -332,8 +332,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     TurnTokenGenerator       turnTokenGenerator = new TurnTokenGenerator(config.getTurnConfiguration());
     RecaptchaClient          recaptchaClient    = new RecaptchaClient(config.getRecaptchaConfiguration().getSecret());
 
-    MessagePersister             messagePersister        = new MessagePersister(messagesClient, messagesManager, pubSubManager, pushSender, accountsManager,config.getMessageCacheConfiguration().getPersistDelayMinutes(), TimeUnit.MINUTES);
+    MessagePersister             messagePersister        = new MessagePersister(messagesClient, messagesManager, pubSubManager, pushSender, accountsManager, featureFlagsManager, config.getMessageCacheConfiguration().getPersistDelayMinutes(), TimeUnit.MINUTES);
-    RedisClusterMessagePersister clusterMessagePersister = new RedisClusterMessagePersister(clusterMessagesCache, messages, pubSubManager, pushSender, accountsManager, Duration.ofMinutes(config.getMessageCacheConfiguration().getPersistDelayMinutes()));
+    RedisClusterMessagePersister clusterMessagePersister = new RedisClusterMessagePersister(clusterMessagesCache, messages, pubSubManager, pushSender, accountsManager, featureFlagsManager, Duration.ofMinutes(config.getMessageCacheConfiguration().getPersistDelayMinutes()));
 
     DirectoryReconciliationClient directoryReconciliationClient = new DirectoryReconciliationClient(config.getDirectoryConfiguration().getDirectoryServerConfiguration());
 

service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java

@@ -53,16 +53,20 @@ public class MessagePersister implements Managed, Runnable {
   private final PubSubManager       pubSubManager;
   private final PushSender          pushSender;
   private final AccountsManager     accountsManager;
+  private final FeatureFlagsManager featureFlagsManager;
 
   private final LuaScript getQueuesScript;
 
   private boolean finished = false;
 
+  private static final String DISABLE_PERSISTENCE_FLAG = "disable-singleton-persister";
+
   public MessagePersister(final ReplicatedJedisPool jedisPool,
                           final MessagesManager messagesManager,
                           final PubSubManager pubSubManager,
                           final PushSender pushSender,
                           final AccountsManager accountsManager,
+                          final FeatureFlagsManager featureFlagsManager,
                           final long delayTime,
                           final TimeUnit delayTimeUnit)
       throws IOException
@@ -73,6 +77,7 @@ public class MessagePersister implements Managed, Runnable {
     this.pubSubManager       = pubSubManager;
     this.pushSender          = pushSender;
     this.accountsManager     = accountsManager;
+    this.featureFlagsManager = featureFlagsManager;
 
     this.delayTime       = delayTime;
     this.delayTimeUnit   = delayTimeUnit;
@@ -87,6 +92,7 @@ public class MessagePersister implements Managed, Runnable {
   @Override
   public void run() {
     while (running.get()) {
+      if (!featureFlagsManager.isFeatureFlagActive(DISABLE_PERSISTENCE_FLAG)) {
         try {
           List<byte[]> queuesToPersist = getQueuesToPersist();
           queueCountHistogram.update(queuesToPersist.size());
@@ -105,6 +111,12 @@ public class MessagePersister implements Managed, Runnable {
         } catch (Throwable t) {
           logger.error("Exception while persisting: ", t);
         }
+      } else {
+        try {
+          Thread.sleep(10_000);
+        } catch (final InterruptedException ignored) {
+        }
+      }
     }
 
     synchronized (this) {

service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagePersister.java

@@ -30,6 +30,7 @@ public class RedisClusterMessagePersister implements Managed {
     private final PubSubManager             pubSubManager;
     private final PushSender                pushSender;
     private final AccountsManager           accountsManager;
+    private final FeatureFlagsManager       featureFlagsManager;
 
     private final Duration persistDelay;
 
@@ -46,15 +47,17 @@ public class RedisClusterMessagePersister implements Managed {
     static final int QUEUE_BATCH_LIMIT   = 100;
     static final int MESSAGE_BATCH_LIMIT = 100;
 
+    static final String ENABLE_PERSISTENCE_FLAG = "enable-cluster-persister";
+
     private static final Logger logger = LoggerFactory.getLogger(RedisClusterMessagePersister.class);
 
-    public RedisClusterMessagePersister(final RedisClusterMessagesCache messagesCache, final Messages messagesDatabase, final PubSubManager pubSubManager, final PushSender pushSender, final AccountsManager accountsManager, final Duration persistDelay) {
+    public RedisClusterMessagePersister(final RedisClusterMessagesCache messagesCache, final Messages messagesDatabase, final PubSubManager pubSubManager, final PushSender pushSender, final AccountsManager accountsManager, final FeatureFlagsManager featureFlagsManager, final Duration persistDelay) {
         this.messagesCache       = messagesCache;
         this.messagesDatabase    = messagesDatabase;
         this.pubSubManager       = pubSubManager;
         this.pushSender          = pushSender;
         this.accountsManager     = accountsManager;
-
+        this.featureFlagsManager = featureFlagsManager;
         this.persistDelay        = persistDelay;
     }
 
@@ -69,7 +72,10 @@ public class RedisClusterMessagePersister implements Managed {
 
         workerThread = new Thread(() -> {
             while (running) {
+                if (featureFlagsManager.isFeatureFlagActive(ENABLE_PERSISTENCE_FLAG)) {
                     persistNextQueues(Instant.now());
+                }
+
                 Util.sleep(100);
             }
         });
@@ -130,7 +136,7 @@ public class RedisClusterMessagePersister implements Managed {
             messagesCache.lockQueueForPersistence(queue);
 
             try {
-                /* int messageCount = 0;
+                int messageCount = 0;
                 List<MessageProtos.Envelope> messages;
 
                 do {
@@ -146,7 +152,7 @@ public class RedisClusterMessagePersister implements Managed {
                     }
                 } while (!messages.isEmpty());
 
-                queueSizeHistogram.update(messageCount); */
+                queueSizeHistogram.update(messageCount);
             } finally {
                 messagesCache.unlockQueueForPersistence(queue);
             }

service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagePersisterTest.java

@@ -4,7 +4,6 @@ import com.google.protobuf.ByteString;
 import io.lettuce.core.cluster.SlotHash;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.push.PushSender;
@@ -29,7 +28,6 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-@Ignore
 public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
 
     private ExecutorService              notificationExecutorService;
@@ -50,6 +48,9 @@ public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
     public void setUp() throws Exception {
         super.setUp();
 
+        final FeatureFlagsManager featureFlagsManager = mock(FeatureFlagsManager.class);
+        when(featureFlagsManager.isFeatureFlagActive(RedisClusterMessagePersister.ENABLE_PERSISTENCE_FLAG)).thenReturn(true);
+
         messagesDatabase = mock(Messages.class);
         accountsManager  = mock(AccountsManager.class);
         pubSubManager    = mock(PubSubManager.class);
@@ -61,7 +62,7 @@ public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
 
         notificationExecutorService = Executors.newSingleThreadExecutor();
         messagesCache               = new RedisClusterMessagesCache(getRedisCluster(), notificationExecutorService);
-        messagePersister            = new RedisClusterMessagePersister(messagesCache, messagesDatabase, pubSubManager, mock(PushSender.class), accountsManager, PERSIST_DELAY);
+        messagePersister            = new RedisClusterMessagePersister(messagesCache, messagesDatabase, pubSubManager, mock(PushSender.class), accountsManager, featureFlagsManager, PERSIST_DELAY);
     }
 
     @Override
@@ -136,6 +137,7 @@ public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
         verify(messagesDatabase, times(queueCount * messagesPerQueue)).store(any(UUID.class), any(MessageProtos.Envelope.class), anyString(), anyLong());
     }
 
+    @SuppressWarnings("SameParameterValue")
     private static String generateRandomQueueNameForSlot(final int slot) {
         final UUID uuid = UUID.randomUUID();