Skip to content

Commit 0987f6a

Browse files
DanielWang2035VGalaxies
DanielWang2035
and
VGalaxies
authored
Subscription: implement IoTConsensus-based subscription (#17238)
* Subscription: implement IoTConsensus-based subscription * fix some issues * support seek and WAL retention * refactor * refactor * fix * fix * fix part1 * fix part2 * fix part3.1 * fix part3.2 * fix part4.1 * fix seek * fix test * refactor thread model * add time based retention and column pattern * clean old epoch * add config for consensus subscription * fix comments * remove some tests * add check for non-IoTConsensus * refactor IT * fix review * fix UT & IT * fix comments * fix comments * remove writerEpoch field & unify the order of fields * fix test * spotless * fix IT * remove special logic of CONTINUOUS_SAME_SEARCH_INDEX_SEPARATOR_NODE * make partial commit responses explicit * keep processor-buffered messages pinned until release * make manual close safe with processor buffers * tolerate non-owning providers during consensus seek * fix UT * fix IT * refactor progress persist * spotless * fix some issues * fix some issues * extract some logic into broker interface * fix ColumnAlignProcessor * simplify some concepts * fix some naming and comments * fix some issues * fix some issues * rename * Avoided persisting writer.meta on every successful write & Grouped adjacent tree-model rows with the same emitted device/schema into one Tablet * rename * fix * fix * Refine subscription broker routing * Fix consensus subscription replay dedup --------- Co-authored-by: VGalaxies <vgalaxies@apache.org>
1 parent faba002 commit 0987f6a

192 files changed

Lines changed: 24588 additions & 412 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎example/session/src/main/java/org/apache/iotdb/ConsensusSubscriptionWalFileAnalyzer.java‎

Lines changed: 528 additions & 0 deletions
Large diffs are not rendered by default.

‎example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java‎

Lines changed: 198 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,198 @@
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;
21+
22+
import org.apache.iotdb.isession.ISession;
23+
import org.apache.iotdb.isession.util.Version;
24+
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
25+
import org.apache.iotdb.session.Session;
26+
import org.apache.iotdb.session.subscription.ISubscriptionTreeSession;
27+
import org.apache.iotdb.session.subscription.SubscriptionTreeSessionBuilder;
28+
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTreePullConsumer;
29+
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumerBuilder;
30+
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
31+
import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
32+
33+
import org.apache.tsfile.read.query.dataset.ResultSet;
34+
35+
import java.util.List;
36+
import java.util.Properties;
37+
38+
public class ConsensusSubscriptionSessionExample {
39+
40+
private static final String HOST = "127.0.0.1";
41+
private static final int PORT = 6667;
42+
private static final String USER = "root";
43+
private static final String PASSWORD = "root";
44+
45+
private static final long POLL_TIMEOUT_MS = 1_000L;
46+
private static final int MAX_POLL_ROUNDS = 20;
47+
private static final int EXPECTED_ROWS = 5;
48+
49+
public static void main(final String[] args) throws Exception {
50+
final long runId = System.currentTimeMillis();
51+
final String database = "root.db_consensus_example_" + runId;
52+
final String device = database + ".d0";
53+
final String topic = "topic_consensus_example_" + runId;
54+
final String consumerGroup = "cg_consensus_example_" + runId;
55+
final String consumerId = "consumer_consensus_example_" + runId;
56+
57+
System.out.println("=== Consensus Subscription Tree Example ===");
58+
System.out.println("database = " + database);
59+
System.out.println("topic = " + topic);
60+
System.out.println("consumerGroup = " + consumerGroup);
61+
62+
prepareBootstrapData(database, device);
63+
createConsensusTopic(topic, database + ".**");
64+
65+
try (final ISubscriptionTreePullConsumer consumer =
66+
new SubscriptionTreePullConsumerBuilder()
67+
.host(HOST)
68+
.port(PORT)
69+
.username(USER)
70+
.password(PASSWORD)
71+
.consumerId(consumerId)
72+
.consumerGroupId(consumerGroup)
73+
.autoCommit(false)
74+
.build()) {
75+
consumer.open();
76+
consumer.subscribe(topic);
77+
78+
writeRealtimeDataAfterSubscribe(device);
79+
pollAndCommit(consumer, EXPECTED_ROWS);
80+
consumer.unsubscribe(topic);
81+
} finally {
82+
dropTopic(topic);
83+
}
84+
}
85+
86+
private static void prepareBootstrapData(final String database, final String device)
87+
throws Exception {
88+
try (final ISession session = openSession()) {
89+
session.executeNonQueryStatement("CREATE DATABASE " + database);
90+
session.executeNonQueryStatement(
91+
"CREATE TIMESERIES "
92+
+ device
93+
+ ".s1 with datatype=INT64, encoding=RLE, compressor=SNAPPY");
94+
session.executeNonQueryStatement(
95+
String.format("insert into %s(time, s1) values (%d, %d)", device, 0L, 0L));
96+
session.executeNonQueryStatement("flush");
97+
}
98+
}
99+
100+
private static void createConsensusTopic(final String topicName, final String path)
101+
throws Exception {
102+
try (final ISubscriptionTreeSession session =
103+
new SubscriptionTreeSessionBuilder()
104+
.host(HOST)
105+
.port(PORT)
106+
.username(USER)
107+
.password(PASSWORD)
108+
.build()) {
109+
session.open();
110+
111+
final Properties config = new Properties();
112+
config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE);
113+
config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
114+
config.put(TopicConstant.PATH_KEY, path);
115+
config.put(TopicConstant.ORDER_MODE_KEY, TopicConstant.ORDER_MODE_PER_WRITER_VALUE);
116+
session.createTopicIfNotExists(topicName, config);
117+
}
118+
}
119+
120+
private static void writeRealtimeDataAfterSubscribe(final String device) throws Exception {
121+
try (final ISession session = openSession()) {
122+
for (int i = 1; i <= EXPECTED_ROWS; i++) {
123+
session.executeNonQueryStatement(
124+
String.format("insert into %s(time, s1) values (%d, %d)", device, i, i * 10L));
125+
}
126+
session.executeNonQueryStatement("flush");
127+
}
128+
}
129+
130+
private static void pollAndCommit(
131+
final ISubscriptionTreePullConsumer consumer, final int expectedRows) throws Exception {
132+
int totalRows = 0;
133+
int consecutiveEmptyPolls = 0;
134+
135+
for (int round = 1; round <= MAX_POLL_ROUNDS; round++) {
136+
final List<SubscriptionMessage> messages = consumer.poll(POLL_TIMEOUT_MS);
137+
if (messages.isEmpty()) {
138+
consecutiveEmptyPolls++;
139+
if (totalRows >= expectedRows && consecutiveEmptyPolls >= 3) {
140+
break;
141+
}
142+
continue;
143+
}
144+
145+
consecutiveEmptyPolls = 0;
146+
147+
for (final SubscriptionMessage message : messages) {
148+
for (final ResultSet resultSet : message.getResultSets()) {
149+
final SubscriptionRecordHandler.SubscriptionResultSet subscriptionResultSet =
150+
(SubscriptionRecordHandler.SubscriptionResultSet) resultSet;
151+
System.out.println("Columns = " + subscriptionResultSet.getColumnNames());
152+
while (subscriptionResultSet.hasNext()) {
153+
System.out.println(subscriptionResultSet.nextRecord());
154+
totalRows++;
155+
}
156+
}
157+
}
158+
159+
consumer.commitSync(messages);
160+
System.out.println("poll round " + round + ", totalRows = " + totalRows);
161+
}
162+
163+
if (totalRows != expectedRows) {
164+
throw new IllegalStateException(
165+
"Expected "
166+
+ expectedRows
167+
+ " realtime rows, but consumed "
168+
+ totalRows
169+
+ ". Please check whether consensus subscription is enabled on the server.");
170+
}
171+
}
172+
173+
private static void dropTopic(final String topicName) throws Exception {
174+
try (final ISubscriptionTreeSession session =
175+
new SubscriptionTreeSessionBuilder()
176+
.host(HOST)
177+
.port(PORT)
178+
.username(USER)
179+
.password(PASSWORD)
180+
.build()) {
181+
session.open();
182+
session.dropTopicIfExists(topicName);
183+
}
184+
}
185+
186+
private static ISession openSession() throws Exception {
187+
final ISession session =
188+
new Session.Builder()
189+
.host(HOST)
190+
.port(PORT)
191+
.username(USER)
192+
.password(PASSWORD)
193+
.version(Version.V_1_0)
194+
.build();
195+
session.open(false);
196+
return session;
197+
}
198+
}

0 commit comments

Comments
 (0)