JeremyXin commented on code in PR #9869:
URL: https://github.com/apache/seatunnel/pull/9869#discussion_r2358892598
##########
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/client/RedisClusterClient.java:
##########
@@ -155,10 +162,66 @@ public void batchWriteZset(
int size = keys.size();
for (int i = 0; i < size; i++) {
if (rowKinds.get(i) == RowKind.DELETE || rowKinds.get(i) ==
RowKind.UPDATE_BEFORE) {
- RedisDataType.ZSET.del(this, keys.get(i), values.get(i));
+ RedisDataType.ZSET.del(jedis, keys.get(i), values.get(i));
+ } else {
+ RedisDataType.ZSET.set(jedis, keys.get(i), values.get(i),
expireSeconds);
+ }
+ }
+ }
+
+ /** In cluster mode, traverse and scan each node key */
+ @Override
+ public ScanResult<String> scanKeyResult(
+ final String cursor, final ScanParams params, final RedisDataType
type) {
+ // Create a composite cursor to traverse the cluster nodes: the format
is "Node Index: Node
+ // cursor"
+ int nodeIndex = 0;
+ String nodeCursor = cursor;
+ boolean isFirstScan = !cursor.contains(":");
+
+ if (!ScanParams.SCAN_POINTER_START.equals(cursor) &&
cursor.contains(":")) {
+ String[] parts = cursor.split(":", 2);
+ nodeIndex = Integer.parseInt(parts[0]);
+ nodeCursor = parts[1];
+ }
+
+ // All nodes have been scanned
+ if (nodeIndex >= nodes.size()) {
+ return new ScanResult<>(ScanParams.SCAN_POINTER_START, new
ArrayList<>());
+ }
+
+ ConnectionPool pool = nodes.get(nodeIndex).getValue();
+ List<String> resultKeys;
+ String nextCursor;
+
+ try (Jedis jedis = new Jedis(pool.getResource())) {
Review Comment:
done, replace the `scanKeyResult` method in the cluster mode with directly
using the `jedisCluster.scan` method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
