This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new c12764c05b [Fix][Connector-V2] Honor Redis named-user authentication
(#12402)
c12764c05b is described below
commit c12764c05bdfb1f8b2d908d0e34ab2d27bb6999e
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Sep 22 02:57:56 2026 +0000
[Fix][Connector-V2] Honor Redis named-user authentication (#12402)
Signed-off-by: Goutam Adwant <[email protected]>
---
docs/en/connectors/sink/Redis.md | 13 ++
docs/en/connectors/source/Redis.md | 12 +-
.../introduction/concepts/incompatible-changes.md | 12 +
docs/zh/connectors/sink/Redis.md | 12 +
docs/zh/connectors/source/Redis.md | 10 +-
.../introduction/concepts/incompatible-changes.md | 11 +
.../seatunnel/redis/config/RedisBaseOptions.java | 11 +-
.../seatunnel/redis/config/RedisParameters.java | 65 ++++--
.../connectors/seatunnel/redis/Redis7Test.java | 257 +++++++++++++++++++++
.../seatunnel/redis/RedisFactoryTest.java | 74 ++++++
.../seatunnel/e2e/connector/redis/Redis7IT.java | 46 ++++
.../connector/redis/RedisTestCaseTemplateIT.java | 2 +-
.../src/test/resources/redis-named-user.conf | 50 ++++
13 files changed, 555 insertions(+), 20 deletions(-)
diff --git a/docs/en/connectors/sink/Redis.md b/docs/en/connectors/sink/Redis.md
index eaa81845b0..760176e433 100644
--- a/docs/en/connectors/sink/Redis.md
+++ b/docs/en/connectors/sink/Redis.md
@@ -58,6 +58,19 @@ downloaded from Maven Central.
| multi_table_sink_replica | int | No | 1 |
Writer replica count for multi-table writes. |
| common-options | config | No | - | Sink
plugin common parameters. See [Sink Common
Options](../common-options/sink-common-options.md). |
+### Authentication
+
+In both `SINGLE` and `CLUSTER` mode, a nonblank `user` selects Redis ACL
authentication
+(`AUTH user auth`, Redis 6 or later). The connector does not create or modify
ACL users.
+Create the user and grant its required command and key permissions before
starting the job, including
+`INFO` for connector initialization, `SELECT` in `SINGLE` mode, and `CLUSTER
SLOTS` for topology
+discovery in `CLUSTER` mode.
+The password is passed unchanged, including whitespace; omitted or empty
`auth` is sent as an empty
+password and only works if the ACL user accepts it (for example, a `nopass`
user).
+
+If `user` is omitted, empty, or whitespace-only, nonblank `auth` uses
password-only authentication
+as the default user. If both options are omitted or blank, no authentication
command is sent.
+
## Write Rules
### key
diff --git a/docs/en/connectors/source/Redis.md
b/docs/en/connectors/source/Redis.md
index 77a203d473..093b4e5cdb 100644
--- a/docs/en/connectors/source/Redis.md
+++ b/docs/en/connectors/source/Redis.md
@@ -232,11 +232,19 @@ redis data types, support `key` `string` `hash` `list`
`set` `zset`
### user [string]
-redis authentication user, you need it when you connect to an encrypted cluster
+Redis ACL username (Redis 6 or later), supported in both `SINGLE` and
`CLUSTER` mode.
+When nonblank, the connector authenticates with `AUTH user auth`; it does not
create or modify ACL users.
+Create the user and grant the required command and key permissions before
starting the job, including
+`INFO` for connector initialization, `SELECT` in `SINGLE` mode, and `CLUSTER
SLOTS` for topology
+discovery in `CLUSTER` mode.
+If `user` is omitted, empty, or whitespace-only, a nonblank `auth` uses
password-only authentication
+as the default user; otherwise no authentication command is sent.
### auth [string]
-redis authentication password, you need it when you connect to an encrypted
cluster
+Redis authentication password. With a nonblank `user`, the password is passed
unchanged, including
+whitespace. An omitted or empty password is sent as an empty string and works
only if that ACL user
+accepts it (for example, a user configured with `nopass`).
### db_num [int]
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 99181067e1..b90ad91efd 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -5,6 +5,18 @@ You need to check this document before you upgrade to related
version.
## dev
+### Redis Authentication
+
+- Redis sources and sinks now authenticate as the configured nonblank `user`
in both `SINGLE` and
+ `CLUSTER` mode. Previously, `SINGLE` used password-only authentication
followed by `ACL SETUSER`,
+ and `CLUSTER` ignored `user`. Connection setup no longer creates or modifies
ACL users.
+- Before upgrading, create the intended ACL user and grant its required
command and key permissions,
+ including `INFO` for connector initialization, `SELECT` in `SINGLE` mode,
and `CLUSTER SLOTS` for
+ topology discovery in `CLUSTER` mode. Set `auth` to that user's password. An
omitted or empty password
+ is sent as an empty string when `user` is nonblank.
+- To keep using the default user, remove `user` and retain `auth` when a
password is required.
+ Named users require Redis 6 or later. Legacy configurations without a
username remain unchanged.
+
### RabbitMQ Connector
- **Breaking Change: `amqps://` connections now verify broker certificates**
diff --git a/docs/zh/connectors/sink/Redis.md b/docs/zh/connectors/sink/Redis.md
index c430f69bb6..3c59c1c1ab 100644
--- a/docs/zh/connectors/sink/Redis.md
+++ b/docs/zh/connectors/sink/Redis.md
@@ -57,6 +57,18 @@ Redis 接收器连接器可以在批处理或流处理作业中把上游数据
| multi_table_sink_replica | int | 否 | 1 |
多表写入时的写入器副本数。 |
| common-options | config | 否 | - |
接收器插件通用参数，详情请参考[接收器通用选项](../common-options/sink-common-options.md)。 |
+### 认证
+
+在 `SINGLE` 和 `CLUSTER` 模式下，非空白的 `user` 使用 Redis ACL 认证
+（`AUTH user auth`，需要 Redis 6 或更新版本）。连接器不会创建或修改 ACL 用户。
+启动作业前，请创建用户并授予所需的命令和键权限，包括初始化连接器所需的 `INFO`，
+`SINGLE` 模式所需的 `SELECT`，以及 `CLUSTER` 模式下拓扑发现所需的 `CLUSTER SLOTS`。
+密码将原样传递，包括空白字符；省略 `auth` 或使用空字符串时将发送空密码，
+仅当该 ACL 用户允许时才能成功认证（例如配置了 `nopass` 的用户）。
+
+若省略 `user`，或其值为空字符串、仅包含空白字符，则非空白的 `auth` 将用于默认用户的密码认证。
+如果两个选项都省略或为空白，则不发送认证命令。
+
## 写入规则
### key
diff --git a/docs/zh/connectors/source/Redis.md
b/docs/zh/connectors/source/Redis.md
index b930d1c824..77f84c018d 100644
--- a/docs/zh/connectors/source/Redis.md
+++ b/docs/zh/connectors/source/Redis.md
@@ -224,11 +224,17 @@ redis 数据类型, 支持 `key` `string` `hash` `list` `set` `zset`。
### user [string]
-Redis 认证身份用户，当连接到加密集群时需要使用
+Redis ACL 用户名（需要 Redis 6 或更新版本），支持 `SINGLE` 和 `CLUSTER` 模式。
+当用户名非空白时，连接器通过 `AUTH user auth` 认证，不会创建或修改 ACL 用户。
+启动作业前，请创建用户并授予所需的命令和键权限，包括初始化连接器所需的 `INFO`，
+`SINGLE` 模式所需的 `SELECT`，以及 `CLUSTER` 模式下拓扑发现所需的 `CLUSTER SLOTS`。
+若省略 `user`，或其值为空字符串、仅包含空白字符，则非空白的 `auth` 将用于默认用户的密码认证；
+否则不发送认证命令。
### auth [string]
-Redis 认证密钥，当连接到加密集群时需要使用
+Redis 认证密码。当 `user` 非空白时，密码将原样传递，包括空白字符。
+省略密码或使用空字符串时，将发送空密码，仅当该 ACL 用户允许时才能成功认证（例如配置了 `nopass` 的用户）。
### db_num [int]
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 8a3b55c330..fb0a541081 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -4,6 +4,17 @@
## dev
+### Redis 认证
+
+- Redis Source 和 Sink 现在会在 `SINGLE` 和 `CLUSTER` 模式下以非空白的 `user` 指定的用户认证。
+ 此前，`SINGLE` 模式先使用仅密码认证，再执行 `ACL SETUSER`；`CLUSTER` 模式忽略 `user`。
+ 连接初始化不再创建或修改 ACL 用户。
+- 升级前，请创建目标 ACL 用户并授予所需的命令和键权限，包括初始化连接器所需的 `INFO`，
+ `SINGLE` 模式所需的 `SELECT`，以及 `CLUSTER` 模式下拓扑发现所需的 `CLUSTER SLOTS`。
+ 将 `auth` 设置为该用户的密码。当 `user` 非空白时，省略密码或使用空字符串将发送空密码。
+- 如需继续使用默认用户，请移除 `user`，并在需要密码时保留 `auth`。
+ 命名用户需要 Redis 6 或更新版本；未配置用户名的旧配置行为保持不变。
+
### RabbitMQ Connector
- **破坏性变更：`amqps://` 连接现在会校验 Broker 证书**
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
index 690f69fd92..8ee37aae4a 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
@@ -50,7 +50,11 @@ public class RedisBaseOptions extends ConnectorCommonOptions
{
.stringType()
.noDefaultValue()
.withDescription(
- "redis authentication password, you need it when
you connect to an encrypted cluster");
+ "Redis authentication password for SINGLE and
CLUSTER modes. With a nonblank user,"
+ + " the password is passed unchanged;
omitted or empty auth sends an empty"
+ + " password, which the ACL user must
accept (for example, via nopass)."
+ + " With a blank user, nonblank auth
authenticates as the default user;"
+ + " otherwise no authentication command is
sent.");
public static final Option<Integer> DB_NUM =
Options.key("db_num")
@@ -64,7 +68,10 @@ public class RedisBaseOptions extends ConnectorCommonOptions
{
.stringType()
.noDefaultValue()
.withDescription(
- "redis authentication user, you need it when you
connect to an encrypted cluster");
+ "Redis ACL username (Redis 6 or later) for SINGLE
and CLUSTER modes."
+ + " When nonblank, the connector
authenticates with AUTH user auth"
+ + " without creating or modifying ACL
users. Omitted, empty, or"
+ + " whitespace-only user retains
default-user authentication behavior.");
public static final Option<String> KEY_PATTERN =
Options.key("keys")
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
index cac06d2b47..d8458a282b 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
@@ -29,6 +29,7 @@ import
org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisConnectorE
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.ConnectionPoolConfig;
+import redis.clients.jedis.DefaultJedisClientConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
@@ -144,11 +145,16 @@ public class RedisParameters implements Serializable {
public RedisClient buildRedisClient() {
Jedis jedis = this.buildJedis();
- this.redisVersion = extractRedisVersion(jedis);
- if (mode.equals(RedisBaseOptions.RedisMode.SINGLE)) {
- return new RedisSingleClient(this, jedis, redisVersion);
- } else {
- return new RedisClusterClient(this, jedis, redisVersion);
+ try {
+ this.redisVersion = extractRedisVersion(jedis);
+ if (mode.equals(RedisBaseOptions.RedisMode.SINGLE)) {
+ return new RedisSingleClient(this, jedis, redisVersion);
+ } else {
+ return new RedisClusterClient(this, jedis, redisVersion);
+ }
+ } catch (RuntimeException | Error failure) {
+ closeAfterFailure(jedis, failure);
+ throw failure;
}
}
@@ -180,18 +186,27 @@ public class RedisParameters implements Serializable {
"Did not get the expected redis_version from the jedis.info()
method");
}
+ /**
+ * Uses named-user AUTH when user is nonblank; otherwise preserves
password-only or no-auth
+ * behavior. Authentication selects an existing identity, never
administers users via ACL
+ * SETUSER.
+ */
public Jedis buildJedis() {
switch (mode) {
case SINGLE:
Jedis jedis = new Jedis(host, port);
- if (StringUtils.isNotBlank(auth)) {
- jedis.auth(auth);
+ try {
+ if (StringUtils.isNotBlank(user)) {
+ jedis.auth(user, StringUtils.defaultString(auth));
+ } else if (StringUtils.isNotBlank(auth)) {
+ jedis.auth(auth);
+ }
+ jedis.select(dbNum);
+ return jedis;
+ } catch (RuntimeException | Error failure) {
+ closeAfterFailure(jedis, failure);
+ throw failure;
}
- if (StringUtils.isNotBlank(user)) {
- jedis.aclSetUser(user);
- }
- jedis.select(dbNum);
- return jedis;
case CLUSTER:
HashSet<HostAndPort> nodes = new HashSet<>();
for (String redisNode : redisNodes) {
@@ -202,7 +217,19 @@ public class RedisParameters implements Serializable {
}
ConnectionPoolConfig connectionPoolConfig = new
ConnectionPoolConfig();
JedisCluster jedisCluster;
- if (StringUtils.isNotBlank(auth)) {
+ if (StringUtils.isNotBlank(user)) {
+ jedisCluster =
+ new JedisCluster(
+ nodes,
+ DefaultJedisClientConfig.builder()
+ .user(user)
+
.password(StringUtils.defaultString(auth))
+
.connectionTimeoutMillis(JedisCluster.DEFAULT_TIMEOUT)
+
.socketTimeoutMillis(JedisCluster.DEFAULT_TIMEOUT)
+ .build(),
+ JedisCluster.DEFAULT_MAX_ATTEMPTS,
+ connectionPoolConfig);
+ } else if (StringUtils.isNotBlank(auth)) {
jedisCluster =
new JedisCluster(
nodes,
@@ -222,4 +249,16 @@ public class RedisParameters implements Serializable {
CommonErrorCode.OPERATION_NOT_SUPPORTED, "Not support
this redis mode");
}
}
+
+ /**
+ * Releases a connection after failed initialization, including an Error,
before the caller
+ * rethrows the original failure. Cleanup failures are suppressed to
preserve that failure.
+ */
+ private void closeAfterFailure(Jedis jedis, Throwable failure) {
+ try {
+ jedis.close();
+ } catch (RuntimeException | Error closeFailure) {
+ failure.addSuppressed(closeFailure);
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/Redis7Test.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/Redis7Test.java
index a2f008e799..87b7e3c3a0 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/Redis7Test.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/Redis7Test.java
@@ -16,16 +16,273 @@
*/
package org.apache.seatunnel.connectors.seatunnel.redis;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.redis.client.RedisClient;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.JedisWrapper;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions;
import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisContainerInfo;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
+import org.testcontainers.containers.GenericContainer;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.exceptions.JedisDataException;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BooleanSupplier;
+import java.util.stream.IntStream;
@DisabledOnOs(
value = OS.WINDOWS,
disabledReason = "There is no docker environment on the windows test
system")
public class Redis7Test extends RedisTemplateTest {
+ @Test
+ public void namedUserAuthentication() {
+ String user = "seatunnel_named";
+ // Sharing the default user's password must not silently select the
default identity.
+ jedis.aclSetUser(
+ user,
+ "reset",
+ "on",
+ ">" + password,
+ "~auth:*",
+ "+select",
+ "+info",
+ "+get",
+ "+set",
+ "+acl|whoami");
+ List<String> aclBefore = jedis.aclList();
+ try {
+ RedisParameters parameters = connectionParameters(user, password);
+ parameters.setDbNum(2);
+ try (Jedis connection = parameters.buildJedis()) {
+ Assertions.assertEquals(user, connection.aclWhoAmI());
+ connection.set("auth:selected", "value");
+ Assertions.assertThrows(
+ JedisDataException.class, () ->
connection.set("outside:auth", "value"));
+ Assertions.assertThrows(
+ JedisDataException.class, () ->
connection.aclSetUser("unexpected"));
+ }
+ jedis.select(2);
+ Assertions.assertEquals("value", jedis.get("auth:selected"));
+ jedis.del("auth:selected");
+ jedis.select(0);
+ RedisClient client = parameters.buildRedisClient();
+ try {
+ Assertions.assertEquals(7, parameters.getRedisVersion());
+ } finally {
+ client.close();
+ }
+ Assertions.assertEquals(aclBefore, jedis.aclList());
+ jedis.aclSetUser(user, "resetpass", ">named-password");
+ try (Jedis connection = connectionParameters(user,
"named-password").buildJedis()) {
+ Assertions.assertEquals(user, connection.aclWhoAmI());
+ }
+ } finally {
+ jedis.select(0);
+ jedis.aclDelUser(user);
+ }
+ }
+
+ @Test
+ public void namedUserMissingBlankAndWrongPassword() {
+ String user = "seatunnel_nopass";
+ jedis.aclSetUser(user, "reset", "on", "nopass", "+select",
"+acl|whoami");
+ try {
+ for (String auth : new String[] {null, "", " "}) {
+ try (Jedis connection = connectionParameters(user,
auth).buildJedis()) {
+ Assertions.assertEquals(user, connection.aclWhoAmI());
+ }
+ }
+ jedis.aclSetUser(user, "resetpass", ">named-password");
+ for (String auth : new String[] {null, "", " ",
"wrong-password"}) {
+ Assertions.assertThrows(
+ JedisDataException.class,
+ () -> connectionParameters(user, auth).buildJedis());
+ }
+ jedis.aclSetUser(user, "resetpass", "> ");
+ try (Jedis connection = connectionParameters(user, "
").buildJedis()) {
+ Assertions.assertEquals(user, connection.aclWhoAmI());
+ }
+ } finally {
+ jedis.aclDelUser(user);
+ }
+ }
+
+ @Test
+ public void legacyDefaultUserAuthentication() {
+ for (String user : new String[] {null, "", " "}) {
+ try (Jedis connection = connectionParameters(user,
password).buildJedis()) {
+ Assertions.assertEquals("default", connection.aclWhoAmI());
+ }
+ }
+ jedis.aclSetUser("default", "nopass");
+ try {
+ for (String user : new String[] {null, "", " "}) {
+ for (String auth : new String[] {null, "", " "}) {
+ try (Jedis connection = connectionParameters(user,
auth).buildJedis()) {
+ Assertions.assertEquals("default",
connection.aclWhoAmI());
+ }
+ }
+ }
+ List<String> aclBefore = jedis.aclList();
+ Assertions.assertThrows(
+ JedisDataException.class,
+ () -> connectionParameters("nonexistent",
null).buildJedis());
+ Assertions.assertEquals(aclBefore, jedis.aclList());
+ } finally {
+ jedis.aclSetUser("default", "resetpass", ">" + password);
+ }
+ }
+
+ @Test
+ public void failedInitializationClosesConnections() throws
InterruptedException {
+ String user = "seatunnel_denied";
+ jedis.aclSetUser(user, "reset", "on", ">named-password", "+select");
+ try {
+ assertConnectionCountUnchanged(() -> connectionParameters(user,
"wrong").buildJedis());
+ RedisParameters parameters = connectionParameters(user,
"named-password");
+ parameters.setDbNum(-1);
+ assertConnectionCountUnchanged(parameters::buildJedis);
+ parameters.setDbNum(0);
+ assertConnectionCountUnchanged(parameters::buildRedisClient);
+ jedis.aclSetUser(user, "-select");
+ assertConnectionCountUnchanged(parameters::buildJedis);
+ } finally {
+ jedis.aclDelUser(user);
+ }
+ }
+
+ private void assertConnectionCountUnchanged(Runnable connect) throws
InterruptedException {
+ int connections = jedis.clientList().split("\n").length;
+ Assertions.assertThrows(JedisDataException.class, connect::run);
+ awaitCondition(() -> connections ==
jedis.clientList().split("\n").length);
+ }
+
+ @Test
+ public void namedClusterAuthentication() throws InterruptedException,
UnknownHostException {
+ // A real cluster-enabled server owns all slots; no external cluster
is required.
+ try (GenericContainer<?> cluster =
+ new GenericContainer<>("redis:7")
+ .withExposedPorts(6379)
+ .withCommand(
+ "redis-server",
+ "--cluster-enabled",
+ "yes",
+ "--save",
+ "",
+ "--appendonly",
+ "no")) {
+ cluster.start();
+ try (Jedis admin = new Jedis(cluster.getHost(),
cluster.getFirstMappedPort())) {
+ admin.configSet(
+ "cluster-announce-ip",
+
InetAddress.getByName(cluster.getHost()).getHostAddress());
+ admin.configSet("cluster-announce-port",
cluster.getFirstMappedPort().toString());
+ admin.clusterAddSlots(IntStream.range(0, 16384).toArray());
+ awaitCondition(() ->
admin.clusterInfo().contains("cluster_state:ok"));
+ admin.aclSetUser(
+ "seatunnel_cluster",
+ "reset",
+ "on",
+ ">named-password",
+ "~auth:*",
+ "+cluster|slots",
+ "+info",
+ "+get",
+ "+set",
+ "+acl|whoami");
+ RedisParameters parameters =
+ connectionParameters("seatunnel_cluster",
"named-password");
+ parameters.setMode(RedisBaseOptions.RedisMode.CLUSTER);
+ parameters.setRedisNodes(
+ Collections.singletonList(
+ cluster.getHost() + ":" +
cluster.getFirstMappedPort()));
+ List<String> aclBefore = admin.aclList();
+ try (JedisWrapper connection = (JedisWrapper)
parameters.buildJedis()) {
+ for (String node : connection.getClusterNodes().keySet()) {
+ Assertions.assertEquals(
+ "seatunnel_cluster",
connection.getJedis(node).aclWhoAmI());
+ }
+ connection.set("auth:cluster", "value");
+ Assertions.assertEquals("value",
connection.get("auth:cluster"));
+ }
+ RedisClient client = parameters.buildRedisClient();
+ client.close();
+ Assertions.assertEquals(aclBefore, admin.aclList());
+
+ parameters.setUser("");
+ parameters.setAuth("");
+ assertClusterIdentity(parameters, "default");
+ admin.aclSetUser("default", "resetpass", ">named-password");
+ parameters.setAuth("named-password");
+ assertClusterIdentity(parameters, "default");
+ parameters.setUser("seatunnel_cluster");
+ assertClusterIdentity(parameters, "seatunnel_cluster");
+
+ parameters.setAuth("wrong-password");
+ int beforeAuthenticationFailure =
admin.clientList().split("\n").length;
+ Assertions.assertThrows(JedisDataException.class,
parameters::buildJedis);
+ awaitCondition(
+ () -> beforeAuthenticationFailure ==
admin.clientList().split("\n").length);
+ parameters.setAuth("named-password");
+ admin.aclSetUser("seatunnel_cluster", "-info");
+ int connections = admin.clientList().split("\n").length;
+ Assertions.assertThrows(RuntimeException.class,
parameters::buildRedisClient);
+ awaitCondition(() -> connections ==
admin.clientList().split("\n").length);
+
+ admin.aclSetUser("seatunnel_cluster", "nopass");
+ for (String auth : new String[] {null, "", " "}) {
+ parameters.setAuth(auth);
+ assertClusterIdentity(parameters, "seatunnel_cluster");
+ }
+ }
+ }
+ }
+
+ private void assertClusterIdentity(RedisParameters parameters, String
user) {
+ try (JedisWrapper connection = (JedisWrapper) parameters.buildJedis())
{
+ for (String node : connection.getClusterNodes().keySet()) {
+ Assertions.assertEquals(user,
connection.getJedis(node).aclWhoAmI());
+ }
+ }
+ }
+
+ private void awaitCondition(BooleanSupplier condition) throws
InterruptedException {
+ for (int attempt = 0; attempt < 100; attempt++) {
+ if (condition.getAsBoolean()) {
+ return;
+ }
+ Thread.sleep(50);
+ }
+ Assertions.assertTrue(condition.getAsBoolean());
+ }
+
+ private RedisParameters connectionParameters(String user, String auth) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("host", redisContainer.getHost());
+ config.put("port", redisContainer.getFirstMappedPort());
+ if (user != null) {
+ config.put("user", user);
+ }
+ if (auth != null) {
+ config.put("auth", auth);
+ }
+ RedisParameters parameters = new RedisParameters();
+ parameters.buildConnectionConfig(ReadonlyConfig.fromMap(config));
+ return parameters;
+ }
+
@Override
public RedisContainerInfo getRedisContainerInfo() {
return new RedisContainerInfo("redis-e2e", 6379, "SeaTunnel",
"redis:7");
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
index 67fb4a8989..f1b6d3850e 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
@@ -29,6 +29,8 @@ import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions;
+import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSink;
import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory;
@@ -38,6 +40,11 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.exceptions.JedisDataException;
import java.io.IOException;
import java.util.ArrayList;
@@ -53,6 +60,73 @@ class RedisFactoryTest {
private static final OptionRule SOURCE_RULE = new
RedisSourceFactory().optionRule();
private static final OptionRule SINK_RULE = new
RedisSinkFactory().optionRule();
+ @Test
+ void namedAuthenticationDoesNotAdministerUsers() {
+ try (MockedConstruction<Jedis> construction =
Mockito.mockConstruction(Jedis.class)) {
+ RedisParameters parameters = singleConnectionParameters();
+ parameters.setUser("named-user");
+ parameters.setAuth("named-password");
+ parameters.setDbNum(3);
+ Jedis connection = parameters.buildJedis();
+ Assertions.assertSame(construction.constructed().get(0),
connection);
+ Mockito.verify(connection).auth("named-user", "named-password");
+ Mockito.verify(connection).select(3);
+ Mockito.verifyNoMoreInteractions(connection);
+ connection.close();
+ }
+ }
+
+ @Test
+ void authenticationFailureClosesConnectionAndPreservesCause() {
+ JedisDataException failure = new JedisDataException("Authentication
failed");
+ RuntimeException closeFailure = new RuntimeException("Close failed");
+ try (MockedConstruction<Jedis> construction =
+ Mockito.mockConstruction(
+ Jedis.class,
+ (connection, context) -> {
+ Mockito.when(connection.auth("named-user",
"wrong")).thenThrow(failure);
+
Mockito.doThrow(closeFailure).when(connection).close();
+ })) {
+ RedisParameters parameters = singleConnectionParameters();
+ parameters.setUser("named-user");
+ parameters.setAuth("wrong");
+ Assertions.assertSame(
+ failure,
+ Assertions.assertThrows(JedisDataException.class,
parameters::buildJedis));
+ Mockito.verify(construction.constructed().get(0)).close();
+ Assertions.assertArrayEquals(new Throwable[] {closeFailure},
failure.getSuppressed());
+ }
+ }
+
+ @Test
+ void versionInitializationFailureClosesConnection() {
+ for (String info : Arrays.asList("redis_version:not-a-version",
"no-version")) {
+ RedisParameters parameters =
Mockito.spy(singleConnectionParameters());
+ Jedis connection = Mockito.mock(Jedis.class);
+ Mockito.doReturn(connection).when(parameters).buildJedis();
+ Mockito.when(connection.info()).thenReturn(info);
+ Assertions.assertThrows(RuntimeException.class,
parameters::buildRedisClient);
+ Mockito.verify(connection).close();
+ }
+ RedisParameters parameters = Mockito.spy(singleConnectionParameters());
+ Jedis connection = Mockito.mock(Jedis.class);
+ Mockito.doReturn(connection).when(parameters).buildJedis();
+ JedisDataException denied = new JedisDataException("INFO denied");
+ Mockito.when(connection.info()).thenThrow(denied);
+ Assertions.assertSame(
+ denied,
+ Assertions.assertThrows(JedisDataException.class,
parameters::buildRedisClient));
+ Mockito.verify(connection).close();
+ }
+
+ private RedisParameters singleConnectionParameters() {
+ RedisParameters parameters = new RedisParameters();
+ parameters.setHost("localhost");
+ parameters.setPort(6379);
+ parameters.setMode(RedisBaseOptions.RedisMode.SINGLE);
+ return parameters;
+ }
+
@Test
void optionRule() {
Assertions.assertNotNull(SOURCE_RULE);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
index 7e065837ff..33c6516539 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
@@ -17,12 +17,58 @@
package org.apache.seatunnel.e2e.connector.redis;
import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisContainerInfo;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.parallel.ResourceLock;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
@ResourceLock("redis-standalone-e2e")
public class Redis7IT extends RedisTestCaseTemplateIT {
+ @TestTemplate
+ public void testNamedUserSourceAndSink(TestContainer container)
+ throws IOException, InterruptedException {
+ jedis.aclSetUser(
+ "seatunnel_reader",
+ "reset",
+ "on",
+ ">reader-password",
+ "~acl:source:*",
+ "+select",
+ "+info",
+ "+scan",
+ "+type",
+ "+get",
+ "+mget");
+ jedis.aclSetUser(
+ "seatunnel_writer",
+ "reset",
+ "on",
+ ">writer-password",
+ "~acl:result",
+ "+select",
+ "+info",
+ "+lpush");
+ List<String> aclBefore = jedis.aclList();
+ jedis.set("acl:source:1", "{\"value\":\"named-user\"}");
+ try {
+ Container.ExecResult result =
container.executeJob("/redis-named-user.conf");
+ Assertions.assertEquals(0, result.getExitCode());
+ Assertions.assertEquals(
+ Collections.singletonList("named-user"),
jedis.lrange("acl:result", 0, -1));
+ Assertions.assertEquals(aclBefore, jedis.aclList());
+ } finally {
+ jedis.del("acl:source:1", "acl:result");
+ jedis.aclDelUser("seatunnel_reader", "seatunnel_writer");
+ }
+ }
+
@Override
public RedisContainerInfo getRedisContainerInfo() {
return new RedisContainerInfo("redis-e2e", 6379, "SeaTunnel",
"redis:7");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
index bb1650f93d..e6aabdae18 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
@@ -89,7 +89,7 @@ public abstract class RedisTestCaseTemplateIT extends
TestSuiteBase implements T
private GenericContainer<?> redisContainer;
- private Jedis jedis;
+ protected Jedis jedis;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-named-user.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-named-user.conf
new file mode 100644
index 0000000000..2f88da84e2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-named-user.conf
@@ -0,0 +1,50 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Redis {
+ host = "redis-e2e"
+ port = 6379
+ user = "seatunnel_reader"
+ auth = "reader-password"
+ keys = "acl:source:*"
+ data_type = string
+ format = json
+ schema = {
+ fields {
+ value = string
+ }
+ }
+ }
+}
+
+sink {
+ Redis {
+ host = "redis-e2e"
+ port = 6379
+ user = "seatunnel_writer"
+ auth = "writer-password"
+ key = "acl:result"
+ data_type = list
+ value_field = "value"
+ }
+}
