Skip to content

Commit 25c34c2

Browse files
codexcodex
committed
Subscription: add topic owner epoch fencing
1 parent 05a816d commit 25c34c2

26 files changed

Lines changed: 946 additions & 2 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionTopicOwnerIT.java‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
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.local;
21+
22+
import org.apache.iotdb.isession.ISession;
23+
import org.apache.iotdb.it.env.EnvFactory;
24+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
25+
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
26+
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
27+
import org.apache.iotdb.rpc.subscription.exception.SubscriptionOwnerFencedException;
28+
import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
29+
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
30+
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
31+
import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
32+
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
33+
34+
import org.apache.tsfile.read.query.dataset.ResultSet;
35+
import org.junit.Assert;
36+
import org.junit.Before;
37+
import org.junit.Ignore;
38+
import org.junit.Test;
39+
import org.junit.experimental.categories.Category;
40+
import org.junit.runner.RunWith;
41+
42+
import java.time.Duration;
43+
import java.util.Collections;
44+
import java.util.List;
45+
import java.util.Properties;
46+
import java.util.concurrent.atomic.AtomicReference;
47+
48+
@RunWith(IoTDBTestRunner.class)
49+
@Category({LocalStandaloneIT.class})
50+
public class IoTDBSubscriptionTopicOwnerIT extends AbstractSubscriptionLocalIT {
51+
52+
@Override
53+
@Before
54+
public void setUp() throws Exception {
55+
super.setUp();
56+
}
57+
58+
@Ignore
59+
@Test
60+
public void testTopicOwnerFencingRejectsStaleOwnerAndAllowsCurrentOwner() throws Exception {
61+
final String host = EnvFactory.getEnv().getIP();
62+
final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
63+
final String topicName = "topic_owner_fencing";
64+
65+
try (final SubscriptionTreeSession session = new SubscriptionTreeSession(host, port)) {
66+
session.open();
67+
final Properties properties = new Properties();
68+
properties.put(TopicConstant.PATH_KEY, "root.topic_owner.**");
69+
properties.put(TopicConstant.START_TIME_KEY, "0");
70+
properties.put(TopicConstant.OWNER_ID_KEY, "sn2");
71+
properties.put(TopicConstant.OWNER_EPOCH_KEY, "6");
72+
session.createTopic(topicName, properties);
73+
}
74+
75+
try {
76+
try (final SubscriptionTreePullConsumer staleOwnerConsumer =
77+
new SubscriptionTreePullConsumer.Builder()
78+
.host(host)
79+
.port(port)
80+
.consumerId("stale_sn")
81+
.consumerGroupId("topic_owner_group")
82+
.ownerId("sn1")
83+
.ownerEpoch(5L)
84+
.autoCommit(false)
85+
.buildPullConsumer()) {
86+
staleOwnerConsumer.open();
87+
Assert.assertThrows(
88+
SubscriptionOwnerFencedException.class, () -> staleOwnerConsumer.subscribe(topicName));
89+
}
90+
91+
try (final SubscriptionTreePullConsumer currentOwnerConsumer =
92+
new SubscriptionTreePullConsumer.Builder()
93+
.host(host)
94+
.port(port)
95+
.consumerId("current_sn")
96+
.consumerGroupId("topic_owner_group")
97+
.ownerId("sn2")
98+
.ownerEpoch(6L)
99+
.autoCommit(false)
100+
.buildPullConsumer()) {
101+
currentOwnerConsumer.open();
102+
currentOwnerConsumer.subscribe(topicName);
103+
104+
insertData();
105+
106+
final AtomicReference<List<SubscriptionMessage>> polledMessages =
107+
new AtomicReference<>(Collections.emptyList());
108+
IoTDBSubscriptionITConstant.AWAIT.untilAsserted(
109+
() -> {
110+
final List<SubscriptionMessage> messages =
111+
currentOwnerConsumer.poll(Duration.ofMillis(1000));
112+
polledMessages.set(messages);
113+
Assert.assertFalse(messages.isEmpty());
114+
Assert.assertTrue(countRows(messages) > 0);
115+
});
116+
117+
currentOwnerConsumer.commitSync(polledMessages.get());
118+
currentOwnerConsumer.unsubscribe(topicName);
119+
}
120+
} finally {
121+
try (final SubscriptionTreeSession session = new SubscriptionTreeSession(host, port)) {
122+
session.open();
123+
session.dropTopicIfExists(topicName);
124+
}
125+
}
126+
}
127+
128+
private void insertData() throws Exception {
129+
try (final ISession session = EnvFactory.getEnv().getSessionConnection()) {
130+
for (int i = 0; i < 10; i++) {
131+
session.executeNonQueryStatement(
132+
String.format("insert into root.topic_owner.d1(time, s1) values (%s, %s)", i, i));
133+
}
134+
session.executeNonQueryStatement("flush");
135+
}
136+
}
137+
138+
private static int countRows(final List<SubscriptionMessage> messages) throws Exception {
139+
int rowCount = 0;
140+
for (final SubscriptionMessage message : messages) {
141+
for (final ResultSet resultSet : message.getResultSets()) {
142+
while (((SubscriptionRecordHandler.SubscriptionResultSet) resultSet).hasNext()) {
143+
resultSet.next();
144+
rowCount++;
145+
}
146+
}
147+
}
148+
return rowCount;
149+
}
150+
}

‎iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -316,6 +316,11 @@ public enum TSStatusCode {
316316
SHOW_SUBSCRIPTION_ERROR(1910),
317317
SUBSCRIPTION_PIPE_TIMEOUT_ERROR(1911),
318318
SUBSCRIPTION_NOT_ENABLED_ERROR(1912),
319+
SUBSCRIPTION_OWNER_FENCED(1913),
320+
SUBSCRIPTION_OWNER_REQUIRED(1914),
321+
SUBSCRIPTION_OWNER_EPOCH_REQUIRED(1915),
322+
SUBSCRIPTION_OWNER_LEASE_EXPIRED(1916),
323+
SUBSCRIPTION_OWNER_EPOCH_CONFLICT(1917),
319324

320325
// Topic
321326
CREATE_TOPIC_ERROR(2000),

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/ConsumerConfig.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,16 @@ public String getConsumerGroupId() {
6868
return getString(ConsumerConstant.CONSUMER_GROUP_ID_KEY);
6969
}
7070

71+
public String getOwnerId() {
72+
return getString(ConsumerConstant.OWNER_ID_KEY);
73+
}
74+
75+
public Long getOwnerEpoch() {
76+
return hasAttribute(ConsumerConstant.OWNER_EPOCH_KEY)
77+
? getLong(ConsumerConstant.OWNER_EPOCH_KEY)
78+
: null;
79+
}
80+
7181
public String getUsername() {
7282
return getString(ConsumerConstant.USERNAME_KEY);
7383
}

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/ConsumerConstant.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ public class ConsumerConstant {
4040

4141
public static final String CONSUMER_ID_KEY = "consumer-id";
4242
public static final String CONSUMER_GROUP_ID_KEY = "group-id";
43+
public static final String OWNER_ID_KEY = "owner-id";
44+
public static final String OWNER_EPOCH_KEY = "owner-epoch";
4345

4446
public static final String HEARTBEAT_INTERVAL_MS_KEY = "heartbeat-interval-ms";
4547
public static final long HEARTBEAT_INTERVAL_MS_DEFAULT_VALUE = 30_000L;

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,10 @@ public class TopicConstant {
6161
public static final String STRICT_KEY = "strict";
6262
public static final String STRICT_DEFAULT_VALUE = "true";
6363

64+
public static final String OWNER_ID_KEY = "owner-id";
65+
public static final String OWNER_EPOCH_KEY = "owner-epoch";
66+
public static final String OWNER_LEASE_EXPIRE_TIME_MS_KEY = "owner-lease-expire-time-ms";
67+
6468
private TopicConstant() {
6569
throw new IllegalStateException(SubscriptionMessages.UTILITY_CLASS);
6670
}

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionOwnerFencedException.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
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.rpc.subscription.exception;
21+
22+
import java.util.Objects;
23+
24+
public class SubscriptionOwnerFencedException extends SubscriptionRuntimeNonCriticalException {
25+
26+
public SubscriptionOwnerFencedException(final String message) {
27+
super(message);
28+
}
29+
30+
public SubscriptionOwnerFencedException(final String message, final Throwable cause) {
31+
super(message, cause);
32+
}
33+
34+
@Override
35+
public boolean equals(final Object obj) {
36+
return obj instanceof SubscriptionOwnerFencedException
37+
&& Objects.equals(getMessage(), ((SubscriptionOwnerFencedException) obj).getMessage())
38+
&& Objects.equals(getTimeStamp(), ((SubscriptionOwnerFencedException) obj).getTimeStamp());
39+
}
40+
41+
@Override
42+
public int hashCode() {
43+
return super.hashCode();
44+
}
45+
}

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
2626
import org.apache.iotdb.rpc.subscription.exception.SubscriptionConnectionException;
2727
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
28+
import org.apache.iotdb.rpc.subscription.exception.SubscriptionOwnerFencedException;
2829
import org.apache.iotdb.rpc.subscription.exception.SubscriptionPipeTimeoutException;
2930
import org.apache.iotdb.rpc.subscription.exception.SubscriptionPollTimeoutException;
3031
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
@@ -105,6 +106,8 @@ abstract class AbstractSubscriptionConsumer implements AutoCloseable {
105106

106107
protected String consumerId;
107108
protected String consumerGroupId;
109+
protected String ownerId;
110+
protected Long ownerEpoch;
108111

109112
private final long heartbeatIntervalMs;
110113
private final long endpointsSyncIntervalMs;
@@ -154,6 +157,14 @@ public String getConsumerGroupId() {
154157
return consumerGroupId;
155158
}
156159

160+
public String getOwnerId() {
161+
return ownerId;
162+
}
163+
164+
public Long getOwnerEpoch() {
165+
return ownerEpoch;
166+
}
167+
157168
/////////////////////////////// ctor ///////////////////////////////
158169

159170
protected AbstractSubscriptionConsumer(final AbstractSubscriptionConsumerBuilder builder) {
@@ -183,6 +194,8 @@ protected AbstractSubscriptionConsumer(final AbstractSubscriptionConsumerBuilder
183194

184195
this.consumerId = builder.consumerId;
185196
this.consumerGroupId = builder.consumerGroupId;
197+
this.ownerId = builder.ownerId;
198+
this.ownerEpoch = builder.ownerEpoch;
186199

187200
this.heartbeatIntervalMs = builder.heartbeatIntervalMs;
188201
this.endpointsSyncIntervalMs = builder.endpointsSyncIntervalMs;
@@ -213,6 +226,8 @@ protected AbstractSubscriptionConsumer(
213226
.encryptedPassword((String) properties.get(ConsumerConstant.ENCRYPTED_PASSWORD_KEY))
214227
.consumerId((String) properties.get(ConsumerConstant.CONSUMER_ID_KEY))
215228
.consumerGroupId((String) properties.get(ConsumerConstant.CONSUMER_GROUP_ID_KEY))
229+
.ownerId((String) properties.get(ConsumerConstant.OWNER_ID_KEY))
230+
.ownerEpoch((Long) properties.get(ConsumerConstant.OWNER_EPOCH_KEY))
216231
.heartbeatIntervalMs(
217232
(Long)
218233
properties.getOrDefault(
@@ -394,6 +409,8 @@ protected abstract AbstractSubscriptionProvider constructSubscriptionProvider(
394409
final String encryptedPassword,
395410
final String consumerId,
396411
final String consumerGroupId,
412+
final String ownerId,
413+
final Long ownerEpoch,
397414
final int thriftMaxFrameSize,
398415
final long heartbeatIntervalMs,
399416
final int connectionTimeoutInMs);
@@ -408,6 +425,8 @@ AbstractSubscriptionProvider constructProviderAndHandshake(final TEndPoint endPo
408425
this.encryptedPassword,
409426
this.consumerId,
410427
this.consumerGroupId,
428+
this.ownerId,
429+
this.ownerEpoch,
411430
this.thriftMaxFrameSize,
412431
this.heartbeatIntervalMs,
413432
this.connectionTimeoutInMs);
@@ -1341,6 +1360,9 @@ private void subscribeWithRedirection(final Set<String> topicNames) throws Subsc
13411360
subscribedTopics = provider.subscribe(topicNames);
13421361
return;
13431362
} catch (final Exception e) {
1363+
if (e instanceof SubscriptionOwnerFencedException) {
1364+
throw (SubscriptionOwnerFencedException) e;
1365+
}
13441366
if (e instanceof SubscriptionPipeTimeoutException) {
13451367
// degrade exception to log for pipe timeout
13461368
LOGGER.warn(e.getMessage());
@@ -1429,6 +1451,8 @@ protected Map<String, String> coreReportMessage() {
14291451
final Map<String, String> result = new HashMap<>();
14301452
result.put("consumerId", consumerId);
14311453
result.put("consumerGroupId", consumerGroupId);
1454+
result.put("ownerId", ownerId);
1455+
result.put("ownerEpoch", String.valueOf(ownerEpoch));
14321456
result.put("isClosed", isClosed.toString());
14331457
result.put("fileSaveDir", fileSaveDir);
14341458
result.put(
@@ -1443,6 +1467,8 @@ protected Map<String, String> allReportMessage() {
14431467
final Map<String, String> result = new HashMap<>();
14441468
result.put("consumerId", consumerId);
14451469
result.put("consumerGroupId", consumerGroupId);
1470+
result.put("ownerId", ownerId);
1471+
result.put("ownerEpoch", String.valueOf(ownerEpoch));
14461472
result.put("heartbeatIntervalMs", String.valueOf(heartbeatIntervalMs));
14471473
result.put("endpointsSyncIntervalMs", String.valueOf(endpointsSyncIntervalMs));
14481474
result.put("providers", providers.toString());

‎iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumerBuilder.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ public class AbstractSubscriptionConsumerBuilder {
4040

4141
protected String consumerId;
4242
protected String consumerGroupId;
43+
protected String ownerId;
44+
protected Long ownerEpoch;
4345

4446
protected long heartbeatIntervalMs = ConsumerConstant.HEARTBEAT_INTERVAL_MS_DEFAULT_VALUE;
4547
protected long endpointsSyncIntervalMs =
@@ -111,6 +113,27 @@ public AbstractSubscriptionConsumerBuilder consumerGroupId(
111113
return this;
112114
}
113115

116+
public AbstractSubscriptionConsumerBuilder ownerId(@Nullable final String ownerId) {
117+
if (Objects.isNull(ownerId)) {
118+
return this;
119+
}
120+
this.ownerId = ownerId;
121+
return this;
122+
}
123+
124+
public AbstractSubscriptionConsumerBuilder ownerEpoch(final long ownerEpoch) {
125+
this.ownerEpoch = ownerEpoch;
126+
return this;
127+
}
128+
129+
public AbstractSubscriptionConsumerBuilder ownerEpoch(@Nullable final Long ownerEpoch) {
130+
if (Objects.isNull(ownerEpoch)) {
131+
return this;
132+
}
133+
this.ownerEpoch = ownerEpoch;
134+
return this;
135+
}
136+
114137
public AbstractSubscriptionConsumerBuilder heartbeatIntervalMs(final long heartbeatIntervalMs) {
115138
this.heartbeatIntervalMs =
116139
Math.max(heartbeatIntervalMs, ConsumerConstant.HEARTBEAT_INTERVAL_MS_MIN_VALUE);

0 commit comments

Comments
 (0)