From a7a3f0ecc65ea3f33be0c31119834b19785a0ea2 Mon Sep 17 00:00:00 2001
From: "kevin.bheda" <kevin.bheda@go-jek.com>
Date: Thu, 3 Feb 2022 12:09:56 +0530
Subject: [PATCH] fix: ttl not being set for keyvalue redis

---
 .../redis/dataentry/RedisKeyValueEntry.java   |  4 ++--
 .../dataentry/RedisKeyValueEntryTest.java     | 24 ++++++++++++-------
 2 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/src/main/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntry.java b/src/main/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntry.java
index 7e1390c2a..62ecaf5a6 100644
--- a/src/main/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntry.java
+++ b/src/main/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntry.java
@@ -19,16 +19,16 @@ public class RedisKeyValueEntry implements RedisDataEntry {
 
     @Override
     public void pushMessage(Pipeline jedisPipelined, RedisTtl redisTTL) {
-        redisTTL.setTtl(jedisPipelined, key);
         instrumentation.logDebug("key: {}, value: {}", key, value);
         jedisPipelined.set(key, value);
+        redisTTL.setTtl(jedisPipelined, key);
     }
 
     @Override
     public void pushMessage(JedisCluster jedisCluster, RedisTtl redisTTL) {
-        redisTTL.setTtl(jedisCluster, key);
         instrumentation.logDebug("key: {}, value: {}", key, value);
         jedisCluster.set(key, value);
+        redisTTL.setTtl(jedisCluster, key);
 
     }
 
diff --git a/src/test/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntryTest.java b/src/test/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntryTest.java
index 5332ec925..5397d6b9b 100644
--- a/src/test/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntryTest.java
+++ b/src/test/java/io/odpf/firehose/sink/redis/dataentry/RedisKeyValueEntryTest.java
@@ -5,7 +5,9 @@
 import io.odpf.firehose.sink.redis.ttl.NoRedisTtl;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.InOrder;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.Pipeline;
@@ -24,9 +26,15 @@ public class RedisKeyValueEntryTest {
     @Mock
     private JedisCluster jedisCluster;
 
+    private InOrder inOrderPipeline;
+    private InOrder inOrderJedis;
+
+
     @Before
     public void setup() {
         MockitoAnnotations.initMocks(this);
+        inOrderPipeline = Mockito.inOrder(pipeline);
+        inOrderJedis = Mockito.inOrder(jedisCluster);
 
     }
 
@@ -36,8 +44,8 @@ public void pushMessageWithNoTtl() {
         String value = "value";
         RedisKeyValueEntry redisKeyValueEntry = new RedisKeyValueEntry(key, value, instrumentation);
         redisKeyValueEntry.pushMessage(pipeline, new NoRedisTtl());
-        verify(pipeline, times(1)).set(key, value);
-        verify(pipeline, times(0)).expireAt(any(String.class), any(Long.class));
+        inOrderPipeline.verify(pipeline, times(1)).set(key, value);
+        inOrderPipeline.verify(pipeline, times(0)).expireAt(any(String.class), any(Long.class));
 
     }
 
@@ -47,8 +55,8 @@ public void pushMessageWithTtl() {
         String value = "value";
         RedisKeyValueEntry redisKeyValueEntry = new RedisKeyValueEntry(key, value, instrumentation);
         redisKeyValueEntry.pushMessage(pipeline, new DurationTtl(100));
-        verify(pipeline, times(1)).set(key, value);
-        verify(pipeline, times(1)).expire(key, 100);
+        inOrderPipeline.verify(pipeline, times(1)).set(key, value);
+        inOrderPipeline.verify(pipeline, times(1)).expire(key, 100);
     }
 
     @Test
@@ -67,8 +75,8 @@ public void pushMessageWithNoTtlUsingJedisCluster() {
         String value = "value";
         RedisKeyValueEntry redisKeyValueEntry = new RedisKeyValueEntry(key, value, instrumentation);
         redisKeyValueEntry.pushMessage(jedisCluster, new NoRedisTtl());
-        verify(jedisCluster, times(1)).set(key, value);
-        verify(jedisCluster, times(0)).expireAt(any(String.class), any(Long.class));
+        inOrderJedis.verify(jedisCluster, times(1)).set(key, value);
+        inOrderJedis.verify(jedisCluster, times(0)).expireAt(any(String.class), any(Long.class));
 
     }
 
@@ -78,8 +86,8 @@ public void pushMessageWithTtlUsingJedisCluster() {
         String value = "value";
         RedisKeyValueEntry redisKeyValueEntry = new RedisKeyValueEntry(key, value, instrumentation);
         redisKeyValueEntry.pushMessage(jedisCluster, new DurationTtl(100));
-        verify(jedisCluster, times(1)).set(key, value);
-        verify(jedisCluster, times(1)).expire(key, 100);
+        inOrderJedis.verify(jedisCluster, times(1)).set(key, value);
+        inOrderJedis.verify(jedisCluster, times(1)).expire(key, 100);
     }
 
     @Test