add getAsyncCommands method for API

pom.xml

@@ -37,7 +37,7 @@ under the License.
 
  <groupId>io.github.jeff-zou</groupId>
  <artifactId>flink-connector-redis</artifactId>
- <version>1.2.3</version>
+ <version>1.2.4</version>
  <packaging>jar</packaging>
 
 

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisClusterContainer.java

@@ -4,6 +4,7 @@
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
+import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -405,4 +406,9 @@ public RedisFuture<Map<String, String>> hgetAll(String key) {
  }
  return result;
  }
+
+ @Override
+ public RedisClusterAsyncCommands getAsyncCommands() {
+ return clusterAsyncCommands;
+ }
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisCommandsContainer.java

@@ -1,6 +1,7 @@
 package org.apache.flink.streaming.connectors.redis.common.container;
 
 import io.lettuce.core.RedisFuture;
+import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -212,4 +213,11 @@ public interface RedisCommandsContainer extends Serializable {
  * @param value
  */
  void srem(String setName, String value);
+
+ /**
+ * get redis async commands.
+ *
+ * @return
+ */
+ RedisClusterAsyncCommands getAsyncCommands();
 }

src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java

@@ -4,6 +4,7 @@
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -421,4 +422,9 @@ public void srem(String setName, String value) {
  throw e;
  }
  }
+
+ @Override
+ public RedisClusterAsyncCommands getAsyncCommands() {
+ return asyncCommands;
+ }
 }