This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 016d58b47fb [Subscription] Relax group credentials and report
connection failures (#18658)
016d58b47fb is described below
commit 016d58b47fbc406541a7a7d0ac8b3a96d6c89a4e
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 17 11:18:34 2026 +0800
[Subscription] Relax group credentials and report connection failures
(#18658)
---
.../subscription/i18n/SubscriptionMessages.java | 2 +
.../subscription/i18n/SubscriptionMessages.java | 2 +
.../base/AbstractSubscriptionConsumer.java | 14 ++
.../base/AbstractSubscriptionProviders.java | 24 ++-
.../SubscriptionConsumerConnectionFailureTest.java | 167 +++++++++++++++++++++
.../iotdb/confignode/i18n/ManagerMessages.java | 2 +
.../iotdb/confignode/i18n/ManagerMessages.java | 2 +
.../persistence/subscription/SubscriptionInfo.java | 8 +-
.../meta/consumer/ConsumerGroupMeta.java | 31 +---
.../consumer/ConsumerGroupDeSerTest.java | 26 ++++
10 files changed, 240 insertions(+), 38 deletions(-)
diff --git
a/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
b/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
index 95f26d55d68..3693c7da2ca 100644
---
a/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
+++
b/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
@@ -155,6 +155,8 @@ public final class SubscriptionMessages {
+ " provider unavailable";
public static final String
LOG_EXCEPTION_OCCURRED_ARG_CLOSING_REMOVING_SUBSCRIPTION_PROVIDER_ARG_BECAUSE_ARG_2EC38739
= "Exception occurred when {} closing and removing subscription provider {}
because of {}";
public static final String
EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_5DB83198
= "Cluster has no available subscription providers to connect with initial
endpoints %s";
+ public static final String
EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_712D99EA
=
+ "Cluster has no available subscription providers to connect with initial
endpoints %s. Connection failures: %s";
public static final String
LOG_ARG_HAS_BEEN_LAUNCHED_SET_CORE_POOL_SIZE_ARG_WILL_0FDECBE3 = "{} has been
launched, set core pool size to {} will be ignored";
public static final String
LOG_INTERRUPT_WORKER_WHICH_MAY_CAUSE_SOME_TASK_INCONSISTENT_PLEASE_CHECK_04926D9F
= "Interrupt the worker, which may cause some task inconsistent. Please check
the biz logs.";
public static final String
LOG_THREAD_POOL_CAN_T_SHUTDOWN_EVEN_INTERRUPTING_WORKER_THREADS_WHICH_A49166F9 =
diff --git
a/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
b/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
index 562a4d72572..e3eea001964 100644
---
a/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
+++
b/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
@@ -143,6 +143,8 @@ public final class SubscriptionMessages {
public static final String
LOG_ARG_FAILED_SENDING_HEARTBEAT_SUBSCRIPTION_PROVIDER_ARG_BECAUSE_ARG_SET_0B38FB1F
= "{} 无法向 SubscriptionProvider {} 发送心跳，原因：{}，将 SubscriptionProvider 设为不可用";
public static final String
LOG_EXCEPTION_OCCURRED_ARG_CLOSING_REMOVING_SUBSCRIPTION_PROVIDER_ARG_BECAUSE_ARG_2EC38739
= "{} 关闭并移除 SubscriptionProvider {} 时发生异常，原因：{}";
public static final String
EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_5DB83198
= "集群没有可连接初始 endpoint %s 的可用 SubscriptionProvider";
+ public static final String
EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_712D99EA
=
+ "集群没有可连接初始 endpoint %s 的可用 SubscriptionProvider。连接失败原因：%s";
public static final String
LOG_ARG_HAS_BEEN_LAUNCHED_SET_CORE_POOL_SIZE_ARG_WILL_0FDECBE3 = "{} 已启动，将 core
pool size 设置为 {} 的操作会被忽略";
public static final String
LOG_INTERRUPT_WORKER_WHICH_MAY_CAUSE_SOME_TASK_INCONSISTENT_PLEASE_CHECK_04926D9F
= "中断 worker 可能造成某些任务不一致。请检查业务日志。";
public static final String
LOG_THREAD_POOL_CAN_T_SHUTDOWN_EVEN_INTERRUPTING_WORKER_THREADS_WHICH_A49166F9 =
diff --git
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index 38b16ed7cf9..73a51aa9923 100644
---
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -585,6 +585,20 @@ abstract class AbstractSubscriptionConsumer implements
AutoCloseable {
return provider;
}
+ String sanitizeConnectionFailureMessage(final Throwable throwable) {
+ String message = throwable.getMessage();
+ if (Objects.isNull(message) || message.isEmpty()) {
+ message = throwable.getClass().getName();
+ }
+ if (Objects.nonNull(password) && !password.isEmpty()) {
+ message = message.replace(password, "***");
+ }
+ if (Objects.nonNull(encryptedPassword) && !encryptedPassword.isEmpty()) {
+ message = message.replace(encryptedPassword, "***");
+ }
+ return message;
+ }
+
/////////////////////////////// file ops ///////////////////////////////
private Path getFileDir(final String topicName) throws IOException {
diff --git
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java
index be164a7da6a..d89fe998cdf 100644
---
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java
+++
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java
@@ -31,6 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -81,6 +82,8 @@ final class AbstractSubscriptionProviders {
// close stale providers
closeProviders();
+ final Map<TEndPoint, String> connectionFailures = new LinkedHashMap<>();
+ final List<Throwable> connectionFailureCauses = new ArrayList<>();
for (final TEndPoint endPoint : initialEndpoints) {
final AbstractSubscriptionProvider defaultProvider;
final int defaultDataNodeId;
@@ -88,6 +91,8 @@ final class AbstractSubscriptionProviders {
try {
defaultProvider = consumer.constructProviderAndHandshake(endPoint);
} catch (final Exception e) {
+ connectionFailures.put(endPoint,
consumer.sanitizeConnectionFailureMessage(e));
+ connectionFailureCauses.add(e);
LOGGER.warn(
SubscriptionMessages.LOG_ARG_FAILED_CREATE_CONNECTION_ARG_BECAUSE_ARG_E536E22A,
consumer,
@@ -136,11 +141,20 @@ final class AbstractSubscriptionProviders {
}
if (hasNoAvailableProviders()) {
- throw new SubscriptionConnectionException(
- String.format(
- SubscriptionMessages
-
.EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_5DB83198,
- initialEndpoints));
+ final SubscriptionConnectionException exception =
+ new SubscriptionConnectionException(
+ String.format(
+ SubscriptionMessages
+
.EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_712D99EA,
+ initialEndpoints,
+ connectionFailures),
+ connectionFailureCauses.isEmpty()
+ ? null
+ : connectionFailureCauses.get(connectionFailureCauses.size()
- 1));
+ for (int i = 0; i < connectionFailureCauses.size() - 1; i++) {
+ exception.addSuppressed(connectionFailureCauses.get(i));
+ }
+ throw exception;
}
nextDataNodeId = subscriptionProviders.firstKey();
diff --git
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerConnectionFailureTest.java
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerConnectionFailureTest.java
new file mode 100644
index 00000000000..9570ec6cf3e
--- /dev/null
+++
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerConnectionFailureTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.session.subscription.consumer.base;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import
org.apache.iotdb.rpc.subscription.exception.SubscriptionConnectionException;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
+import
org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribeHeartbeatResp;
+import org.apache.iotdb.session.AbstractSessionBuilder;
+import org.apache.iotdb.session.subscription.SubscriptionTreeSessionBuilder;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+public class SubscriptionConsumerConnectionFailureTest {
+
+ private static final String HOST = "127.0.0.1";
+ private static final int FIRST_PORT = 10_011;
+ private static final String PASSWORD = "test-password";
+
+ @Test
+ public void testOpenReportsAllInitialEndpointHandshakeFailures() {
+ final TestPullConsumer consumer = new TestPullConsumer();
+
+ final SubscriptionConnectionException exception =
+ Assert.assertThrows(SubscriptionConnectionException.class,
consumer::open);
+
+ Assert.assertTrue(exception.getMessage().contains(HOST));
+
Assert.assertTrue(exception.getMessage().contains(String.valueOf(FIRST_PORT)));
+
Assert.assertTrue(exception.getMessage().contains(String.valueOf(FIRST_PORT +
1)));
+ Assert.assertTrue(exception.getMessage().contains("first endpoint
handshake failed"));
+ Assert.assertTrue(exception.getMessage().contains("second endpoint
handshake failed"));
+ Assert.assertFalse(exception.getMessage().contains(PASSWORD));
+ Assert.assertNotNull(exception.getCause());
+ Assert.assertEquals(1, exception.getSuppressed().length);
+ }
+
+ private static class TestPullConsumer extends
AbstractSubscriptionPullConsumer {
+
+ private TestPullConsumer() {
+ super(
+ new AbstractSubscriptionPullConsumerBuilder()
+ .nodeUrls(Arrays.asList(HOST + ":" + FIRST_PORT, HOST + ":" +
(FIRST_PORT + 1)))
+ .username("test-user")
+ .password(PASSWORD));
+ }
+
+ @Override
+ protected AbstractSubscriptionProvider constructSubscriptionProvider(
+ final TEndPoint endPoint,
+ final String username,
+ final String password,
+ final String encryptedPassword,
+ final String consumerId,
+ final String consumerGroupId,
+ final String ownerId,
+ final Long ownerEpoch,
+ final int thriftMaxFrameSize,
+ final long heartbeatIntervalMs,
+ final int connectionTimeoutInMs) {
+ return new TestSubscriptionProvider(
+ endPoint,
+ username,
+ password,
+ encryptedPassword,
+ consumerId,
+ consumerGroupId,
+ ownerId,
+ ownerEpoch,
+ thriftMaxFrameSize,
+ heartbeatIntervalMs,
+ connectionTimeoutInMs);
+ }
+ }
+
+ private static class TestSubscriptionProvider extends
AbstractSubscriptionProvider {
+
+ private TestSubscriptionProvider(
+ final TEndPoint endPoint,
+ final String username,
+ final String password,
+ final String encryptedPassword,
+ final String consumerId,
+ final String consumerGroupId,
+ final String ownerId,
+ final Long ownerEpoch,
+ final int thriftMaxFrameSize,
+ final long heartbeatIntervalMs,
+ final int connectionTimeoutInMs) {
+ super(
+ endPoint,
+ username,
+ password,
+ encryptedPassword,
+ consumerId,
+ consumerGroupId,
+ ownerId,
+ ownerEpoch,
+ thriftMaxFrameSize,
+ heartbeatIntervalMs,
+ connectionTimeoutInMs);
+ }
+
+ @Override
+ protected AbstractSessionBuilder constructSubscriptionSessionBuilder(
+ final String host,
+ final int port,
+ final String username,
+ final String password,
+ final String encryptedPassword,
+ final int thriftMaxFrameSize,
+ final int connectionTimeoutInMs) {
+ final boolean useEncryptedPassword = Objects.nonNull(encryptedPassword);
+ return new SubscriptionTreeSessionBuilder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(useEncryptedPassword ? encryptedPassword : password)
+ .useEncryptedPassword(useEncryptedPassword)
+ .thriftMaxFrameSize(thriftMaxFrameSize)
+ .connectionTimeoutInMs(connectionTimeoutInMs);
+ }
+
+ @Override
+ synchronized void handshake() throws SubscriptionException {
+ throw new SubscriptionConnectionException(
+ (getEndPoint().port == FIRST_PORT
+ ? "first endpoint handshake failed"
+ : "second endpoint handshake failed")
+ + " with password "
+ + PASSWORD);
+ }
+
+ @Override
+ synchronized void close() {
+ setUnavailable();
+ }
+
+ @Override
+ PipeSubscribeHeartbeatResp heartbeat(
+ final List<SubscriptionCommitContext> processorBufferedCommitContexts)
{
+ return new PipeSubscribeHeartbeatResp();
+ }
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 10ea4763af9..a61e8e2aa5b 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -143,6 +143,8 @@ public final class ManagerMessages {
"Failed to collect user name for user id {}";
public static final String
FAILED_TO_CREATE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS =
"Failed to create consumer {} in consumer group {}. Result status: {}.";
+ public static final String
EXCEPTION_FAILED_TO_SUBSCRIBE_TOPIC_ARG_FOR_CONSUMER_ARG_CONSUMER_DOES_NOT_EXIST_IN_CONSUMER_GROUP_BFF68F12
=
+ "Failed to subscribe topic %s for consumer %s because the consumer does
not exist in the consumer group";
public static final String FAILED_TO_CREATE_PEER_FOR_CONSENSUS_GROUP =
"Failed to create peer for consensus group";
public static final String FAILED_TO_CREATE_PIPE_RESULT_STATUS =
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 499e922f7a7..fa8fc8e3217 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -142,6 +142,8 @@ public final class ManagerMessages {
"为用户 id {} 收集用户名失败";
public static final String
FAILED_TO_CREATE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS =
"在 consumer group {} 中创建 consumer {} 失败。结果状态：{}。";
+ public static final String
EXCEPTION_FAILED_TO_SUBSCRIBE_TOPIC_ARG_FOR_CONSUMER_ARG_CONSUMER_DOES_NOT_EXIST_IN_CONSUMER_GROUP_BFF68F12
=
+ "订阅 topic %s 失败，因为 consumer %s 不存在于 consumer group 中";
public static final String FAILED_TO_CREATE_PEER_FOR_CONSENSUS_GROUP =
"为 consensus group 创建 peer 失败";
public static final String FAILED_TO_CREATE_PIPE_RESULT_STATUS =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
index 36b6f35c31d..6ea5152f4df 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
@@ -1144,12 +1144,14 @@ public class SubscriptionInfo implements
SnapshotProcessor {
throw new SubscriptionException(exceptionMessage);
}
- // 2.2. check username
+ // 2.2. check consumer membership
if (!consumerGroupMeta.allowSubscribeTopicForConsumer(topicName,
consumerId)) {
final String exceptionMessage =
String.format(
- "Failed to subscribe topic %s for consumer %s because
inconsistent username under the same consumer group",
- topicName, consumerId);
+ ManagerMessages
+
.EXCEPTION_FAILED_TO_SUBSCRIBE_TOPIC_ARG_FOR_CONSUMER_ARG_CONSUMER_DOES_NOT_EXIST_IN_CONSUMER_GROUP_BFF68F12,
+ topicName,
+ consumerId);
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
index c6153db8668..2efffb89182 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
@@ -149,22 +149,6 @@ public class ConsumerGroupMeta {
return;
}
final ConsumerMeta existedConsumerMeta =
consumerIdToConsumerMeta.values().iterator().next();
- final boolean match =
- Objects.equals(existedConsumerMeta.getUsername(),
consumerMeta.getUsername())
- && Objects.equals(existedConsumerMeta.getPassword(),
consumerMeta.getPassword());
- if (!match) {
- final String exceptionMessage =
- String.format(
- "Failed to create consumer %s because inconsistent username &
password under the same consumer group, expected %s:%s, actual %s:%s",
- consumerMeta.getConsumerId(),
- existedConsumerMeta.getUsername(),
- existedConsumerMeta.getPassword(),
- consumerMeta.getUsername(),
- consumerMeta.getPassword());
- LOGGER.warn(exceptionMessage);
- throw new SubscriptionException(exceptionMessage);
- }
-
final String expectedSqlDialect =
existedConsumerMeta.getConfig().getSqlDialect();
final String actualSqlDialect = consumerMeta.getConfig().getSqlDialect();
final boolean isExpectedTableModel =
@@ -268,20 +252,7 @@ public class ConsumerGroupMeta {
}
public boolean allowSubscribeTopicForConsumer(final String topic, final
String consumerId) {
- if (!consumerIdToConsumerMeta.containsKey(consumerId)) {
- return false;
- }
- final Set<String> subscribedConsumerIdSet =
topicNameToSubscribedConsumerIdSet.get(topic);
- if (Objects.isNull(subscribedConsumerIdSet)) {
- return true;
- }
- if (subscribedConsumerIdSet.isEmpty()) {
- return true;
- }
- final String subscribedConsumerId =
subscribedConsumerIdSet.iterator().next();
- return Objects.equals(
-
Objects.requireNonNull(consumerIdToConsumerMeta.get(subscribedConsumerId)).getUsername(),
-
Objects.requireNonNull(consumerIdToConsumerMeta.get(consumerId)).getUsername());
+ return consumerIdToConsumerMeta.containsKey(consumerId);
}
public void addSubscription(final String consumerId, final Set<String>
topics) {
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
index fc8d0122b2b..589571ab2b2 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
@@ -114,4 +114,30 @@ public class ConsumerGroupDeSerTest {
consumerGroupMeta.checkAuthorityBeforeJoinConsumerGroup(
new ConsumerMeta("table_consumer", 2,
tableConsumerAttributes)));
}
+
+ @Test
+ public void testConsumerGroupShouldAllowDifferentCredentials() throws
SubscriptionException {
+ final Map<String, String> firstConsumerAttributes = new HashMap<>();
+ firstConsumerAttributes.put("username", "first_user");
+ firstConsumerAttributes.put("password", "first_password");
+
+ final Map<String, String> secondConsumerAttributes = new HashMap<>();
+ secondConsumerAttributes.put("username", "second_user");
+ secondConsumerAttributes.put("password", "second_password");
+
+ final ConsumerGroupMeta consumerGroupMeta =
+ new ConsumerGroupMeta(
+ "test_consumer_group",
+ 1,
+ new ConsumerMeta("first_consumer", 1, firstConsumerAttributes));
+ final ConsumerMeta secondConsumer =
+ new ConsumerMeta("second_consumer", 2, secondConsumerAttributes);
+
+ consumerGroupMeta.checkAuthorityBeforeJoinConsumerGroup(secondConsumer);
+ consumerGroupMeta.addConsumer(secondConsumer);
+ consumerGroupMeta.addSubscription("first_consumer",
Collections.singleton("test_topic"));
+
+ Assert.assertTrue(
+ consumerGroupMeta.allowSubscribeTopicForConsumer("test_topic",
"second_consumer"));
+ }
}
