Change to enable support for Redis Cluster and Sentinel (#23)

* change to enable support for Redis Cluster and Sentinel * Update to unit tests to reflect changes --------- Co-authored-by: Neil <[email protected]>
orkes-io · Sep 13, 2023 · b407582 · b407582
1 parent 7f212ec
commit b407582
Show file tree

Hide file tree

Showing 11 changed files with 27 additions and 17 deletions.
diff --git a/persistence/src/main/java/com/netflix/conductor/redis/config/RedisClusterConfiguration.java b/persistence/src/main/java/com/netflix/conductor/redis/config/RedisClusterConfiguration.java
@@ -20,7 +20,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import com.netflix.conductor.redis.dynoqueue.ConfigurationHostSupplier;
@@ -40,7 +39,6 @@ public class RedisClusterConfiguration {
     // Same as redis.clients.jedis.BinaryJedisCluster
     protected static final int DEFAULT_MAX_ATTEMPTS = 5;
 
-    @Bean
     public JedisCluster getJedisCluster(RedisProperties properties) {
         GenericObjectPoolConfig<Jedis> genericObjectPoolConfig = new GenericObjectPoolConfig<>();
         genericObjectPoolConfig.setMaxTotal(properties.getMaxConnectionsPerHost());

diff --git a/persistence/src/main/java/com/netflix/conductor/redis/config/RedisSentinelConfiguration.java b/persistence/src/main/java/com/netflix/conductor/redis/config/RedisSentinelConfiguration.java
@@ -20,7 +20,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import com.netflix.conductor.redis.dynoqueue.ConfigurationHostSupplier;
@@ -37,7 +36,6 @@ public class RedisSentinelConfiguration {
 
     private static final Logger log = LoggerFactory.getLogger(RedisSentinelConfiguration.class);
 
-    @Bean
     protected JedisSentinel getJedisSentinel(RedisProperties properties) {
         GenericObjectPoolConfig<Jedis> genericObjectPoolConfig = new GenericObjectPoolConfig<>();
         genericObjectPoolConfig.setMinIdle(properties.getMinIdleConnections());

diff --git a/...stence/src/main/java/com/netflix/conductor/redis/config/RedisStandaloneConfiguration.java b/...stence/src/main/java/com/netflix/conductor/redis/config/RedisStandaloneConfiguration.java
@@ -15,7 +15,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import com.netflix.conductor.redis.dynoqueue.ConfigurationHostSupplier;
@@ -29,9 +28,8 @@
 @ConditionalOnProperty(name = "conductor.db.type", havingValue = "redis_standalone")
 public class RedisStandaloneConfiguration {
 
-    private static final Logger log = LoggerFactory.getLogger(RedisSentinelConfiguration.class);
+    private static final Logger log = LoggerFactory.getLogger(RedisStandaloneConfiguration.class);
 
-    @Bean
     public JedisPool getJedisPool(RedisProperties redisProperties) {
         JedisPoolConfig config = new JedisPoolConfig();
         config.setMinIdle(2);

diff --git a/persistence/src/main/java/com/netflix/conductor/redis/jedis/JedisStandalone.java b/persistence/src/main/java/com/netflix/conductor/redis/jedis/JedisStandalone.java
@@ -29,7 +29,7 @@
 /** A {@link JedisCommands} implementation that delegates to {@link JedisPool}. */
 @Component
 @ConditionalOnProperty(name = "conductor.db.type", havingValue = "redis_standalone")
-public class JedisStandalone implements JedisCommands {
+public class JedisStandalone implements OrkesJedisCommands {
 
     private final JedisPool jedisPool;
 
@@ -1154,10 +1154,12 @@ public List<StreamConsumersInfo> xinfoConsumers(String key, String group) {
         return executeInJedis(jedis -> jedis.xinfoConsumers(key, group));
     }
 
+    @Override
     public String set(byte[] key, byte[] value) {
         return executeInJedis(jedis -> jedis.set(key, value));
     }
 
+    @Override
     public byte[] getBytes(byte[] key) {
         return executeInJedis(jedis -> jedis.get(key));
     }

diff --git a/persistence/src/main/java/com/netflix/conductor/redis/jedis/OrkesJedisCommands.java b/persistence/src/main/java/com/netflix/conductor/redis/jedis/OrkesJedisCommands.java
@@ -0,0 +1,10 @@
+package com.netflix.conductor.redis.jedis;
+
+import redis.clients.jedis.commands.JedisCommands;
+
+public interface OrkesJedisCommands extends JedisCommands {
+
+    String set(byte[] key, byte[] value);
+
+    byte[] getBytes(byte[] key);
+}
diff --git a/persistence/src/main/java/com/netflix/conductor/redis/jedis/OrkesJedisProxy.java b/persistence/src/main/java/com/netflix/conductor/redis/jedis/OrkesJedisProxy.java
@@ -23,7 +23,6 @@
 
 import com.netflix.conductor.redis.config.AnyRedisCondition;
 
-import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.ScanResult;
 import redis.clients.jedis.Tuple;
@@ -37,10 +36,10 @@ public class OrkesJedisProxy {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(OrkesJedisProxy.class);
 
-    protected JedisStandalone jedisCommands;
+    protected OrkesJedisCommands jedisCommands;
 
-    public OrkesJedisProxy(JedisPool jedisPool) {
-        this.jedisCommands = new JedisStandalone(jedisPool);
+    public OrkesJedisProxy(OrkesJedisCommands jedisCommands) {
+        this.jedisCommands = jedisCommands;
     }
 
     public Set<String> zrange(String key, long start, long end) {

diff --git a/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisEventHandlerDAOTest.java b/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisEventHandlerDAOTest.java
@@ -31,6 +31,7 @@
 import com.netflix.conductor.core.config.ConductorProperties;
 import com.netflix.conductor.redis.config.RedisProperties;
 import com.netflix.conductor.redis.jedis.JedisMock;
+import com.netflix.conductor.redis.jedis.JedisStandalone;
 import com.netflix.conductor.redis.jedis.OrkesJedisProxy;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -56,7 +57,7 @@ public void init() {
         when(properties.getTaskDefCacheRefreshInterval()).thenReturn(Duration.ofSeconds(60));
         JedisPool jedisPool = mock(JedisPool.class);
         when(jedisPool.getResource()).thenReturn(new JedisMock());
-        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(jedisPool);
+        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(new JedisStandalone(jedisPool));
 
         redisEventHandlerDAO =
                 new RedisEventHandlerDAO(

diff --git a/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisExecutionDAOTest.java b/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisExecutionDAOTest.java
@@ -31,6 +31,7 @@
 import com.netflix.conductor.model.TaskModel;
 import com.netflix.conductor.redis.config.RedisProperties;
 import com.netflix.conductor.redis.jedis.JedisMock;
+import com.netflix.conductor.redis.jedis.JedisStandalone;
 import com.netflix.conductor.redis.jedis.OrkesJedisProxy;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -56,7 +57,7 @@ public void init() {
         when(properties.getTaskDefCacheRefreshInterval()).thenReturn(Duration.ofSeconds(60));
         JedisPool jedisPool = mock(JedisPool.class);
         when(jedisPool.getResource()).thenReturn(new JedisMock());
-        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(jedisPool);
+        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(new JedisStandalone(jedisPool));
 
         executionDAO =
                 new RedisExecutionDAO(

diff --git a/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisMetadataDAOTest.java b/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisMetadataDAOTest.java
@@ -34,6 +34,7 @@
 import com.netflix.conductor.dao.EventHandlerDAO;
 import com.netflix.conductor.redis.config.RedisProperties;
 import com.netflix.conductor.redis.jedis.JedisMock;
+import com.netflix.conductor.redis.jedis.JedisStandalone;
 import com.netflix.conductor.redis.jedis.OrkesJedisProxy;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -58,7 +59,7 @@ public void init() {
         when(properties.getTaskDefCacheRefreshInterval()).thenReturn(Duration.ofSeconds(60));
         JedisPool jedisPool = mock(JedisPool.class);
         when(jedisPool.getResource()).thenReturn(new JedisMock());
-        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(jedisPool);
+        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(new JedisStandalone(jedisPool));
         EventHandlerDAO eventHandlerDAO =
                 new RedisEventHandlerDAO(
                         orkesJedisProxy, objectMapper, conductorProperties, properties);

diff --git a/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisPollDataDAOTest.java b/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisPollDataDAOTest.java
@@ -26,6 +26,7 @@
 import com.netflix.conductor.dao.PollDataDAOTest;
 import com.netflix.conductor.redis.config.RedisProperties;
 import com.netflix.conductor.redis.jedis.JedisMock;
+import com.netflix.conductor.redis.jedis.JedisStandalone;
 import com.netflix.conductor.redis.jedis.OrkesJedisProxy;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -49,7 +50,7 @@ public void init() {
         when(properties.getTaskDefCacheRefreshInterval()).thenReturn(Duration.ofSeconds(60));
         JedisPool jedisPool = mock(JedisPool.class);
         when(jedisPool.getResource()).thenReturn(new JedisMock());
-        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(jedisPool);
+        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(new JedisStandalone(jedisPool));
 
         redisPollDataDAO =
                 new RedisPollDataDAO(

diff --git a/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisRateLimitDAOTest.java b/persistence/src/test/java/com/netflix/conductor/redis/dao/RedisRateLimitDAOTest.java
@@ -28,6 +28,7 @@
 import com.netflix.conductor.model.TaskModel;
 import com.netflix.conductor.redis.config.RedisProperties;
 import com.netflix.conductor.redis.jedis.JedisMock;
+import com.netflix.conductor.redis.jedis.JedisStandalone;
 import com.netflix.conductor.redis.jedis.OrkesJedisProxy;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -53,7 +54,7 @@ public void init() {
         when(properties.getTaskDefCacheRefreshInterval()).thenReturn(Duration.ofSeconds(60));
         JedisPool jedisPool = mock(JedisPool.class);
         when(jedisPool.getResource()).thenReturn(new JedisMock());
-        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(jedisPool);
+        OrkesJedisProxy orkesJedisProxy = new OrkesJedisProxy(new JedisStandalone(jedisPool));
 
         rateLimitingDao =
                 new RedisRateLimitingDAO(