[BitSail-316][Connector] Fix bug in Redis sink connector (#229)

* [BitSail-316][Connector] Fix bug in Redis sink connector.
bytedance · Jan 31, 2023 · 2797644 · 2797644
1 parent 598e47c
commit 2797644
Show file tree

Hide file tree

Showing 27 changed files with 299 additions and 1,086 deletions.
diff --git a/bitsail-connectors/connector-redis/pom.xml b/bitsail-connectors/connector-redis/pom.xml
@@ -35,13 +35,6 @@
 
     <artifactId>connector-redis</artifactId>
     <dependencies>
-        <dependency>
-            <groupId>com.bytedance.bitsail</groupId>
-            <artifactId>bitsail-connector-redis</artifactId>
-            <version>${revision}</version>
-            <scope>compile</scope>
-        </dependency>
-
         <dependency>
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>
@@ -57,13 +50,6 @@
         </dependency>
 
         <!-- test -->
-        <dependency>
-            <groupId>com.bytedance.bitsail</groupId>
-            <artifactId>bitsail-connector-fake</artifactId>
-            <version>${revision}</version>
-            <scope>test</scope>
-        </dependency>
-
         <dependency>
             <groupId>com.bytedance.bitsail</groupId>
             <artifactId>connector-fake</artifactId>

diff --git a/...or-redis/src/main/java/com/bytedance/bitsail/connector/redis/config/JedisPoolOptions.java b/...or-redis/src/main/java/com/bytedance/bitsail/connector/redis/config/JedisPoolOptions.java
diff --git a/...nector-redis/src/main/java/com/bytedance/bitsail/connector/redis/config/RedisOptions.java b/...nector-redis/src/main/java/com/bytedance/bitsail/connector/redis/config/RedisOptions.java
diff --git a/...or-redis/src/main/java/com/bytedance/bitsail/connector/redis/constant/RedisConstants.java b/...or-redis/src/main/java/com/bytedance/bitsail/connector/redis/constant/RedisConstants.java
@@ -21,9 +21,4 @@ public class RedisConstants {
    * connector name
    */
   public static final String REDIS_CONNECTOR_NAME = "redis";
-
-  /**
-   * column size in sorted_set or hash
-   */
-  public static final int SORTED_SET_OR_HASH_COLUMN_SIZE = 3;
 }
diff --git a/...s/src/main/java/com/bytedance/bitsail/connector/redis/core/AbstractPipelineProcessor.java b/...s/src/main/java/com/bytedance/bitsail/connector/redis/core/AbstractPipelineProcessor.java
@@ -22,7 +22,7 @@
 import com.bytedance.bitsail.connector.redis.core.api.PipelineProcessor;
 import com.bytedance.bitsail.connector.redis.core.api.SplitPolicy;
 import com.bytedance.bitsail.connector.redis.core.jedis.JedisCommand;
-import com.bytedance.bitsail.connector.redis.core.jedis.JedisPluginErrorCode;
+import com.bytedance.bitsail.connector.redis.error.JedisPluginErrorCode;
 import com.bytedance.bitsail.connector.redis.error.RedisUnexpectedException;
 
 import com.github.rholder.retry.RetryException;

diff --git a/...ors/connector-redis/src/main/java/com/bytedance/bitsail/connector/redis/core/Command.java b/...ors/connector-redis/src/main/java/com/bytedance/bitsail/connector/redis/core/Command.java
@@ -17,7 +17,6 @@
 package com.bytedance.bitsail.connector.redis.core;
 
 import com.bytedance.bitsail.connector.redis.core.jedis.JedisCommand;
-import com.bytedance.bitsail.connector.redis.core.jedis.JedisCommandDescription;
 
 import com.alibaba.fastjson.JSON;
 import lombok.Data;
@@ -35,28 +34,28 @@ public class Command {
   private Map<byte[], byte[]> hash;
   private int ttlInSeconds;
 
-  public Command(JedisCommandDescription commandDescription, byte[] key, byte[] hashField, byte[] value) {
-    this(commandDescription, key, value);
+  public Command(JedisCommand jedisCommand, byte[] key, byte[] hashField, byte[] value, Integer additionalTTL) {
+    this(jedisCommand, key, value, additionalTTL);
     this.hashField = hashField;
   }
 
-  public Command(JedisCommandDescription commandDescription, byte[] key, Map<byte[], byte[]> hash) {
-    this.jedisCommand = commandDescription.getJedisCommand();
+  public Command(JedisCommand jedisCommand, byte[] key, Map<byte[], byte[]> hash, Integer additionalTTL) {
+    this.jedisCommand = jedisCommand;
     this.key = key;
     this.hash = hash;
-    this.ttlInSeconds = commandDescription.getAdditionalTTL() == null ? 0 : commandDescription.getAdditionalTTL();
+    this.ttlInSeconds = additionalTTL == null ? 0 : additionalTTL;
   }
 
-  public Command(JedisCommandDescription commandDescription, byte[] key, double score, byte[] value) {
-    this(commandDescription, key, value);
+  public Command(JedisCommand jedisCommand, byte[] key, double score, byte[] value, Integer additionalTTL) {
+    this(jedisCommand, key, value, additionalTTL);
     this.score = score;
   }
 
-  public Command(JedisCommandDescription commandDescription, byte[] key, byte[] value) {
-    this.jedisCommand = commandDescription.getJedisCommand();
+  public Command(JedisCommand jedisCommand, byte[] key, byte[] value, Integer additionalTTL) {
+    this.jedisCommand = jedisCommand;
     this.key = key;
     this.value = value;
-    this.ttlInSeconds = commandDescription.getAdditionalTTL() == null ? 0 : commandDescription.getAdditionalTTL();
+    this.ttlInSeconds = additionalTTL == null ? 0 : additionalTTL;
   }
 
   public String print() {

diff --git a/...ors/connector-redis/src/main/java/com/bytedance/bitsail/connector/redis/core/TtlType.java b/...ors/connector-redis/src/main/java/com/bytedance/bitsail/connector/redis/core/TtlType.java
@@ -29,7 +29,7 @@ public enum TtlType {
   SECOND(1);
 
   @Getter
-  int containSeconds;
+  final int containSeconds;
 
   TtlType(int containSeconds) {
     this.containSeconds = containSeconds;