Skip to content

Commit 44dd90f

Browse files
CaideyipiCaideyipi
authored
[Subscription] Restore consensus subscription brokers after restart (#18916)
1 parent da407ee commit 44dd90f

3 files changed

Lines changed: 373 additions & 5 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionRecoveryTableIT.java‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
21+
22+
import org.apache.iotdb.it.env.EnvFactory;
23+
import org.apache.iotdb.it.env.cluster.config.MppCommonConfig;
24+
import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
25+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
26+
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
27+
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumer;
28+
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
29+
import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
30+
import org.apache.iotdb.subscription.it.consensus.local.AbstractSubscriptionConsensusLocalIT;
31+
32+
import org.apache.tsfile.read.query.dataset.ResultSet;
33+
import org.awaitility.Awaitility;
34+
import org.junit.Assert;
35+
import org.junit.Before;
36+
import org.junit.Test;
37+
import org.junit.experimental.categories.Category;
38+
import org.junit.runner.RunWith;
39+
40+
import java.time.Duration;
41+
import java.util.LinkedHashSet;
42+
import java.util.List;
43+
import java.util.Properties;
44+
import java.util.Set;
45+
46+
@RunWith(IoTDBTestRunner.class)
47+
@Category({LocalStandaloneIT.class})
48+
public class IoTDBConsensusSubscriptionRecoveryTableIT
49+
extends AbstractSubscriptionConsensusLocalIT {
50+
51+
@Override
52+
@Before
53+
public void setUp() throws Exception {
54+
// Isolate subscription recovery from data loss caused by ASYNC WAL during a forced shutdown.
55+
EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
56+
// Persist each acknowledged batch before killing the servers.
57+
final Properties properties = new Properties();
58+
properties.setProperty("subscription_consensus_commit_persist_interval", "1");
59+
properties.setProperty("subscription_consensus_commit_fsync_enabled", "true");
60+
((MppCommonConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig())
61+
.updateProperties(properties);
62+
super.setUp();
63+
}
64+
65+
@Test
66+
public void testRebuiltConsumersResumeAfterFullClusterRestart() throws Exception {
67+
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
68+
ConsensusSubscriptionTableITSupport.newIdentifiers("table_full_cluster_restart");
69+
final String tableName = "t1";
70+
final Set<String> expectedRows = new LinkedHashSet<>();
71+
final Set<String> consumedRows = new LinkedHashSet<>();
72+
SubscriptionTablePullConsumer firstConsumer = null;
73+
SubscriptionTablePullConsumer secondConsumer = null;
74+
75+
try {
76+
ConsensusSubscriptionTableITSupport.bootstrapDatabaseAndTable(
77+
ids.getDatabase(), tableName, ConsensusSubscriptionTableITSupport.DEFAULT_TABLE_SCHEMA);
78+
ConsensusSubscriptionTableITSupport.createConsensusTopic(ids.getTopic(), ".*", ".*");
79+
firstConsumer =
80+
ConsensusSubscriptionTableITSupport.createConsumer(
81+
ids.consumer("first"), ids.getConsumerGroupId());
82+
secondConsumer =
83+
ConsensusSubscriptionTableITSupport.createConsumer(
84+
ids.consumer("second"), ids.getConsumerGroupId());
85+
firstConsumer.subscribe(ids.getTopic());
86+
secondConsumer.subscribe(ids.getTopic());
87+
88+
final Set<String> committedRows =
89+
ConsensusSubscriptionTableITSupport.insertRows(
90+
ids.getDatabase(), tableName, 1_000L, 16, true);
91+
expectedRows.addAll(committedRows);
92+
assertConsumedRows(expectedRows, consumedRows, firstConsumer, secondConsumer);
93+
94+
final Set<String> pendingRows =
95+
ConsensusSubscriptionTableITSupport.insertRows(
96+
ids.getDatabase(), tableName, 2_000L, 64, true);
97+
98+
EnvFactory.getEnv().shutdownForciblyAllDataNodes();
99+
EnvFactory.getEnv().shutdownForciblyAllConfigNodes();
100+
// Close while the servers are down so the old clients cannot remove persisted subscriptions.
101+
firstConsumer.close();
102+
firstConsumer = null;
103+
secondConsumer.close();
104+
secondConsumer = null;
105+
106+
EnvFactory.getEnv().startAllConfigNodes();
107+
EnvFactory.getEnv().startAllDataNodes();
108+
((AbstractEnv) EnvFactory.getEnv()).checkClusterStatusWithoutUnknown();
109+
110+
firstConsumer =
111+
ConsensusSubscriptionTableITSupport.createConsumer(
112+
ids.consumer("first"), ids.getConsumerGroupId());
113+
secondConsumer =
114+
ConsensusSubscriptionTableITSupport.createConsumer(
115+
ids.consumer("second"), ids.getConsumerGroupId());
116+
firstConsumer.subscribe(ids.getTopic());
117+
secondConsumer.subscribe(ids.getTopic());
118+
119+
// Do not seek, recreate the topic, or insert new rows to trigger subscription setup.
120+
// Track unique rows as a receiver would, allowing redelivery around the forced shutdown.
121+
expectedRows.addAll(pendingRows);
122+
assertConsumedRows(expectedRows, consumedRows, firstConsumer, secondConsumer);
123+
124+
final Set<String> newRows =
125+
ConsensusSubscriptionTableITSupport.insertRows(
126+
ids.getDatabase(), tableName, 3_000L, 16, true);
127+
expectedRows.addAll(newRows);
128+
assertConsumedRows(expectedRows, consumedRows, firstConsumer, secondConsumer);
129+
} finally {
130+
ConsensusSubscriptionTableITSupport.cleanup(firstConsumer, ids.getTopic(), ids.getDatabase());
131+
ConsensusSubscriptionTableITSupport.cleanup(
132+
secondConsumer, ids.getTopic(), ids.getDatabase());
133+
}
134+
}
135+
136+
private static void assertConsumedRows(
137+
final Set<String> expectedRows,
138+
final Set<String> consumedRows,
139+
final SubscriptionTablePullConsumer... consumers) {
140+
Awaitility.await()
141+
.pollInSameThread()
142+
.atMost(Duration.ofMinutes(2))
143+
.untilAsserted(
144+
() -> {
145+
for (final SubscriptionTablePullConsumer consumer : consumers) {
146+
final List<SubscriptionMessage> messages = consumer.poll(Duration.ofSeconds(1));
147+
for (final SubscriptionMessage message : messages) {
148+
for (final ResultSet resultSet : message.getResultSets()) {
149+
final SubscriptionRecordHandler.SubscriptionResultSet records =
150+
(SubscriptionRecordHandler.SubscriptionResultSet) resultSet;
151+
while (records.hasNext()) {
152+
consumedRows.add(
153+
ConsensusSubscriptionTableITSupport.rowKey(
154+
records.getDatabaseName(),
155+
records.getTableName(),
156+
records.nextRecord().getTimestamp()));
157+
}
158+
}
159+
}
160+
consumer.commitSync(messages);
161+
}
162+
Assert.assertEquals(expectedRows, consumedRows);
163+
});
164+
}
165+
}

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -111,11 +111,18 @@ public class IoTConsensus implements IConsensus {
111111
private IoTConsensusConfig config;
112112

113113
/**
114-
* Optional callback invoked after a new local peer is created via {@link #createLocalPeer}. Used
115-
* by the subscription system to auto-bind prefetching queues to new DataRegions.
114+
* Optional callback invoked after a local peer is started, either during recovery or via {@link
115+
* #createLocalPeer}. Used by the subscription system to auto-bind prefetching queues to available
116+
* DataRegions.
116117
*/
117118
public static volatile BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> onNewPeerCreated;
118119

120+
/**
121+
* Optional callback invoked after a recovered local peer is constructed and before it starts.
122+
* Used by the subscription system to restore WAL-retention guards before WAL cleanup is enabled.
123+
*/
124+
public static volatile BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> onPeerRecovering;
125+
119126
/**
120127
* Optional callback invoked before a local peer is deleted via {@link #deleteLocalPeer}. Used by
121128
* the subscription system to unbind and clean up prefetching queues before the region is removed.
@@ -244,7 +251,18 @@ private void initAndRecover() throws IOException {
244251
consensusGroupId ->
245252
resetPeerListWithoutThrow.accept(consensusGroupId, Collections.emptyList()));
246253
}
247-
stateMachineMap.values().forEach(IoTConsensusServerImpl::start);
254+
stateMachineMap.forEach(
255+
(consensusGroupId, consensus) -> {
256+
final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> callback = onPeerRecovering;
257+
if (callback != null) {
258+
callback.accept(consensusGroupId, consensus);
259+
}
260+
consensus.start();
261+
// Subscription metadata may have been loaded before any local consensus peers existed.
262+
// Recovered peers must trigger the same binding as newly created peers, after startup
263+
// has restored their indices and installed the pre-start WAL-retention guards.
264+
notifyPeerCreated(consensusGroupId, consensus);
265+
});
248266
}
249267

250268
@Override
@@ -349,11 +367,15 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
349367
throw new ConsensusGroupAlreadyExistException(groupId);
350368
}
351369

352-
// Notify subscription system about new peer creation for auto-binding
370+
notifyPeerCreated(groupId, stateMachineMap.get(groupId));
371+
}
372+
373+
private void notifyPeerCreated(
374+
final ConsensusGroupId groupId, final IoTConsensusServerImpl serverImpl) {
353375
final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> callback = onNewPeerCreated;
354376
if (callback != null) {
355377
try {
356-
callback.accept(groupId, stateMachineMap.get(groupId));
378+
callback.accept(groupId, serverImpl);
357379
} catch (final Exception e) {
358380
logger.warn(
359381
IoTConsensusMessages.LOG_ONNEWPEERCREATED_CALLBACK_FAILED_GROUP_ARG_2671FCDA,

0 commit comments

Comments
 (0)