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 389e374abc1 Repair missing subscription brokers from consumer metadata
(#19021)
389e374abc1 is described below
commit 389e374abc15e876e68a1a413c1b5fb999bb8553
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 27 10:36:11 2026 +0800
Repair missing subscription brokers from consumer metadata (#19021)
* Repair missing subscription brokers from consumer metadata
* Fix subscription broker recovery rollback
* Serialize subscription broker repair with region creation
* Add cluster coverage for subscription broker recovery
---
...IoTDBConsensusSubscriptionBrokerRecoveryIT.java | 428 +++++++++++++++++++++
.../apache/iotdb/consensus/iot/IoTConsensus.java | 36 +-
.../consensus/iot/IoTConsensusRecoveryTest.java | 22 ++
.../agent/SubscriptionBrokerAgent.java | 27 ++
.../agent/SubscriptionConsumerAgent.java | 77 +++-
.../broker/ConsensusSubscriptionBroker.java | 46 +++
.../ConsensusSubscriptionCommitManager.java | 27 +-
.../ConsensusSubscriptionSetupHandler.java | 278 ++++++++-----
.../receiver/SubscriptionReceiverV1.java | 6 +
...ubscriptionConsumerAgentBrokerRecoveryTest.java | 279 ++++++++++++++
.../ConsensusSubscriptionSetupHandlerTest.java | 365 ++++++++++++++++++
.../receiver/SubscriptionReceiverV1Test.java | 33 ++
12 files changed, 1489 insertions(+), 135 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/IoTDBConsensusSubscriptionBrokerRecoveryIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/IoTDBConsensusSubscriptionBrokerRecoveryIT.java
new file mode 100644
index 00000000000..b46d6990722
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/IoTDBConsensusSubscriptionBrokerRecoveryIT.java
@@ -0,0 +1,428 @@
+/*
+ * 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.subscription.it.consensus.local;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.ClientPoolFactory;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp;
+import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq;
+import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.config.MppBaseConfig;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushSingleConsumerGroupMetaReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
+import org.apache.iotdb.rpc.subscription.payload.poll.PollPayload;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollRequest;
+import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollRequestType;
+import org.apache.iotdb.rpc.subscription.payload.poll.TopicProgress;
+import
org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeHandshakeReq;
+import org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribePollReq;
+import
org.apache.iotdb.rpc.subscription.payload.request.SubscriptionHeartbeatReq;
+import
org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribeHeartbeatResp;
+import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeResp;
+import org.apache.iotdb.session.subscription.SubscriptionSessionWrapper;
+import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
+import
org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
+import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({ClusterIT.class})
+public class IoTDBConsensusSubscriptionBrokerRecoveryIT extends
AbstractSubscriptionIT {
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(1)
+ .setDataReplicationFactor(3)
+ .setAutoCreateSchemaEnabled(true)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ // Recovery is driven explicitly below; a periodic metadata push must not
mask a broken
+ // handshake.
+ final Properties config = new Properties();
+ config.setProperty("subscription_enabled", "true");
+ ((MppBaseConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig())
+ .updateProperties(config);
+ config.setProperty("subscription_meta_syncer_initial_sync_delay_minutes",
"60");
+ config.setProperty("subscription_meta_syncer_sync_interval_minutes", "60");
+ ((MppBaseConfig)
EnvFactory.getEnv().getConfig().getConfigNodeCommonConfig())
+ .updateProperties(config);
+ EnvFactory.getEnv().initClusterEnvironment(1, 3);
+ }
+
+ @Override
+ @After
+ public void tearDown() throws Exception {
+ try {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ } finally {
+ super.tearDown();
+ }
+ }
+
+ @Test
+ public void testHandshakeRepairsMissingBrokersWithoutResubscribing() throws
Exception {
+ testMissingBrokerRecovery(true);
+ }
+
+ @Test
+ public void
testMetadataPushRepairsMissingBrokersWithoutRestoringCanceledTopics()
+ throws Exception {
+ testMissingBrokerRecovery(false);
+ }
+
+ private void testMissingBrokerRecovery(final boolean repairByHandshake)
throws Exception {
+ final ConsensusSubscriptionITSupport.TestIdentifiers ids =
+ ConsensusSubscriptionITSupport.newIdentifiers(
+ repairByHandshake ? "broker_handshake" : "broker_metadata");
+ final String canceledTopic = ids.getTopic() + "_canceled";
+ final List<SubscriptionSessionWrapper> probes = new ArrayList<>();
+ SubscriptionTreePullConsumer consumer = null;
+ final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clients
=
+ new IClientManager.Factory<TEndPoint,
SyncDataNodeInternalServiceClient>()
+ .createClientManager(
+ new
ClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
+ try {
+ ConsensusSubscriptionITSupport.bootstrapDatabase(ids.getDatabase());
+ ConsensusSubscriptionITSupport.createConsensusTopic(
+ ids.getTopic(), ids.getDatabase() + ".**");
+ ConsensusSubscriptionITSupport.createConsensusTopic(canceledTopic,
ids.getDatabase() + ".**");
+ consumer = createConsumer(ids, TimeUnit.HOURS.toMillis(1));
+ consumer.subscribe(ids.getTopic());
+ final Set<String> committedRows =
+ ConsensusSubscriptionITSupport.insertRows(ids.getDatabase(), "d0",
100L, 8, true);
+ assertRowsConsumed(consumer, committedRows);
+ // Creating a new subscription runs ConfigNode's commit-progress
synchronization without
+ // background metadata pushes, then this topic is left unsubscribed.
+ consumer.subscribe(canceledTopic);
+ consumer.unsubscribe(canceledTopic);
+ awaitConfigNodeCheckpoint(ids,
consumer.committedPositions(ids.getTopic()));
+ final ConsumerGroupMeta authoritativeMeta =
getConsumerGroupMeta(ids.getConsumerGroupId());
+ Assert.assertEquals(
+ Collections.singleton(ids.getTopic()),
authoritativeMeta.getSubscribedTopicNames());
+ final ConsumerConfig consumerConfig =
createConsumerConfig(authoritativeMeta, consumer);
+
+ for (final DataNodeWrapper node :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ final SubscriptionSessionWrapper probe = getProviderSession(consumer,
node);
+ probes.add(probe);
+ // A replacement-generation push can remove the old broker and then
fail while resolving
+ // topic metadata. Exercise that real RPC failure without changing
ConfigNode's metadata.
+ final ConsumerGroupMeta replacement =
+ new ConsumerGroupMeta(
+ ids.getConsumerGroupId(),
+ authoritativeMeta.getCreationTime() + 1,
+ authoritativeMeta.getConsumerMeta(ids.getConsumerId()));
+ replacement.addSubscription(
+ ids.getConsumerId(), Collections.singleton(ids.getTopic() +
"_missing"));
+ final TPushConsumerGroupMetaResp failure = pushMeta(clients, node,
replacement);
+ Assert.assertEquals(
+ failure.toString(),
+ TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode(),
+ failure.getStatus().getCode());
+ Assert.assertTrue(failure.isSetExceptionMessages());
+ Assert.assertFalse(failure.getExceptionMessages().isEmpty());
+ final TPipeSubscribeResp missingBroker =
+ probe
+ .getSessionConnection()
+ .pipeSubscribe(
+ PipeSubscribePollReq.toTPipeSubscribeReq(
+ new SubscriptionPollRequest(
+ SubscriptionPollRequestType.POLL.getType(),
+ new
PollPayload(Collections.singleton(ids.getTopic())),
+ 1_000L,
+ 1_048_576L)));
+ Assert.assertEquals(
+ missingBroker.getStatus().toString(),
+ TSStatusCode.SUBSCRIPTION_POLL_ERROR.getStatusCode(),
+ missingBroker.getStatus().getCode());
+ }
+ Assert.assertEquals(authoritativeMeta,
getConsumerGroupMeta(ids.getConsumerGroupId()));
+ final Set<String> pendingRows =
+ ConsensusSubscriptionITSupport.insertRows(ids.getDatabase(), "d0",
200L, 8, true);
+
+ for (int index = 0; index < probes.size(); index++) {
+ final SubscriptionSessionWrapper probe = probes.get(index);
+ if (repairByHandshake) {
+ assertSuccess(
+ probe
+ .getSessionConnection()
+
.pipeSubscribe(PipeSubscribeHandshakeReq.toTPipeSubscribeReq(consumerConfig)));
+ } else {
+ final TPushConsumerGroupMetaResp repaired =
+ pushMeta(clients, EnvFactory.getEnv().getDataNodeWrapper(index),
authoritativeMeta);
+ Assert.assertEquals(
+ repaired.toString(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ repaired.getStatus().getCode());
+ }
+ final TPipeSubscribeResp heartbeat =
+
probe.getSessionConnection().pipeSubscribe(SubscriptionHeartbeatReq.toThriftReq());
+ assertSuccess(heartbeat);
+ Assert.assertEquals(
+ Collections.singleton(ids.getTopic()),
+
PipeSubscribeHeartbeatResp.fromTPipeSubscribeResp(heartbeat).getTopics().keySet());
+ }
+ Assert.assertEquals(authoritativeMeta,
getConsumerGroupMeta(ids.getConsumerGroupId()));
+ assertRowsConsumed(consumer, pendingRows);
+ assertRowsConsumed(
+ consumer,
+ ConsensusSubscriptionITSupport.insertRows(ids.getDatabase(), "d0",
300L, 8, true));
+ } finally {
+ try {
+ clients.close();
+ } finally {
+ ConsensusSubscriptionITSupport.cleanup(consumer, ids.getTopic(),
ids.getDatabase());
+ try (final SubscriptionTreeSession session =
+ new SubscriptionTreeSession(
+ EnvFactory.getEnv().getIP(),
Integer.parseInt(EnvFactory.getEnv().getPort()))) {
+ session.open();
+ session.dropTopicIfExists(canceledTopic);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testProviderReplacementPreservesUncommittedBatch() throws
Exception {
+ final ConsensusSubscriptionITSupport.TestIdentifiers ids =
+
ConsensusSubscriptionITSupport.newIdentifiers("broker_provider_replacement");
+ SubscriptionTreePullConsumer consumer = null;
+ try {
+ ConsensusSubscriptionITSupport.bootstrapDatabase(ids.getDatabase());
+ ConsensusSubscriptionITSupport.createConsensusTopic(
+ ids.getTopic(), ids.getDatabase() + ".**");
+ consumer = createConsumer(ids, 1_000L);
+ consumer.subscribe(ids.getTopic());
+ final Set<String> rows =
+ ConsensusSubscriptionITSupport.insertRows(ids.getDatabase(), "d0",
100L, 8, true);
+ final ConsensusSubscriptionITSupport.PolledMessageBatch batch =
+ ConsensusSubscriptionITSupport.pollFirstNonEmptyBatchWithoutCommit(
+ consumer, 60, Duration.ofSeconds(1));
+ Assert.assertFalse(batch.getMessages().isEmpty());
+ final TopicProgress progressBeforeReplacement =
consumer.committedPositions(ids.getTopic());
+ final ConsumerGroupMeta metaBeforeReplacement =
+ getConsumerGroupMeta(ids.getConsumerGroupId());
+ final Map<?, ?> providers = getProviders(consumer);
+ final Map<?, ?> oldProviders = new HashMap<>(providers);
+ Assert.assertEquals(3, oldProviders.size());
+ for (final Object provider : oldProviders.values()) {
+ ((SubscriptionSessionWrapper) getField(provider, "session"))
+ .getSessionConnection()
+ .forceClose();
+ }
+ Awaitility.await()
+ .pollInSameThread()
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(
+ () -> {
+ Assert.assertEquals(oldProviders.keySet(), providers.keySet());
+ for (final Object nodeId : oldProviders.keySet()) {
+ Assert.assertNotSame(oldProviders.get(nodeId),
providers.get(nodeId));
+ }
+ });
+ Assert.assertEquals(metaBeforeReplacement,
getConsumerGroupMeta(ids.getConsumerGroupId()));
+ Assert.assertEquals(progressBeforeReplacement,
consumer.committedPositions(ids.getTopic()));
+ consumer.commitSync(batch.getMessages());
+ final ConsensusSubscriptionITSupport.ConsumedRecords consumed =
batch.getConsumedRecords();
+ consumed.merge(
+ ConsensusSubscriptionITSupport.pollAndCommitUntilAtLeast(
+ consumer, rows.size() - consumed.getUniqueRowCount(), 20,
Duration.ofSeconds(1)));
+ ConsensusSubscriptionITSupport.assertExactRowKeys(rows, consumed);
+ assertRowsConsumed(
+ consumer,
+ ConsensusSubscriptionITSupport.insertRows(ids.getDatabase(), "d0",
200L, 8, true));
+ } finally {
+ ConsensusSubscriptionITSupport.cleanup(consumer, ids.getTopic(),
ids.getDatabase());
+ }
+ }
+
+ private static SubscriptionTreePullConsumer createConsumer(
+ final ConsensusSubscriptionITSupport.TestIdentifiers ids, final long
controlIntervalMs)
+ throws Exception {
+ final SubscriptionTreePullConsumer consumer =
+ new SubscriptionTreePullConsumer.Builder()
+ .host(EnvFactory.getEnv().getIP())
+ .port(Integer.parseInt(EnvFactory.getEnv().getPort()))
+ .consumerId(ids.getConsumerId())
+ .consumerGroupId(ids.getConsumerGroupId())
+ .heartbeatIntervalMs(controlIntervalMs)
+ .endpointsSyncIntervalMs(controlIntervalMs)
+ .autoCommit(false)
+ .buildPullConsumer();
+ consumer.open();
+ return consumer;
+ }
+
+ private static ConsumerConfig createConsumerConfig(
+ final ConsumerGroupMeta meta, final SubscriptionTreePullConsumer
consumer) throws Exception {
+ final Map<String, String> attributes =
+ new
HashMap<>(meta.getConsumerMeta(consumer.getConsumerId()).getConfig().getAttribute());
+ attributes.put(
+ ConsumerConstant.CONSUMER_INSTANCE_ID_KEY,
+ (String) getField(consumer, "consumerInstanceId"));
+ return new ConsumerConfig(attributes);
+ }
+
+ private static SubscriptionSessionWrapper getProviderSession(
+ final SubscriptionTreePullConsumer consumer, final DataNodeWrapper node)
throws Exception {
+ final TEndPoint endpoint = new TEndPoint(node.getIp(), node.getPort());
+ for (final Object provider : getProviders(consumer).values()) {
+ if (endpoint.equals(getField(provider, "endPoint"))) {
+ return (SubscriptionSessionWrapper) getField(provider, "session");
+ }
+ }
+ throw new AssertionError("Missing subscription provider for " + endpoint);
+ }
+
+ private static ConsumerGroupMeta getConsumerGroupMeta(final String
consumerGroupId)
+ throws Exception {
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ final TGetAllSubscriptionInfoResp resp = client.getAllSubscriptionInfo();
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
resp.getStatus().getCode());
+ for (final ByteBuffer buffer : resp.getAllSubscriptionInfo()) {
+ final ConsumerGroupMeta meta = ConsumerGroupMeta.deserialize(buffer);
+ if (consumerGroupId.equals(meta.getConsumerGroupId())) {
+ return meta;
+ }
+ }
+ throw new AssertionError("Missing consumer group " + consumerGroupId);
+ }
+ }
+
+ private static TPushConsumerGroupMetaResp pushMeta(
+ final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
clients,
+ final DataNodeWrapper node,
+ final ConsumerGroupMeta meta)
+ throws Exception {
+ try (final SyncDataNodeInternalServiceClient client =
+ clients.borrowClient(new TEndPoint(node.getInternalAddress(),
node.getInternalPort()))) {
+ return client.pushSingleConsumerGroupMeta(
+ new
TPushSingleConsumerGroupMetaReq().setConsumerGroupMeta(meta.serialize()));
+ }
+ }
+
+ private static void awaitConfigNodeCheckpoint(
+ final ConsensusSubscriptionITSupport.TestIdentifiers ids, final
TopicProgress checkpoint) {
+ Assert.assertNotNull(checkpoint);
+ Assert.assertFalse(checkpoint.getRegionProgress().isEmpty());
+ Awaitility.await()
+ .pollInSameThread()
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(
+ () -> {
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
+ EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ for (final Map.Entry<String, RegionProgress> entry :
+ checkpoint.getRegionProgress().entrySet()) {
+ final TGetCommitProgressResp resp =
+ client.getCommitProgress(
+ new TGetCommitProgressReq(
+ ids.getConsumerGroupId(),
+ ids.getTopic(),
+
ConsensusGroupId.Factory.createFromString(entry.getKey()).getId(),
+ 0));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
resp.getStatus().getCode());
+ Assert.assertTrue(resp.isSetCommittedRegionProgress());
+ Assert.assertEquals(
+ entry.getValue(),
+ RegionProgress.deserialize(
+ ByteBuffer.wrap(resp.getCommittedRegionProgress())));
+ }
+ }
+ });
+ }
+
+ private static void assertRowsConsumed(
+ final SubscriptionTreePullConsumer consumer, final Set<String>
expectedRows)
+ throws Exception {
+ ConsensusSubscriptionITSupport.assertExactRowKeys(
+ expectedRows,
+ ConsensusSubscriptionITSupport.pollAndCommitUntilAtLeast(
+ consumer, expectedRows.size(), 60, Duration.ofSeconds(1)));
+ }
+
+ private static void assertSuccess(final TPipeSubscribeResp resp) {
+ Assert.assertEquals(
+ resp.getStatus().toString(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ resp.getStatus().getCode());
+ }
+
+ private static Map<?, ?> getProviders(final SubscriptionTreePullConsumer
consumer)
+ throws Exception {
+ return (Map<?, ?>) getField(getField(consumer, "providers"),
"subscriptionProviders");
+ }
+
+ private static Object getField(final Object target, final String name)
throws Exception {
+ for (Class<?> type = target.getClass(); type != null; type =
type.getSuperclass()) {
+ try {
+ final Field field = type.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(target);
+ } catch (final NoSuchFieldException ignored) {
+ // Client state is declared in the consumer/provider base classes.
+ }
+ }
+ throw new NoSuchFieldException(name);
+ }
+}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 63ddcc6eba8..28b0458c8b6 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -82,6 +82,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
@@ -104,6 +105,7 @@ public class IoTConsensus implements IConsensus {
private final IStateMachine.Registry registry;
private final Map<ConsensusGroupId, IoTConsensusServerImpl> stateMachineMap =
new ConcurrentHashMap<>();
+ private final Set<ConsensusGroupId> pendingPeerCreatedCallbacks =
ConcurrentHashMap.newKeySet();
private final IoTConsensusRPCService service;
private final RegisterManager registerManager = new RegisterManager();
private final UserDataTransferAuditHandler userDataTransferAuditHandler;
@@ -223,6 +225,7 @@ public class IoTConsensus implements IConsensus {
config,
userDataTransferAuditHandler,
userDataTransferAuditClassifier);
+ pendingPeerCreatedCallbacks.add(consensusGroupId);
stateMachineMap.put(consensusGroupId, consensus);
}
} catch (DiskSpaceInsufficientException e) {
@@ -253,15 +256,19 @@ public class IoTConsensus implements IConsensus {
}
stateMachineMap.forEach(
(consensusGroupId, consensus) -> {
- final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> callback
= onPeerRecovering;
- if (callback != null) {
- callback.accept(consensusGroupId, consensus);
+ try {
+ final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl>
callback = onPeerRecovering;
+ if (callback != null) {
+ callback.accept(consensusGroupId, consensus);
+ }
+ consensus.start();
+ // Subscription metadata may have been loaded before any local
consensus peers existed.
+ // Recovered peers must trigger the same binding as newly created
peers, after startup
+ // has restored their indices and installed the pre-start
WAL-retention guards.
+ notifyPeerCreated(consensusGroupId, consensus);
+ } finally {
+ pendingPeerCreatedCallbacks.remove(consensusGroupId);
}
- consensus.start();
- // Subscription metadata may have been loaded before any local
consensus peers existed.
- // Recovered peers must trigger the same binding as newly created
peers, after startup
- // has restored their indices and installed the pre-start
WAL-retention guards.
- notifyPeerCreated(consensusGroupId, consensus);
});
}
@@ -355,6 +362,7 @@ public class IoTConsensus implements IConsensus {
throw new RuntimeException(e);
}
impl.start();
+ pendingPeerCreatedCallbacks.add(groupId);
return impl;
}))
.orElseThrow(
@@ -367,7 +375,16 @@ public class IoTConsensus implements IConsensus {
throw new ConsensusGroupAlreadyExistException(groupId);
}
- notifyPeerCreated(groupId, stateMachineMap.get(groupId));
+ try {
+ notifyPeerCreated(groupId, stateMachineMap.get(groupId));
+ } finally {
+ pendingPeerCreatedCallbacks.remove(groupId);
+ }
+ }
+
+ /** Returns whether a new or recovered local peer is still waiting for its
creation callback. */
+ public boolean isPeerCreationCallbackPending(final ConsensusGroupId groupId)
{
+ return pendingPeerCreatedCallbacks.contains(groupId);
}
private void notifyPeerCreated(
@@ -412,6 +429,7 @@ public class IoTConsensus implements IConsensus {
if (!exist.get()) {
throw new ConsensusGroupNotExistException(groupId);
}
+ pendingPeerCreatedCallbacks.remove(groupId);
FileUtils.deleteFileOrDirectoryWithRateLimiter(
new File(buildPeerDir(storageDir, groupId)),
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusRecoveryTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusRecoveryTest.java
index fcbb2772a62..4e7131f1a5c 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusRecoveryTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusRecoveryTest.java
@@ -101,12 +101,14 @@ public class IoTConsensusRecoveryTest {
IoTConsensus.onPeerRecovering =
(groupId, server) -> {
assertSame(server, consensus.getImpl(groupId));
+ assertTrue(consensus.isPeerCreationCallbackPending(groupId));
assertFalse(stateMachines.get(groupId).started);
notifications.computeIfAbsent(groupId, ignored -> new
ArrayList<>()).add("recovering");
};
IoTConsensus.onNewPeerCreated =
(groupId, server) -> {
assertSame(server, consensus.getImpl(groupId));
+ assertTrue(consensus.isPeerCreationCallbackPending(groupId));
assertTrue(stateMachines.get(groupId).started);
notifications.computeIfAbsent(groupId, ignored -> new
ArrayList<>()).add("available");
};
@@ -116,6 +118,8 @@ public class IoTConsensusRecoveryTest {
assertEquals(2, notifications.size());
assertEquals(Arrays.asList("recovering", "available"),
notifications.get(firstRegion));
assertEquals(Arrays.asList("recovering", "available"),
notifications.get(secondRegion));
+ assertFalse(consensus.isPeerCreationCallbackPending(firstRegion));
+ assertFalse(consensus.isPeerCreationCallbackPending(secondRegion));
}
@Test
@@ -141,6 +145,24 @@ public class IoTConsensusRecoveryTest {
assertTrue(notifiedRegions.containsAll(Arrays.asList(firstRegion,
secondRegion)));
assertTrue(stateMachines.get(firstRegion).started);
assertTrue(stateMachines.get(secondRegion).started);
+ assertFalse(consensus.isPeerCreationCallbackPending(firstRegion));
+ assertFalse(consensus.isPeerCreationCallbackPending(secondRegion));
+ }
+
+ @Test
+ public void testNewPeerRemainsPendingUntilCallbackReturns() throws Exception
{
+ final DataRegionId regionId = new DataRegionId(3);
+ IoTConsensus.onNewPeerCreated =
+ (groupId, server) -> {
+ assertSame(server, consensus.getImpl(groupId));
+ assertTrue(consensus.getAllConsensusGroupIds().contains(groupId));
+ assertTrue(consensus.isPeerCreationCallbackPending(groupId));
+ throw new IllegalStateException("subscription binding failed");
+ };
+
+ createPeer(regionId);
+
+ assertFalse(consensus.isPeerCreationCallbackPending(regionId));
}
private IoTConsensus newConsensus() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
index 89d877665eb..9c448e743c1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
@@ -962,6 +962,33 @@ public class SubscriptionBrokerAgent {
prefetchingQueueCount.invalidate();
}
+ public Map<String, Set<String>> getConsensusQueueRegionIds(
+ final String consumerGroupId, final Set<String> topicNames) {
+ final Map<String, Set<String>> regionIdsByTopic = new LinkedHashMap<>();
+ final ConsensusSubscriptionBroker broker =
getConsensusBroker(consumerGroupId);
+ if (Objects.nonNull(broker)) {
+ for (final String topicName : topicNames) {
+ regionIdsByTopic.put(topicName, broker.getQueueRegionIds(topicName));
+ }
+ }
+ return regionIdsByTopic;
+ }
+
+ public boolean hasConsensusQueueForRegion(
+ final String consumerGroupId, final String topicName, final String
regionId) {
+ final ConsensusSubscriptionBroker broker =
getConsensusBroker(consumerGroupId);
+ return Objects.nonNull(broker) && broker.hasQueueForRegion(topicName,
regionId);
+ }
+
+ public void rollbackConsensusPrefetchingQueues(
+ final String consumerGroupId, final String topicName, final Set<String>
existingRegionIds) {
+ final ConsensusSubscriptionBroker broker =
getConsensusBroker(consumerGroupId);
+ if (Objects.nonNull(broker)) {
+ broker.rollbackConsensusPrefetchingQueues(topicName, existingRegionIds);
+ prefetchingQueueCount.invalidate();
+ }
+ }
+
public void unbindByRegion(final ConsensusGroupId regionId) {
int totalClosed = 0;
for (final ConsensusSubscriptionBroker broker :
getBrokers(ConsensusSubscriptionBroker.class)) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
index 8af488c5923..129330e88ce 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
@@ -30,6 +30,7 @@ import
org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Objects;
@@ -116,11 +117,7 @@ public class SubscriptionConsumerAgent {
// if consumer group meta does not exist on local agent
if (Objects.isNull(metaInAgent)) {
- SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId);
- ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
- consumerGroupId,
- metaFromCoordinator.getSubscribedTopicNames(),
- metaFromCoordinator.visibleUnder(true));
+ ensureConsumerGroupBrokerInternal(metaFromCoordinator);
replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator);
return;
}
@@ -145,12 +142,8 @@ public class SubscriptionConsumerAgent {
}
}
- ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
- consumerGroupId,
- metaFromCoordinator.getSubscribedTopicNames(),
- metaFromCoordinator.visibleUnder(true));
+ ensureConsumerGroupBrokerInternal(metaFromCoordinator);
replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator);
- // no need to create broker manually
return;
}
@@ -179,12 +172,7 @@ public class SubscriptionConsumerAgent {
newlySubscribedTopics);
applyTopicDiff(
- () -> {
- if (!newlySubscribedTopics.isEmpty()) {
- ConsensusSubscriptionSetupHandler.handleNewSubscriptions(
- consumerGroupId, newlySubscribedTopics);
- }
- },
+ () -> ensureConsumerGroupBrokerInternal(metaFromCoordinator),
() -> {
for (final String topicName : pipeTopicsUnsubByGroup) {
SubscriptionAgent.broker().removePrefetchingQueue(consumerGroupId,
topicName);
@@ -279,6 +267,48 @@ public class SubscriptionConsumerAgent {
}
}
+ /**
+ * Ensures that a non-empty consumer group has the local broker and its
consensus subscriptions.
+ *
+ * <p>The coordinator metadata is authoritative here. In particular, an
empty metadata must not
+ * recreate a broker or restore topics that were explicitly removed from the
group.
+ */
+ public void ensureConsumerGroupBroker(final String consumerGroupId) {
+ executeWithConsumerGroupLock(
+ consumerGroupId,
+ () -> {
+ final ConsumerGroupMeta consumerGroupMeta =
getConsumerGroupMeta(consumerGroupId);
+ if (Objects.nonNull(consumerGroupMeta)) {
+ ensureConsumerGroupBrokerInternal(consumerGroupMeta);
+ }
+ });
+ }
+
+ private void ensureConsumerGroupBrokerInternal(final ConsumerGroupMeta
consumerGroupMeta) {
+ if (consumerGroupMeta.isEmpty()) {
+ return;
+ }
+
+ final String consumerGroupId = consumerGroupMeta.getConsumerGroupId();
+ final boolean brokerExisted =
SubscriptionAgent.broker().isBrokerExist(consumerGroupId);
+ SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId);
+ try {
+ // setupConsensusSubscriptions is idempotent and also repairs a missing
consensus broker or
+ // queue when a pipe broker still exists.
+ ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
+ consumerGroupId,
+ consumerGroupMeta.getSubscribedTopicNames(),
+ consumerGroupMeta.visibleUnder(true));
+ } catch (final RuntimeException e) {
+ // Allow a later metadata push or handshake to retry after a partial
setup failure. Preserve
+ // an existing broker because it may still contain usable queues for the
current metadata.
+ if (!brokerExisted) {
+ SubscriptionAgent.broker().dropBroker(consumerGroupId);
+ }
+ throw e;
+ }
+ }
+
private ConsumerGroupMeta getConsumerGroupMeta(final String consumerGroupId)
{
acquireReadLock();
try {
@@ -307,7 +337,7 @@ public class SubscriptionConsumerAgent {
}
}
- void executeWithConsumerGroupLock(final String consumerGroupId, final
Runnable operation) {
+ public void executeWithConsumerGroupLock(final String consumerGroupId, final
Runnable operation) {
final ConsumerGroupOperationLock operationLock =
consumerGroupOperationLocks.compute(
consumerGroupId,
@@ -416,6 +446,19 @@ public class SubscriptionConsumerAgent {
}
}
+ public Set<String> getTopicNamesSubscribedByConsumerGroup(final String
consumerGroupId) {
+ acquireReadLock();
+ try {
+ final ConsumerGroupMeta consumerGroupMeta =
+ consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId);
+ return Objects.isNull(consumerGroupMeta)
+ ? Collections.emptySet()
+ : new LinkedHashSet<>(consumerGroupMeta.getSubscribedTopicNames());
+ } finally {
+ releaseReadLock();
+ }
+ }
+
/**
* Get all active subscriptions: consumerGroupId → set of subscribed topic
names. Used by
* consensus subscription auto-binding when a new DataRegion is created.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
index 60b268504a2..22c79ac7303 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
@@ -751,6 +751,52 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
closeAndRemoveConsensusPrefetchingQueues(topicName, true,
retainProgressAfterUnsubscribe);
}
+ public Set<String> getQueueRegionIds(final String topicName) {
+ synchronized (queueLifecycleLock) {
+ return topicNameToConsensusPrefetchingQueues
+ .getOrDefault(topicName, Collections.emptyList())
+ .stream()
+ .filter(queue -> !queue.isClosed())
+ .map(queue -> queue.getConsensusGroupId().toString())
+ .collect(Collectors.toSet());
+ }
+ }
+
+ public boolean hasQueueForRegion(final String topicName, final String
regionId) {
+ synchronized (queueLifecycleLock) {
+ return topicNameToConsensusPrefetchingQueues
+ .getOrDefault(topicName, Collections.emptyList())
+ .stream()
+ .anyMatch(
+ queue ->
+ !queue.isClosed() &&
regionId.equals(queue.getConsensusGroupId().toString()));
+ }
+ }
+
+ public void rollbackConsensusPrefetchingQueues(
+ final String topicName, final Set<String> existingRegionIds) {
+ final List<ConsensusPrefetchingQueue> queuesToClose = new ArrayList<>();
+ synchronized (queueLifecycleLock) {
+ final List<ConsensusPrefetchingQueue> queues =
+ topicNameToConsensusPrefetchingQueues.get(topicName);
+ if (Objects.isNull(queues)) {
+ return;
+ }
+ for (final ConsensusPrefetchingQueue queue : queues) {
+ if
(!existingRegionIds.contains(queue.getConsensusGroupId().toString())) {
+ queuesToClose.add(queue);
+ }
+ }
+ queues.removeAll(queuesToClose);
+ if (queues.isEmpty()) {
+ topicNameToConsensusPrefetchingQueues.remove(topicName, queues);
+ topicConsumerLastPollMs.remove(topicName);
+ topicOwnershipSnapshots.remove(topicName);
+ }
+ }
+ queuesToClose.forEach(queue -> queue.closeAsync(false, false));
+ }
+
public void wakeUpPrefetchingQueues(final String topicName, final boolean
isTableModel) {
final List<ConsensusPrefetchingQueue> queues =
topicNameToConsensusPrefetchingQueues.get(topicName);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
index ed49dda3f46..f9a27372d5e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
@@ -274,17 +274,36 @@ public class ConsensusSubscriptionCommitManager {
/** Restores only attempted topics to their exact state before setup began.
*/
void restoreSetupSnapshot(
final SetupSnapshot setupSnapshot, final Set<String>
attemptedTopicNames) {
+ restoreSetupSnapshot(setupSnapshot, attemptedTopicNames,
Collections.emptyMap());
+ }
+
+ void restoreSetupSnapshot(
+ final SetupSnapshot setupSnapshot,
+ final Set<String> attemptedTopicNames,
+ final Map<String, Set<String>> existingQueueRegionIds) {
for (final String topicName : attemptedTopicNames) {
final TopicSetupSnapshot topicSnapshot =
setupSnapshot.topicSnapshots.get(topicName);
if (Objects.isNull(topicSnapshot)) {
continue;
}
synchronized (getTopicPersistLock(topicSnapshot.topicFileKey)) {
- removeTopicStateFromMemory(setupSnapshot.consumerGroupId, topicName);
+ final Set<String> preservedRegionIds =
+ existingQueueRegionIds.getOrDefault(topicName,
Collections.emptySet());
+ // Existing queues can commit while setup is running. Keep their state
objects, progress,
+ // and initialization markers instead of rewinding them to the setup
snapshot.
+ removeTopicStateFromMemory(setupSnapshot.consumerGroupId, topicName,
preservedRegionIds);
CommitStateKey stateKeyForRewrite = null;
+ if (!preservedRegionIds.isEmpty()) {
+ stateKeyForRewrite =
+ getCommitStateKey(
+ setupSnapshot.consumerGroupId, topicName,
preservedRegionIds.iterator().next());
+ }
for (final Map.Entry<String, StateSetupSnapshot> entry :
topicSnapshot.stateSnapshots.entrySet()) {
+ if (preservedRegionIds.contains(entry.getKey())) {
+ continue;
+ }
final CommitStateKey stateKey =
getCommitStateKey(setupSnapshot.consumerGroupId, topicName,
entry.getKey());
final StateSetupSnapshot stateSnapshot = entry.getValue();
@@ -334,12 +353,14 @@ public class ConsensusSubscriptionCommitManager {
}
}
- private void removeTopicStateFromMemory(final String consumerGroupId, final
String topicName) {
+ private void removeTopicStateFromMemory(
+ final String consumerGroupId, final String topicName, final Set<String>
preservedRegionIds) {
final Iterator<Map.Entry<String, CommitStateKey>> keyIterator =
commitStateKeys.entrySet().iterator();
while (keyIterator.hasNext()) {
final Map.Entry<String, CommitStateKey> entry = keyIterator.next();
- if (!entry.getValue().sameTopic(consumerGroupId, topicName)) {
+ if (!entry.getValue().sameTopic(consumerGroupId, topicName)
+ || preservedRegionIds.contains(entry.getValue().regionIdStr)) {
continue;
}
commitStates.remove(entry.getKey());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
index 474deb3c0cf..1ae30b123e1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
@@ -172,7 +172,7 @@ public class ConsensusSubscriptionSetupHandler {
* existing subscription metadata to find all active consensus subscriptions
and binds prefetching
* queues to the new region.
*/
- private static void onNewRegionCreated(
+ static void onNewRegionCreated(
final ConsensusGroupId groupId, final IoTConsensusServerImpl serverImpl)
{
if (!(groupId instanceof DataRegionId)) {
return;
@@ -195,111 +195,129 @@ public class ConsensusSubscriptionSetupHandler {
allSubscriptions.size(),
serverImpl.getSearchIndex());
- for (final Map.Entry<String, java.util.Set<String>> groupEntry :
allSubscriptions.entrySet()) {
- final String consumerGroupId = groupEntry.getKey();
- final boolean isTableModel =
SubscriptionAgent.consumer().isTableModel(consumerGroupId);
- for (final String topicName : groupEntry.getValue()) {
- if (!isConsensusBasedTopic(topicName, isTableModel)) {
+ for (final String consumerGroupId : allSubscriptions.keySet()) {
+ // Serialize binding with metadata repair and its rollback. Re-read the
group's topics after
+ // acquiring the lock so an unsubscribe while this callback was waiting
cannot be undone.
+ SubscriptionAgent.consumer()
+ .executeWithConsumerGroupLock(
+ consumerGroupId,
+ () ->
+ bindNewRegionForConsumerGroup(
+ consumerGroupId, groupId, serverImpl, commitManager));
+ }
+ }
+
+ private static void bindNewRegionForConsumerGroup(
+ final String consumerGroupId,
+ final ConsensusGroupId groupId,
+ final IoTConsensusServerImpl serverImpl,
+ final ConsensusSubscriptionCommitManager commitManager) {
+ final boolean isTableModel =
SubscriptionAgent.consumer().isTableModel(consumerGroupId);
+ for (final String topicName :
+
SubscriptionAgent.consumer().getTopicNamesSubscribedByConsumerGroup(consumerGroupId))
{
+ if (!isConsensusBasedTopic(topicName, isTableModel)) {
+ continue;
+ }
+ if (SubscriptionAgent.broker()
+ .hasConsensusQueueForRegion(consumerGroupId, topicName,
groupId.toString())) {
+ continue;
+ }
+ try {
+ final Map<String, TopicConfig> topicConfigs =
+ SubscriptionAgent.topic()
+ .getTopicConfigs(java.util.Collections.singleton(topicName),
isTableModel);
+ final TopicConfig topicConfig = topicConfigs.get(topicName);
+ if (topicConfig == null) {
continue;
}
- try {
- final Map<String, TopicConfig> topicConfigs =
- SubscriptionAgent.topic()
- .getTopicConfigs(java.util.Collections.singleton(topicName),
isTableModel);
- final TopicConfig topicConfig = topicConfigs.get(topicName);
- if (topicConfig == null) {
- continue;
- }
- // Resolve the new DataRegion's actual database name
- final DataRegion dataRegion =
- StorageEngine.getInstance().getDataRegion((DataRegionId)
groupId);
- if (dataRegion == null) {
- continue;
- }
- final String dbRaw = dataRegion.getDatabaseName();
- final String dbTableModel = dbRaw.startsWith("root.") ?
dbRaw.substring(5) : dbRaw;
+ // Resolve the new DataRegion's actual database name
+ final DataRegion dataRegion =
+ StorageEngine.getInstance().getDataRegion((DataRegionId) groupId);
+ if (dataRegion == null) {
+ continue;
+ }
+ final String dbRaw = dataRegion.getDatabaseName();
+ final String dbTableModel = dbRaw.startsWith("root.") ?
dbRaw.substring(5) : dbRaw;
- if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
- continue;
- }
+ if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
+ continue;
+ }
- final String actualDbName = topicConfig.isTableTopic() ?
dbTableModel : null;
- final ConsensusLogToTabletConverter converter =
- buildConverter(topicName, topicConfig, actualDbName);
- final SubscriptionWalRetentionPolicy retentionPolicy =
- buildSubscriptionWalRetentionPolicy(topicName, topicConfig,
serverImpl);
-
- // Recover from persisted per-writer region progress when available.
The queue will
- // resolve a replay start from that progress on first poll via the
region-level locator.
- // A newly-created region starts from its first consensus index and
never proposes its
- // current local tail as an initial committed frontier.
- commitManager.initializeStateWithoutTailProposal(consumerGroupId,
topicName, groupId);
- final RegionProgress committedRegionProgress =
- resolveFallbackCommittedRegionProgress(
- commitManager, consumerGroupId, topicName, groupId);
- final boolean hasLocalPersistedState =
- commitManager.hasPersistedState(consumerGroupId, topicName,
groupId);
- // This region is created after the subscription already exists. Its
first WAL entry may
- // already be visible by the time this callback runs, so start from
the beginning of this
- // new region instead of tail+1 to avoid skipping the first write.
- final long tailStartSearchIndex = FIRST_CONSENSUS_SEARCH_INDEX;
- final long initialRuntimeVersion =
-
regionRuntimeVersion.getOrDefault(groupId.convertToTConsensusGroupId(), 0L);
- final boolean initialActive =
-
lastKnownPreferredWriter.getOrDefault(groupId.convertToTConsensusGroupId(), -1)
- == IOTDB_CONFIG.getDataNodeId();
- final Set<Integer> initialActiveWriterNodeIds =
- regionActiveWriterNodeIds.getOrDefault(
- groupId.convertToTConsensusGroupId(),
- initialActive
- ? Collections.singleton(IOTDB_CONFIG.getDataNodeId())
- : Collections.emptySet());
- final ConsensusRegionRuntimeState initialRuntimeState =
- new ConsensusRegionRuntimeState(
- initialRuntimeVersion,
-
lastKnownPreferredWriter.getOrDefault(groupId.convertToTConsensusGroupId(), -1),
- initialActive,
- initialActiveWriterNodeIds);
+ final String actualDbName = topicConfig.isTableTopic() ? dbTableModel
: null;
+ final ConsensusLogToTabletConverter converter =
+ buildConverter(topicName, topicConfig, actualDbName);
+ final SubscriptionWalRetentionPolicy retentionPolicy =
+ buildSubscriptionWalRetentionPolicy(topicName, topicConfig,
serverImpl);
+
+ // Recover from persisted per-writer region progress when available.
The queue will
+ // resolve a replay start from that progress on first poll via the
region-level locator.
+ // A newly-created region starts from its first consensus index and
never proposes its
+ // current local tail as an initial committed frontier.
+ commitManager.initializeStateWithoutTailProposal(consumerGroupId,
topicName, groupId);
+ final RegionProgress committedRegionProgress =
+ resolveFallbackCommittedRegionProgress(
+ commitManager, consumerGroupId, topicName, groupId);
+ final boolean hasLocalPersistedState =
+ commitManager.hasPersistedState(consumerGroupId, topicName,
groupId);
+ // This region is created after the subscription already exists. Its
first WAL entry may
+ // already be visible by the time this callback runs, so start from
the beginning of this
+ // new region instead of tail+1 to avoid skipping the first write.
+ final long tailStartSearchIndex = FIRST_CONSENSUS_SEARCH_INDEX;
+ final long initialRuntimeVersion =
+
regionRuntimeVersion.getOrDefault(groupId.convertToTConsensusGroupId(), 0L);
+ final boolean initialActive =
+
lastKnownPreferredWriter.getOrDefault(groupId.convertToTConsensusGroupId(), -1)
+ == IOTDB_CONFIG.getDataNodeId();
+ final Set<Integer> initialActiveWriterNodeIds =
+ regionActiveWriterNodeIds.getOrDefault(
+ groupId.convertToTConsensusGroupId(),
+ initialActive
+ ? Collections.singleton(IOTDB_CONFIG.getDataNodeId())
+ : Collections.emptySet());
+ final ConsensusRegionRuntimeState initialRuntimeState =
+ new ConsensusRegionRuntimeState(
+ initialRuntimeVersion,
+
lastKnownPreferredWriter.getOrDefault(groupId.convertToTConsensusGroupId(), -1),
+ initialActive,
+ initialActiveWriterNodeIds);
- LOGGER.info(
- DataNodePipeMessages
-
.PIPE_LOG_AUTO_BINDING_CONSENSUS_QUEUE_FOR_TOPIC_IN_GROUP_TO_NEW_REGION_86F21649,
- topicName,
- consumerGroupId,
- groupId,
- dbTableModel,
- tailStartSearchIndex,
- hasLocalPersistedState,
- committedRegionProgress,
- initialRuntimeVersion,
- initialActive);
+ LOGGER.info(
+ DataNodePipeMessages
+
.PIPE_LOG_AUTO_BINDING_CONSENSUS_QUEUE_FOR_TOPIC_IN_GROUP_TO_NEW_REGION_86F21649,
+ topicName,
+ consumerGroupId,
+ groupId,
+ dbTableModel,
+ tailStartSearchIndex,
+ hasLocalPersistedState,
+ committedRegionProgress,
+ initialRuntimeVersion,
+ initialActive);
- SubscriptionAgent.broker()
- .bindConsensusPrefetchingQueue(
- consumerGroupId,
- topicName,
- topicConfig.getOrderMode(),
- groupId,
- serverImpl,
- retentionPolicy,
- converter,
- commitManager,
- committedRegionProgress,
- tailStartSearchIndex,
- initialRuntimeVersion,
- initialActive,
- hasLocalPersistedState);
- SubscriptionAgent.broker().applyRuntimeStateForRegion(groupId,
initialRuntimeState);
- } catch (final Exception e) {
- LOGGER.error(
- DataNodePipeMessages
-
.PIPE_LOG_FAILED_TO_AUTO_BIND_TOPIC_IN_GROUP_TO_NEW_REGION_5BFD0E7D,
- topicName,
- consumerGroupId,
- groupId,
- e);
- }
+ SubscriptionAgent.broker()
+ .bindConsensusPrefetchingQueue(
+ consumerGroupId,
+ topicName,
+ topicConfig.getOrderMode(),
+ groupId,
+ serverImpl,
+ retentionPolicy,
+ converter,
+ commitManager,
+ committedRegionProgress,
+ tailStartSearchIndex,
+ initialRuntimeVersion,
+ initialActive,
+ hasLocalPersistedState);
+ SubscriptionAgent.broker().applyRuntimeStateForRegion(groupId,
initialRuntimeState);
+ } catch (final Exception e) {
+ LOGGER.error(
+
DataNodePipeMessages.PIPE_LOG_FAILED_TO_AUTO_BIND_TOPIC_IN_GROUP_TO_NEW_REGION_5BFD0E7D,
+ topicName,
+ consumerGroupId,
+ groupId,
+ e);
}
}
}
@@ -391,6 +409,14 @@ public class ConsensusSubscriptionSetupHandler {
public static void setupConsensusSubscriptions(
final String consumerGroupId, final Set<String> topicNames, final
boolean isTableModel) {
+ SubscriptionAgent.consumer()
+ .executeWithConsumerGroupLock(
+ consumerGroupId,
+ () -> setupConsensusSubscriptionsInternal(consumerGroupId,
topicNames, isTableModel));
+ }
+
+ private static void setupConsensusSubscriptionsInternal(
+ final String consumerGroupId, final Set<String> topicNames, final
boolean isTableModel) {
final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
if (!(dataRegionConsensus instanceof IoTConsensus)) {
final String configuredProtocol =
IOTDB_CONFIG.getDataRegionConsensusProtocolClass();
@@ -420,6 +446,10 @@ public class ConsensusSubscriptionSetupHandler {
topicNames,
ioTConsensus.getAllConsensusGroupIds().size());
+ // Recovery also visits topics with live queues. Roll back only newly
bound regions so a
+ // failure on another topic cannot discard their in-flight events or
committed progress.
+ final Map<String, Set<String>> existingQueueRegionIds =
+ SubscriptionAgent.broker().getConsensusQueueRegionIds(consumerGroupId,
topicNames);
final ConsensusSubscriptionCommitManager.SetupSnapshot setupSnapshot =
commitManager.captureSetupSnapshot(consumerGroupId, topicNames);
setupConsensusTopics(
@@ -431,19 +461,27 @@ public class ConsensusSubscriptionSetupHandler {
consumerGroupId, topicName, isTableModel, ioTConsensus,
commitManager),
attemptedTopicNames ->
rollbackConsensusSubscriptionSetup(
- consumerGroupId, attemptedTopicNames, commitManager,
setupSnapshot));
+ consumerGroupId,
+ attemptedTopicNames,
+ commitManager,
+ setupSnapshot,
+ existingQueueRegionIds));
}
private static void rollbackConsensusSubscriptionSetup(
final String consumerGroupId,
final Set<String> attemptedTopicNames,
final ConsensusSubscriptionCommitManager commitManager,
- final ConsensusSubscriptionCommitManager.SetupSnapshot setupSnapshot) {
+ final ConsensusSubscriptionCommitManager.SetupSnapshot setupSnapshot,
+ final Map<String, Set<String>> existingQueueRegionIds) {
RuntimeException rollbackFailure = null;
for (final String topicName : attemptedTopicNames) {
try {
SubscriptionAgent.broker()
- .unbindConsensusPrefetchingQueue(consumerGroupId, topicName,
false);
+ .rollbackConsensusPrefetchingQueues(
+ consumerGroupId,
+ topicName,
+ existingQueueRegionIds.getOrDefault(topicName,
Collections.emptySet()));
} catch (final RuntimeException e) {
if (Objects.isNull(rollbackFailure)) {
rollbackFailure = e;
@@ -453,9 +491,14 @@ public class ConsensusSubscriptionSetupHandler {
}
}
try {
- commitManager.restoreSetupSnapshot(setupSnapshot, attemptedTopicNames);
+ commitManager.restoreSetupSnapshot(
+ setupSnapshot, attemptedTopicNames, existingQueueRegionIds);
restoreStartupDetachedRetentions(
- consumerGroupId, attemptedTopicNames, setupSnapshot, commitManager);
+ consumerGroupId,
+ attemptedTopicNames,
+ setupSnapshot,
+ commitManager,
+ existingQueueRegionIds);
} catch (final RuntimeException e) {
if (Objects.isNull(rollbackFailure)) {
rollbackFailure = e;
@@ -472,7 +515,8 @@ public class ConsensusSubscriptionSetupHandler {
final String consumerGroupId,
final Set<String> attemptedTopicNames,
final ConsensusSubscriptionCommitManager.SetupSnapshot setupSnapshot,
- final ConsensusSubscriptionCommitManager commitManager) {
+ final ConsensusSubscriptionCommitManager commitManager,
+ final Map<String, Set<String>> existingQueueRegionIds) {
final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
if (!(dataRegionConsensus instanceof IoTConsensus)) {
return;
@@ -480,6 +524,11 @@ public class ConsensusSubscriptionSetupHandler {
final IoTConsensus ioTConsensus = (IoTConsensus) dataRegionConsensus;
for (final String topicName : attemptedTopicNames) {
for (final String regionId : setupSnapshot.getRegionIds(topicName)) {
+ if (existingQueueRegionIds
+ .getOrDefault(topicName, Collections.emptySet())
+ .contains(regionId)) {
+ continue;
+ }
final ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromString(regionId);
final IoTConsensusServerImpl serverImpl =
ioTConsensus.getImpl(groupId);
if (Objects.nonNull(serverImpl)) {
@@ -554,7 +603,7 @@ public class ConsensusSubscriptionSetupHandler {
* DATABASE_KEY} filter. Additionally, the {@link #onNewRegionCreated}
callback ensures that
* regions created after this method runs are also automatically bound.
*/
- private static void setupConsensusQueueForTopic(
+ static void setupConsensusQueueForTopic(
final String consumerGroupId,
final String topicName,
final boolean isTableModel,
@@ -598,6 +647,23 @@ public class ConsensusSubscriptionSetupHandler {
if (!(groupId instanceof DataRegionId)) {
continue;
}
+ if (ioTConsensus.isPeerCreationCallbackPending(groupId)
+ && SubscriptionAgent.consumer()
+ .isTopicSubscribedByConsumerGroup(consumerGroupId, topicName)) {
+ // The new-region callback binds this peer from its first WAL index.
Binding it here from
+ // the current tail could skip writes replicated before the callback
finishes.
+ bound = true;
+ continue;
+ }
+ if (SubscriptionAgent.broker()
+ .hasConsensusQueueForRegion(consumerGroupId, topicName,
groupId.toString())) {
+ // In particular, a queue created by onNewRegionCreated must keep its
initialization from
+ // the beginning of the region instead of being reinitialized from a
local tail proposal.
+ // Check the live broker as well because a region callback can finish
after the setup
+ // snapshot was captured.
+ bound = true;
+ continue;
+ }
final IoTConsensusServerImpl serverImpl = ioTConsensus.getImpl(groupId);
if (serverImpl == null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index cd53d9376ab..93d309cc5db 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -454,6 +454,12 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
consumerConfig);
}
+ // A previous inactivity cleanup or metadata race may have removed the
local broker while the
+ // coordinator still has a non-empty consumer-group metadata record.
Reconcile that state before
+ // exposing the consumer to subsequent requests. The operation is
idempotent and serialized with
+ // metadata pushes by SubscriptionConsumerAgent.
+ SubscriptionAgent.consumer().ensureConsumerGroupBroker(consumerGroupId);
+
activateConsumer(consumerConfig);
final int dataNodeId =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java
new file mode 100644
index 00000000000..d281c85d5cc
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java
@@ -0,0 +1,279 @@
+/*
+ * 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.db.subscription.agent;
+
+import
org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.consensus.IConsensus;
+import org.apache.iotdb.consensus.iot.IoTConsensus;
+import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
+import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
+import org.apache.iotdb.db.subscription.broker.ISubscriptionBroker;
+import org.apache.iotdb.db.subscription.broker.SubscriptionBroker;
+import org.apache.iotdb.db.subscription.task.subtask.SubscriptionSinkSubtask;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class SubscriptionConsumerAgentBrokerRecoveryTest {
+
+ private static final long TEST_TIMEOUT_SECONDS = 5L;
+
+ @Test
+ public void testNonEmptyMetadataRecreatesMissingBroker() {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ try {
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+ assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+
+ assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId));
+ assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+
+ agent.ensureConsumerGroupBroker(consumerGroupId);
+
+ assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+ assertTrue(agent.isConsumerExisted(consumerGroupId, consumerId));
+ } finally {
+ agent.handleDropConsumerGroup(consumerGroupId);
+ }
+ }
+
+ @Test
+ public void testMetadataPushSetsUpNewConsensusTopicOnce() {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final String topicName = unique("topic");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ final IConsensus originalConsensus = DataRegionConsensusImpl.getInstance();
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.emptyList());
+ DataRegionConsensusImpl.setInstance(consensus);
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+ clearInvocations(consensus);
+
+ final ConsumerGroupMeta subscribedMeta = meta.deepCopy();
+ subscribedMeta.addSubscription(consumerId,
Collections.singleton(topicName));
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(subscribedMeta));
+
+ // One discovery for setup logging and one for binding the topic's
regions.
+ verify(consensus, times(2)).getAllConsensusGroupIds();
+ assertEquals(
+ Collections.singleton(topicName),
+ agent.getTopicNamesSubscribedByConsumer(consumerGroupId,
consumerId));
+ } finally {
+ agent.handleDropConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ DataRegionConsensusImpl.setInstance(originalConsensus);
+ }
+ }
+
+ @Test
+ public void testEmptyMetadataDoesNotRecreateBroker() {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ try {
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+ assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+
+ final ConsumerGroupMeta emptyMeta = meta.deepCopy();
+ emptyMeta.removeConsumer(consumerId);
+ assertTrue(emptyMeta.isEmpty());
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(emptyMeta));
+
+ assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+ agent.ensureConsumerGroupBroker(consumerGroupId);
+ assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+ } finally {
+ agent.handleDropConsumerGroup(consumerGroupId);
+ }
+ }
+
+ @Test
+ public void testConcurrentMetadataPushAndBrokerEnsureCreateOneBroker()
throws Exception {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ final ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+ assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId));
+
+ final Future<?> metadataPush =
+ executor.submit(() ->
agent.handleSingleConsumerGroupMetaChanges(meta.deepCopy()));
+ final Future<?> handshakeRepair =
+ executor.submit(() ->
agent.ensureConsumerGroupBroker(consumerGroupId));
+ metadataPush.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ handshakeRepair.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+ assertEquals(1, getBrokers(consumerGroupId).size());
+ } finally {
+ executor.shutdownNow();
+ executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ agent.handleDropConsumerGroup(consumerGroupId);
+ }
+ }
+
+ @Test
+ public void testBrokerAndExistingQueueSurviveProviderReplacementRepair()
throws Exception {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final String topicName = unique("topic");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(new TopicMeta(topicName, 1L,
Collections.emptyMap())));
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ meta.addSubscription(consumerId, Collections.singleton(topicName));
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+
+ bindPipeQueue(consumerGroupId, topicName);
+ final SubscriptionBroker brokerBeforeRepair =
getPipeBroker(consumerGroupId);
+ assertTrue(brokerBeforeRepair.hasQueue(topicName));
+
+ // A provider replacement closes only its session. The server-side
repair must preserve the
+ // existing broker and queue when it reconciles the same non-empty
metadata.
+ agent.ensureConsumerGroupBroker(consumerGroupId);
+
+ assertSame(brokerBeforeRepair, getPipeBroker(consumerGroupId));
+ assertTrue(getPipeBroker(consumerGroupId).hasQueue(topicName));
+ } finally {
+ agent.handleDropConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ }
+ }
+
+ @Test
+ public void testCanceledTopicIsNotRestoredWhenBrokerIsRecreated() throws
Exception {
+ final String consumerGroupId = unique("group");
+ final String consumerId = unique("consumer");
+ final String topicName = unique("topic");
+ final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer();
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(new TopicMeta(topicName, 1L,
Collections.emptyMap())));
+ final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId);
+ meta.addSubscription(consumerId, Collections.singleton(topicName));
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(meta));
+
+ bindPipeQueue(consumerGroupId, topicName);
+ assertTrue(getPipeBroker(consumerGroupId).hasQueue(topicName));
+
+ final ConsumerGroupMeta unsubscribedMeta = meta.deepCopy();
+ unsubscribedMeta.removeSubscription(consumerId,
Collections.singleton(topicName));
+ assertNull(agent.handleSingleConsumerGroupMetaChanges(unsubscribedMeta));
+ assertFalse(getPipeBroker(consumerGroupId).hasQueue(topicName));
+
+ assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId));
+ agent.ensureConsumerGroupBroker(consumerGroupId);
+
+ assertFalse(getPipeBroker(consumerGroupId).hasQueue(topicName));
+ assertFalse(agent.getAllSubscriptions().containsKey(consumerGroupId));
+ } finally {
+ agent.handleDropConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ }
+ }
+
+ private static ConsumerGroupMeta createMeta(
+ final String consumerGroupId, final String consumerId) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(ConsumerConstant.CONSUMER_ID_KEY, consumerId);
+ attributes.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, consumerGroupId);
+ final ConsumerConfig consumerConfig = new ConsumerConfig(attributes);
+ return new ConsumerGroupMeta(
+ consumerGroupId, 1L, new ConsumerMeta(consumerId, 1L,
consumerConfig.getAttribute()));
+ }
+
+ private static void bindPipeQueue(final String consumerGroupId, final String
topicName) {
+ final UnboundedBlockingPendingQueue<Event> inputQueue =
+ new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter());
+ final SubscriptionSinkSubtask subtask =
+ new SubscriptionSinkSubtask(
+ unique("task"), 1L, "", 0, inputQueue, null, topicName,
consumerGroupId);
+ SubscriptionAgent.broker().bindPrefetchingQueue(subtask);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<ISubscriptionBroker> getBrokers(final String
consumerGroupId)
+ throws Exception {
+ final Field field =
SubscriptionBrokerAgent.class.getDeclaredField("consumerGroupIdToBrokers");
+ field.setAccessible(true);
+ final Map<String, List<ISubscriptionBroker>> brokersByConsumerGroup =
+ (Map<String, List<ISubscriptionBroker>>)
field.get(SubscriptionAgent.broker());
+ return brokersByConsumerGroup.getOrDefault(consumerGroupId,
Collections.emptyList());
+ }
+
+ private static SubscriptionBroker getPipeBroker(final String
consumerGroupId) throws Exception {
+ return getBrokers(consumerGroupId).stream()
+ .filter(SubscriptionBroker.class::isInstance)
+ .map(SubscriptionBroker.class::cast)
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ }
+
+ private static String unique(final String prefix) {
+ return prefix + '-' + UUID.randomUUID();
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
index 9401fc3e4af..19f944b52d0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
@@ -21,12 +21,26 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.consensus.IConsensus;
+import org.apache.iotdb.consensus.config.IoTConsensusConfig;
+import org.apache.iotdb.consensus.iot.IoTConsensus;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
+import org.apache.iotdb.db.storageengine.StorageEngine;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.db.subscription.agent.SubscriptionBrokerAgent;
+import org.apache.iotdb.db.subscription.broker.ConsensusSubscriptionBroker;
+import org.apache.iotdb.db.subscription.broker.ISubscriptionBroker;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
@@ -35,25 +49,39 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
+import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.LongSupplier;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class ConsensusSubscriptionSetupHandlerTest {
@@ -130,6 +158,343 @@ public class ConsensusSubscriptionSetupHandlerTest {
assertEquals(attemptedTopicNames, rolledBackTopicNames);
}
+ @Test
+ public void
testRepairFailurePreservesExistingQueuesAndConcurrentCommitProgress()
+ throws Exception {
+ final IConsensus originalConsensus = DataRegionConsensusImpl.getInstance();
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String topicName = "topic-" + UUID.randomUUID();
+ final DataRegionId existingRegion = new DataRegionId(51);
+ final DataRegionId addedRegion = new DataRegionId(52);
+ final WriterId writerId = new WriterId(existingRegion.toString(), 7);
+ final ConsensusSubscriptionCommitManager commitManager =
+ ConsensusSubscriptionCommitManager.getInstance();
+ final ConsensusSubscriptionBroker broker = new
ConsensusSubscriptionBroker(consumerGroupId);
+ final ConsensusPrefetchingQueue existingQueue =
mock(ConsensusPrefetchingQueue.class);
+ final ConsensusPrefetchingQueue addedQueue =
mock(ConsensusPrefetchingQueue.class);
+ when(existingQueue.getConsensusGroupId()).thenReturn(existingRegion);
+ when(addedQueue.getConsensusGroupId()).thenReturn(addedRegion);
+ final List<ConsensusPrefetchingQueue> queues =
+ new CopyOnWriteArrayList<>(Collections.singletonList(existingQueue));
+ getConsensusQueues(broker).put(topicName, queues);
+ getBrokers()
+ .put(consumerGroupId, new
CopyOnWriteArrayList<>(Collections.singletonList(broker)));
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+ commitManager.receiveProgressBroadcast(
+ consumerGroupId,
+ topicName,
+ existingRegion.toString(),
+ writerId,
+ new WriterProgress(100L, 1L));
+ final
ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState
existingState =
+ commitManager.getOrCreateState(consumerGroupId, topicName,
existingRegion);
+
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+ final AtomicInteger discoveryCount = new AtomicInteger();
+ when(consensus.getAllConsensusGroupIds())
+ .thenAnswer(
+ ignored -> {
+ if (discoveryCount.incrementAndGet() == 2) {
+ // These changes happen after setup captured its rollback
snapshot.
+ queues.add(addedQueue);
+ commitManager.receiveProgressBroadcast(
+ consumerGroupId,
+ topicName,
+ existingRegion.toString(),
+ writerId,
+ new WriterProgress(200L, 2L));
+ commitManager.receiveProgressBroadcast(
+ consumerGroupId,
+ topicName,
+ addedRegion.toString(),
+ new WriterId(addedRegion.toString(), 7),
+ new WriterProgress(100L, 1L));
+ }
+ return Collections.emptyList();
+ });
+ DataRegionConsensusImpl.setInstance(consensus);
+
+ try {
+ ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
+ consumerGroupId,
+ new LinkedHashSet<>(Arrays.asList(topicName, "missing-" +
UUID.randomUUID())),
+ false);
+ fail();
+ } catch (final SubscriptionException expected) {
+ assertTrue(expected.getMessage().contains("missing-"));
+ }
+
+ assertEquals(
+ Collections.singleton(existingRegion.toString()),
broker.getQueueRegionIds(topicName));
+ assertEquals(Collections.singletonList(existingQueue), queues);
+ verify(existingQueue, never()).closeAsync(false, false);
+ verify(addedQueue).closeAsync(false, false);
+ assertSame(
+ existingState,
+ commitManager.getOrCreateState(consumerGroupId, topicName,
existingRegion));
+ assertEquals(
+ new WriterProgress(200L, 2L),
+ commitManager
+ .getCommittedRegionProgress(consumerGroupId, topicName,
existingRegion)
+ .getWriterPositions()
+ .get(writerId));
+ assertFalse(commitManager.hasPersistedState(consumerGroupId, topicName,
addedRegion));
+ } finally {
+ getBrokers().remove(consumerGroupId);
+ commitManager.removeAllStatesForConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ DataRegionConsensusImpl.setInstance(originalConsensus);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, List<ConsensusPrefetchingQueue>>
getConsensusQueues(
+ final ConsensusSubscriptionBroker broker) throws Exception {
+ final Field field =
+
ConsensusSubscriptionBroker.class.getDeclaredField("topicNameToConsensusPrefetchingQueues");
+ field.setAccessible(true);
+ return (Map<String, List<ConsensusPrefetchingQueue>>) field.get(broker);
+ }
+
+ @Test
+ public void testRepairPreservesInitializationOfExistingNewRegionQueue()
throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String topicName = "topic-" + UUID.randomUUID();
+ final DataRegionId regionId = new DataRegionId(53);
+ final DataRegion region = mock(DataRegion.class);
+ when(region.getDatabaseName()).thenReturn("root.repair");
+ final Field regionField =
StorageEngine.class.getDeclaredField("dataRegionMap");
+ regionField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ final Map<DataRegionId, DataRegion> regions =
+ (Map<DataRegionId, DataRegion>)
regionField.get(StorageEngine.getInstance());
+ final DataRegion previousRegion = regions.put(regionId, region);
+ try {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+
.setSystemDir(temporaryFolder.newFolder("existingNewRegion").getAbsolutePath());
+ final ConsensusSubscriptionCommitManager manager =
+ new ConsensusSubscriptionCommitManager(
+ (group, topic, id) ->
+
ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent());
+ manager.initializeStateWithoutTailProposal(consumerGroupId, topicName,
regionId);
+ assertFalse(manager.isInitialProgressPending(consumerGroupId, topicName,
regionId));
+ assertTrue(manager.isConfigNodeProgressAvailable(consumerGroupId,
topicName, regionId));
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+
+ final ConsensusPrefetchingQueue queue =
mock(ConsensusPrefetchingQueue.class);
+ when(queue.getConsensusGroupId()).thenReturn(regionId);
+ when(queue.computeTailRegionProgress())
+ .thenReturn(new RegionProgress(Collections.emptyMap()));
+ final ConsensusSubscriptionBroker broker = new
ConsensusSubscriptionBroker(consumerGroupId);
+ getConsensusQueues(broker)
+ .put(topicName, new
CopyOnWriteArrayList<>(Collections.singletonList(queue)));
+ getBrokers()
+ .put(consumerGroupId, new
CopyOnWriteArrayList<>(Collections.singletonList(broker)));
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+ final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
+
when(server.getConfig()).thenReturn(IoTConsensusConfig.newBuilder().build());
+ when(consensus.getImpl(regionId)).thenReturn(server);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.singletonList(regionId));
+
+ ConsensusSubscriptionSetupHandler.setupConsensusQueueForTopic(
+ consumerGroupId, topicName, false, consensus, manager);
+
+ assertFalse(manager.isInitialProgressPending(consumerGroupId, topicName,
regionId));
+ assertTrue(manager.isConfigNodeProgressAvailable(consumerGroupId,
topicName, regionId));
+ verify(queue, never()).computeTailRegionProgress();
+ } finally {
+ if (previousRegion == null) {
+ regions.remove(regionId);
+ } else {
+ regions.put(regionId, previousRegion);
+ }
+ getBrokers().remove(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testRepairSkipsRegionWhileNewPeerCallbackIsPending() throws
Exception {
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String topicName = "topic-" + UUID.randomUUID();
+ final DataRegionId regionId = new DataRegionId(54);
+ final IConsensus originalConsensus = DataRegionConsensusImpl.getInstance();
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.emptyList());
+ DataRegionConsensusImpl.setInstance(consensus);
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+ createSubscribedConsumerGroup(consumerGroupId, topicName);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.singletonList(regionId));
+ when(consensus.isPeerCreationCallbackPending(regionId)).thenReturn(true);
+
+ ConsensusSubscriptionSetupHandler.setupConsensusQueueForTopic(
+ consumerGroupId,
+ topicName,
+ false,
+ consensus,
+ ConsensusSubscriptionCommitManager.getInstance());
+
+ verify(consensus, never()).getImpl(regionId);
+ assertTrue(
+ getBrokers().getOrDefault(consumerGroupId,
Collections.emptyList()).stream()
+ .noneMatch(ConsensusSubscriptionBroker.class::isInstance));
+ } finally {
+ SubscriptionAgent.consumer().handleDropConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ DataRegionConsensusImpl.setInstance(originalConsensus);
+ }
+ }
+
+ @Test
+ public void testNewSubscriptionDoesNotWaitForAnUnrelatedPeerCallback()
throws Exception {
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String topicName = "topic-" + UUID.randomUUID();
+ final DataRegionId regionId = new DataRegionId(55);
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.singletonList(regionId));
+ when(consensus.isPeerCreationCallbackPending(regionId)).thenReturn(true);
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+
+ ConsensusSubscriptionSetupHandler.setupConsensusQueueForTopic(
+ consumerGroupId,
+ topicName,
+ false,
+ consensus,
+ ConsensusSubscriptionCommitManager.getInstance());
+
+ verify(consensus).getImpl(regionId);
+ } finally {
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ }
+ }
+
+ @Test
+ public void testNewRegionCallbackWaitsForRepairAndRereadsSubscriptions()
throws Exception {
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String topicName = "topic-" + UUID.randomUUID();
+ final DataRegionId regionId = new DataRegionId(56);
+ final IConsensus originalConsensus = DataRegionConsensusImpl.getInstance();
+ final IoTConsensus consensus = mock(IoTConsensus.class);
+
when(consensus.getAllConsensusGroupIds()).thenReturn(Collections.emptyList());
+ DataRegionConsensusImpl.setInstance(consensus);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch callbackStarted = new CountDownLatch(1);
+ final CountDownLatch callbackFinished = new CountDownLatch(1);
+ final AtomicReference<Future<?>> callback = new AtomicReference<>();
+ final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
+ when(server.getSearchIndex())
+ .thenAnswer(
+ ignored -> {
+ callbackStarted.countDown();
+ return 1L;
+ });
+ try {
+ assertNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
+ final ConsumerGroupMeta meta =
createSubscribedConsumerGroup(consumerGroupId, topicName);
+
+ SubscriptionAgent.consumer()
+ .executeWithConsumerGroupLock(
+ consumerGroupId,
+ () -> {
+ callback.set(
+ executor.submit(
+ () -> {
+ try {
+
ConsensusSubscriptionSetupHandler.onNewRegionCreated(regionId, server);
+ } finally {
+ callbackFinished.countDown();
+ }
+ }));
+ try {
+ assertTrue(callbackStarted.await(5, TimeUnit.SECONDS));
+ assertFalse(callbackFinished.await(200,
TimeUnit.MILLISECONDS));
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ final ConsumerGroupMeta unsubscribedMeta = meta.deepCopy();
+ unsubscribedMeta.removeSubscription(null,
Collections.singleton(topicName));
+ assertNull(
+ SubscriptionAgent.consumer()
+
.handleSingleConsumerGroupMetaChanges(unsubscribedMeta));
+ });
+
+ callback.get().get(5, TimeUnit.SECONDS);
+ assertTrue(
+ SubscriptionAgent.consumer()
+ .getTopicNamesSubscribedByConsumerGroup(consumerGroupId)
+ .isEmpty());
+ assertTrue(
+ getBrokers().getOrDefault(consumerGroupId,
Collections.emptyList()).stream()
+ .noneMatch(ConsensusSubscriptionBroker.class::isInstance));
+ } finally {
+ executor.shutdownNow();
+ executor.awaitTermination(5, TimeUnit.SECONDS);
+ SubscriptionAgent.consumer().handleDropConsumerGroup(consumerGroupId);
+ SubscriptionAgent.topic().handleDropTopic(topicName, false);
+ DataRegionConsensusImpl.setInstance(originalConsensus);
+ }
+ }
+
+ private static ConsumerGroupMeta createSubscribedConsumerGroup(
+ final String consumerGroupId, final String topicName) {
+ final String consumerId = "consumer-" + UUID.randomUUID();
+ final ConsumerGroupMeta meta =
+ new ConsumerGroupMeta(
+ consumerGroupId, 1L, new ConsumerMeta(consumerId, 1L,
Collections.emptyMap()));
+ meta.addSubscription(consumerId, Collections.singleton(topicName));
+
assertNull(SubscriptionAgent.consumer().handleSingleConsumerGroupMetaChanges(meta));
+ return meta;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<String, List<ISubscriptionBroker>> getBrokers() throws
Exception {
+ final Field field =
SubscriptionBrokerAgent.class.getDeclaredField("consumerGroupIdToBrokers");
+ field.setAccessible(true);
+ return (Map<String, List<ISubscriptionBroker>>)
field.get(SubscriptionAgent.broker());
+ }
+
@Test
public void
testFallbackLookupDoesNotQueryConfigNodeWithoutLocalPersistence() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
index 40482fe1b20..05eccb978f9 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
@@ -221,6 +221,39 @@ public class SubscriptionReceiverV1Test {
Assert.assertTrue((boolean) getField(receiver, "consumerFenced"));
}
+ @Test
+ public void
testHandshakeRepairsMissingBrokerWithoutChangingConsumerMetadata() throws
Exception {
+ final String consumerGroupId = "group-" + UUID.randomUUID();
+ final String consumerId = "consumer-" + UUID.randomUUID();
+ final ConsumerConfig consumerConfig =
+ createConsumerConfig(1_000L, null, null, consumerId, consumerGroupId);
+ final ConsumerGroupMeta consumerGroupMeta =
+ new ConsumerGroupMeta(
+ consumerGroupId,
+ System.currentTimeMillis(),
+ new ConsumerMeta(
+ consumerId, System.currentTimeMillis(),
consumerConfig.getAttribute()));
+ final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
+ Assert.assertNull(
+
SubscriptionAgent.consumer().handleSingleConsumerGroupMetaChanges(consumerGroupMeta));
+ try {
+
Assert.assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId));
+
Assert.assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+
+ final TSStatus status =
+ receiver
+
.handle(PipeSubscribeHandshakeReq.toTPipeSubscribeReq(consumerConfig))
+ .getStatus();
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+
Assert.assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId));
+ Assert.assertTrue(
+ SubscriptionAgent.consumer().isConsumerExisted(consumerGroupId,
consumerId));
+ } finally {
+ SubscriptionAgent.consumer().handleDropConsumerGroup(consumerGroupId);
+ }
+ }
+
@Test
public void testNeverHandshakenConsumerStillReturnsMissingConsumerStatus() {
final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
