Skip to content

Commit 9647e38

Browse files
CaideyipiCaideyipi
authored
[Subscription] Restore consensus subscription brokers after restart (#18916)
1 parent e7fd289 commit 9647e38

3 files changed

Lines changed: 359 additions & 4 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: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -111,8 +111,9 @@ 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

@@ -257,6 +258,10 @@ private void initAndRecover() throws IOException {
257258
callback.accept(consensusGroupId, consensus);
258259
}
259260
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);
260265
});
261266
}
262267

@@ -362,11 +367,15 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
362367
throw new ConsensusGroupAlreadyExistException(groupId);
363368
}
364369

365-
// 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) {
366375
final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> callback = onNewPeerCreated;
367376
if (callback != null) {
368377
try {
369-
callback.accept(groupId, stateMachineMap.get(groupId));
378+
callback.accept(groupId, serverImpl);
370379
} catch (final Exception e) {
371380
logger.warn(
372381
IoTConsensusMessages.LOG_ONNEWPEERCREATED_CALLBACK_FAILED_GROUP_ARG_2671FCDA,

‎iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusRecoveryTest.java‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
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.consensus.iot;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
23+
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
24+
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
25+
import org.apache.iotdb.commons.consensus.DataRegionId;
26+
import org.apache.iotdb.consensus.common.Peer;
27+
import org.apache.iotdb.consensus.config.ConsensusConfig;
28+
import org.apache.iotdb.consensus.iot.util.TestStateMachine;
29+
30+
import org.junit.After;
31+
import org.junit.Before;
32+
import org.junit.Rule;
33+
import org.junit.Test;
34+
import org.junit.rules.TemporaryFolder;
35+
36+
import java.util.ArrayList;
37+
import java.util.Arrays;
38+
import java.util.Collections;
39+
import java.util.HashMap;
40+
import java.util.List;
41+
import java.util.Map;
42+
import java.util.function.BiConsumer;
43+
44+
import static org.junit.Assert.assertEquals;
45+
import static org.junit.Assert.assertFalse;
46+
import static org.junit.Assert.assertSame;
47+
import static org.junit.Assert.assertTrue;
48+
49+
public class IoTConsensusRecoveryTest {
50+
51+
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
52+
53+
private final Map<ConsensusGroupId, LifecycleStateMachine> stateMachines = new HashMap<>();
54+
private final Map<ConsensusGroupId, List<Peer>> peerLists = new HashMap<>();
55+
private BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> originalNewPeerCallback;
56+
private BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> originalRecoveringCallback;
57+
private ConsensusConfig config;
58+
private IoTConsensus consensus;
59+
60+
@Before
61+
public void setUp() throws Exception {
62+
originalNewPeerCallback = IoTConsensus.onNewPeerCreated;
63+
originalRecoveringCallback = IoTConsensus.onPeerRecovering;
64+
IoTConsensus.onNewPeerCreated = null;
65+
IoTConsensus.onPeerRecovering = null;
66+
config =
67+
ConsensusConfig.newBuilder()
68+
.setThisNodeId(1)
69+
.setThisNode(new TEndPoint("127.0.0.1", 0))
70+
.setStorageDir(temporaryFolder.newFolder("consensus").getAbsolutePath())
71+
.setRecvSnapshotDirs(
72+
Collections.singletonList(temporaryFolder.newFolder("snapshots").getAbsolutePath()))
73+
.setConsensusGroupType(TConsensusGroupType.DataRegion)
74+
.build();
75+
consensus = newConsensus();
76+
consensus.start();
77+
}
78+
79+
@After
80+
public void tearDown() {
81+
try {
82+
if (consensus != null) {
83+
consensus.stop();
84+
}
85+
} finally {
86+
IoTConsensus.onNewPeerCreated = originalNewPeerCallback;
87+
IoTConsensus.onPeerRecovering = originalRecoveringCallback;
88+
}
89+
}
90+
91+
@Test
92+
public void testRecoveredPeersNotifyAfterStarting() throws Exception {
93+
final DataRegionId firstRegion = new DataRegionId(1);
94+
final DataRegionId secondRegion = new DataRegionId(2);
95+
createPeer(firstRegion);
96+
createPeer(secondRegion);
97+
consensus.stop();
98+
consensus = newConsensus();
99+
100+
final Map<ConsensusGroupId, List<String>> notifications = new HashMap<>();
101+
IoTConsensus.onPeerRecovering =
102+
(groupId, server) -> {
103+
assertSame(server, consensus.getImpl(groupId));
104+
assertFalse(stateMachines.get(groupId).started);
105+
notifications.computeIfAbsent(groupId, ignored -> new ArrayList<>()).add("recovering");
106+
};
107+
IoTConsensus.onNewPeerCreated =
108+
(groupId, server) -> {
109+
assertSame(server, consensus.getImpl(groupId));
110+
assertTrue(stateMachines.get(groupId).started);
111+
notifications.computeIfAbsent(groupId, ignored -> new ArrayList<>()).add("available");
112+
};
113+
114+
consensus.start();
115+
116+
assertEquals(2, notifications.size());
117+
assertEquals(Arrays.asList("recovering", "available"), notifications.get(firstRegion));
118+
assertEquals(Arrays.asList("recovering", "available"), notifications.get(secondRegion));
119+
}
120+
121+
@Test
122+
public void testRecoveryContinuesAfterPeerNotificationFailure() throws Exception {
123+
final DataRegionId firstRegion = new DataRegionId(1);
124+
final DataRegionId secondRegion = new DataRegionId(2);
125+
createPeer(firstRegion);
126+
createPeer(secondRegion);
127+
consensus.stop();
128+
consensus = newConsensus();
129+
130+
final List<ConsensusGroupId> notifiedRegions = new ArrayList<>();
131+
IoTConsensus.onNewPeerCreated =
132+
(groupId, server) -> {
133+
assertTrue(stateMachines.get(groupId).started);
134+
notifiedRegions.add(groupId);
135+
throw new IllegalStateException("subscription binding failed");
136+
};
137+
138+
consensus.start();
139+
140+
assertEquals(2, notifiedRegions.size());
141+
assertTrue(notifiedRegions.containsAll(Arrays.asList(firstRegion, secondRegion)));
142+
assertTrue(stateMachines.get(firstRegion).started);
143+
assertTrue(stateMachines.get(secondRegion).started);
144+
}
145+
146+
private IoTConsensus newConsensus() {
147+
final IoTConsensus instance =
148+
new IoTConsensus(
149+
config,
150+
groupId -> {
151+
final LifecycleStateMachine stateMachine = new LifecycleStateMachine();
152+
stateMachines.put(groupId, stateMachine);
153+
return stateMachine;
154+
});
155+
instance.recordCorrectPeerListBeforeStarting(peerLists);
156+
return instance;
157+
}
158+
159+
private void createPeer(final DataRegionId regionId) throws Exception {
160+
final List<Peer> peers =
161+
Collections.singletonList(
162+
new Peer(regionId, config.getThisNodeId(), config.getThisNodeEndPoint()));
163+
consensus.createLocalPeer(regionId, peers);
164+
peerLists.put(regionId, peers);
165+
}
166+
167+
private static final class LifecycleStateMachine extends TestStateMachine {
168+
169+
private boolean started;
170+
171+
@Override
172+
public void start() {
173+
started = true;
174+
}
175+
176+
@Override
177+
public void stop() {
178+
started = false;
179+
}
180+
}
181+
}

0 commit comments

Comments
 (0)