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 ec473f7fd1d [Subscription] Isolate tree and table topics (#18397)
ec473f7fd1d is described below
commit ec473f7fd1d64d690bc770d034ace348fb016e3f
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 6 18:47:10 2026 +0800
[Subscription] Isolate tree and table topics (#18397)
* [Subscription] Isolate tree and table topics
* [Subscription] Fix retry test topic lookup mocks
---
.../tablemodel/IoTDBSubscriptionIsolationIT.java | 161 ++++++++++-----------
.../write/subscription/topic/DropTopicPlan.java | 31 +++-
.../iotdb/confignode/manager/ProcedureManager.java | 23 ++-
.../subscription/SubscriptionCoordinator.java | 76 +++++++---
.../persistence/subscription/SubscriptionInfo.java | 93 +++++++++---
.../procedure/env/ConfigNodeProcedureEnv.java | 12 ++
.../subscription/CreateSubscriptionProcedure.java | 3 +-
.../subscription/DropSubscriptionProcedure.java | 3 +-
.../subscription/topic/AlterTopicProcedure.java | 6 +-
.../subscription/topic/CreateTopicProcedure.java | 6 +-
.../subscription/topic/DropTopicProcedure.java | 41 +++++-
.../request/ConfigPhysicalPlanSerDeTest.java | 8 +-
.../subscription/SubscriptionCoordinatorTest.java | 19 ++-
.../subscription/SubscriptionInfoTest.java | 66 +++++++++
.../CreateSubscriptionProcedureTest.java | 2 +-
.../DropSubscriptionProcedureTest.java | 2 +-
.../subscription/topic/DropTopicProcedureTest.java | 19 +++
.../impl/DataNodeInternalRPCServiceImpl.java | 6 +-
.../agent/SubscriptionBrokerAgent.java | 47 ++++--
.../agent/SubscriptionConsumerAgent.java | 22 ++-
.../subscription/agent/SubscriptionTopicAgent.java | 83 ++++++++---
.../db/subscription/broker/SubscriptionBroker.java | 14 +-
.../broker/SubscriptionPrefetchingQueue.java | 9 +-
.../consensus/ConsensusLogToTabletConverter.java | 2 +-
.../consensus/ConsensusPrefetchingQueue.java | 9 +-
.../ConsensusSubscriptionSetupHandler.java | 32 +++-
.../batch/SubscriptionPipeTabletEventBatch.java | 9 +-
.../batch/SubscriptionPipeTsFileEventBatch.java | 5 +-
.../response/SubscriptionEventTabletResponse.java | 5 +-
.../response/SubscriptionEventTsFileResponse.java | 5 +-
.../receiver/SubscriptionReceiverV1.java | 18 ++-
.../task/subtask/SubscriptionSinkSubtask.java | 3 +-
.../subtask/SubscriptionSinkSubtaskLifeCycle.java | 7 +-
.../apache/iotdb/commons/i18n/PipeMessages.java | 1 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 1 +
.../meta/consumer/ConsumerGroupMeta.java | 28 ++++
.../meta/consumer/ConsumerGroupMetaKeeper.java | 20 +++
.../subscription/meta/consumer/ConsumerMeta.java | 4 +
.../subscription/meta/topic/TopicMetaKeeper.java | 42 ++++--
.../consumer/ConsumerGroupDeSerTest.java | 25 ++++
.../subscription/topic/TopicMetaKeeperTest.java | 120 +++++++++++++++
.../src/main/thrift/datanode.thrift | 2 +
42 files changed, 865 insertions(+), 225 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionIsolationIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionIsolationIT.java
index 99b674d1079..2d15cff252b 100644
---
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionIsolationIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionIsolationIT.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.subscription.it.local.tablemodel;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
import org.apache.iotdb.session.subscription.ISubscriptionTreeSession;
import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
@@ -34,12 +35,11 @@ import
org.apache.iotdb.subscription.it.local.AbstractSubscriptionLocalIT;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
-import static org.junit.Assert.fail;
+import java.util.Properties;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class})
@@ -51,26 +51,35 @@ public class IoTDBSubscriptionIsolationIT extends
AbstractSubscriptionLocalIT {
super.setUp();
}
- @Ignore
@Test
public void testTopicIsolation() throws Exception {
- final String treeTopicName = "treeTopic";
- final String tableTopicName = "tableTopic";
+ final String topicName = "topic";
final String host = EnvFactory.getEnv().getIP();
final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
- // create tree topic
+ final Properties treeTopicProperties = new Properties();
+ treeTopicProperties.setProperty(TopicConstant.PATH_KEY, "root.tree.**");
try (final ISubscriptionTreeSession session =
new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
session.open();
- session.createTopic(treeTopicName);
+ session.createTopic(topicName, treeTopicProperties);
+
+ final Properties alteredProperties = new Properties();
+ alteredProperties.setProperty(TopicConstant.PATH_KEY,
"root.tree_altered.**");
+ session.alterTopic(topicName, alteredProperties);
}
- // create table topic
+ final Properties tableTopicProperties = new Properties();
+ tableTopicProperties.setProperty(TopicConstant.DATABASE_KEY, "table_db");
+ tableTopicProperties.setProperty(TopicConstant.TABLE_KEY, "table_name");
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
- session.createTopic(tableTopicName);
+ session.createTopic(topicName, tableTopicProperties);
+
+ final Properties alteredProperties = new Properties();
+ alteredProperties.setProperty(TopicConstant.DATABASE_KEY,
"table_db_altered");
+ session.alterTopic(topicName, alteredProperties);
}
// show topic on tree session
@@ -78,58 +87,43 @@ public class IoTDBSubscriptionIsolationIT extends
AbstractSubscriptionLocalIT {
new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
session.open();
Assert.assertEquals(1, session.getTopics().size());
- Assert.assertTrue(session.getTopic(treeTopicName).isPresent());
- Assert.assertFalse(session.getTopic(tableTopicName).isPresent());
+ Assert.assertTrue(session.getTopic(topicName).isPresent());
+ Assert.assertTrue(
+
session.getTopic(topicName).get().getTopicAttributes().contains("root.tree_altered.**"));
+ Assert.assertFalse(
+
session.getTopic(topicName).get().getTopicAttributes().contains("table_db_altered"));
}
// show topic on table session
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
Assert.assertEquals(1, session.getTopics().size());
- Assert.assertTrue(session.getTopic(tableTopicName).isPresent());
- Assert.assertFalse(session.getTopic(treeTopicName).isPresent());
- }
-
- // drop table topic on tree session
- try (final ISubscriptionTreeSession session =
- new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
- session.open();
- try {
- session.dropTopic(tableTopicName);
- fail();
- } catch (final Exception ignored) {
- }
+ Assert.assertTrue(session.getTopic(topicName).isPresent());
+ Assert.assertTrue(
+
session.getTopic(topicName).get().getTopicAttributes().contains("table_db_altered"));
+ Assert.assertFalse(
+
session.getTopic(topicName).get().getTopicAttributes().contains("root.tree_altered.**"));
}
- // drop tree topic on table session
- try (final ISubscriptionTableSession session =
- new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
- try {
- session.dropTopic(treeTopicName);
- fail();
- } catch (final Exception ignored) {
- }
- }
-
- // drop tree topic on tree session
+ // Dropping the tree-model topic must not affect the same-named
table-model topic.
try (final ISubscriptionTreeSession session =
new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
session.open();
- session.dropTopic(treeTopicName);
+ session.dropTopic(topicName);
+ Assert.assertFalse(session.getTopic(topicName).isPresent());
}
- // drop table topic on table session
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
- session.dropTopic(tableTopicName);
+ Assert.assertTrue(session.getTopic(topicName).isPresent());
+ session.dropTopic(topicName);
+ Assert.assertFalse(session.getTopic(topicName).isPresent());
}
}
- @Ignore
@Test
public void testSubscriptionIsolation() throws Exception {
- final String treeTopicName = "treeTopic";
- final String tableTopicName = "tableTopic";
+ final String topicName = "topic";
final String host = EnvFactory.getEnv().getIP();
final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
@@ -138,83 +132,74 @@ public class IoTDBSubscriptionIsolationIT extends
AbstractSubscriptionLocalIT {
try (final ISubscriptionTreeSession session =
new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
session.open();
- session.createTopic(treeTopicName);
+ session.createTopic(topicName);
}
// create table topic
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
- session.createTopic(tableTopicName);
+ session.createTopic(topicName);
}
- // subscribe table topic on tree consumer
- try (final ISubscriptionTreePullConsumer consumer =
- new
SubscriptionTreePullConsumerBuilder().host(host).port(port).build()) {
- consumer.open();
- try {
- consumer.subscribe(tableTopicName);
- fail();
- } catch (final Exception ignored) {
- }
- }
-
- // subscribe tree topic on table consumer
- try (final ISubscriptionTablePullConsumer consumer =
- new
SubscriptionTablePullConsumerBuilder().host(host).port(port).build()) {
- consumer.open();
- try {
- consumer.subscribe(treeTopicName);
- fail();
- } catch (final Exception ignored) {
- }
- }
-
- // subscribe tree topic on tree consumer
final ISubscriptionTreePullConsumer treeConsumer =
- new
SubscriptionTreePullConsumerBuilder().host(host).port(port).build();
+ new SubscriptionTreePullConsumerBuilder()
+ .host(host)
+ .port(port)
+ .consumerId("tree_consumer")
+ .consumerGroupId("tree_consumer_group")
+ .build();
treeConsumer.open();
- treeConsumer.subscribe(treeTopicName);
+ treeConsumer.subscribe(topicName);
- // subscribe table topic on table consumer
final ISubscriptionTablePullConsumer tableConsumer =
- new
SubscriptionTablePullConsumerBuilder().host(host).port(port).build();
+ new SubscriptionTablePullConsumerBuilder()
+ .host(host)
+ .port(port)
+ .consumerId("table_consumer")
+ .consumerGroupId("table_consumer_group")
+ .build();
tableConsumer.open();
- tableConsumer.subscribe(tableTopicName);
+ tableConsumer.subscribe(topicName);
// show subscription on tree session
try (final ISubscriptionTreeSession session =
new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
session.open();
Assert.assertEquals(1, session.getSubscriptions().size());
- Assert.assertEquals(1, session.getSubscriptions(treeTopicName).size());
- Assert.assertEquals(0, session.getSubscriptions(tableTopicName).size());
+ Assert.assertEquals(1, session.getSubscriptions(topicName).size());
+ Assert.assertEquals(
+ "tree_consumer_group",
+
session.getSubscriptions(topicName).iterator().next().getConsumerGroupId());
}
// show subscription on table session
try (final ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
Assert.assertEquals(1, session.getSubscriptions().size());
- Assert.assertEquals(1, session.getSubscriptions(tableTopicName).size());
- Assert.assertEquals(0, session.getSubscriptions(treeTopicName).size());
+ Assert.assertEquals(1, session.getSubscriptions(topicName).size());
+ Assert.assertEquals(
+ "table_consumer_group",
+
session.getSubscriptions(topicName).iterator().next().getConsumerGroupId());
}
- // unsubscribe table topic on tree consumer
- try {
- treeConsumer.unsubscribe(tableTopicName);
- fail();
- } catch (final Exception ignored) {
-
+ // Unsubscribing the tree-model topic must not affect the same-named
table-model subscription.
+ treeConsumer.unsubscribe(topicName);
+ try (final ISubscriptionTreeSession session =
+ new SubscriptionTreeSessionBuilder().host(host).port(port).build()) {
+ session.open();
+ Assert.assertEquals(0, session.getSubscriptions(topicName).size());
+ }
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
+ Assert.assertEquals(1, session.getSubscriptions(topicName).size());
}
- // unsubscribe tree topic on table consumer
- try {
- tableConsumer.unsubscribe(treeTopicName);
- fail();
- } catch (final Exception ignored) {
-
+ tableConsumer.unsubscribe(topicName);
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder().host(host).port(port).build()) {
+ Assert.assertEquals(0, session.getSubscriptions(topicName).size());
}
- // close consumers
treeConsumer.close();
tableConsumer.close();
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/subscription/topic/DropTopicPlan.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/subscription/topic/DropTopicPlan.java
index 75642f92f11..5604fc311bd 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/subscription/topic/DropTopicPlan.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/subscription/topic/DropTopicPlan.java
@@ -32,6 +32,8 @@ import java.util.Objects;
public class DropTopicPlan extends ConfigPhysicalPlan {
private String topicName;
+ private boolean isTableModel;
+ private boolean isTableModelSet;
public DropTopicPlan() {
super(ConfigPhysicalPlanType.DropTopic);
@@ -42,19 +44,40 @@ public class DropTopicPlan extends ConfigPhysicalPlan {
this.topicName = topicName;
}
+ public DropTopicPlan(String topicName, boolean isTableModel) {
+ this(topicName);
+ this.isTableModel = isTableModel;
+ this.isTableModelSet = true;
+ }
+
public String getTopicName() {
return topicName;
}
+ public boolean isTableModel() {
+ return isTableModel;
+ }
+
+ public boolean isTableModelSet() {
+ return isTableModelSet;
+ }
+
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ReadWriteIOUtils.write(topicName, stream);
+ if (isTableModelSet) {
+ ReadWriteIOUtils.write(isTableModel, stream);
+ }
}
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
topicName = ReadWriteIOUtils.readString(buffer);
+ isTableModelSet = buffer.hasRemaining();
+ if (isTableModelSet) {
+ isTableModel = ReadWriteIOUtils.readBool(buffer);
+ }
}
@Override
@@ -66,16 +89,18 @@ public class DropTopicPlan extends ConfigPhysicalPlan {
return false;
}
DropTopicPlan that = (DropTopicPlan) obj;
- return Objects.equals(topicName, that.topicName);
+ return isTableModel == that.isTableModel
+ && isTableModelSet == that.isTableModelSet
+ && Objects.equals(topicName, that.topicName);
}
@Override
public int hashCode() {
- return Objects.hash(topicName);
+ return Objects.hash(topicName, isTableModel, isTableModelSet);
}
@Override
public String toString() {
- return "DropTopicPlan{" + "topicName='" + topicName + "'}";
+ return "DropTopicPlan{" + "topicName='" + topicName + "', isTableModel=" +
isTableModel + '}';
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index 45cea0a31e3..50b003d0494 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -1922,7 +1922,8 @@ public class ProcedureManager {
public TSStatus alterTopic(TAlterTopicReq req) {
final SubscriptionCoordinator subscriptionCoordinator =
configManager.getSubscriptionManager().getSubscriptionCoordinator();
- subscriptionCoordinator.lockTopicAlteration(req.getTopicName());
+ final boolean isTableModel = new
TopicConfig(req.getTopicAttributes()).isTableTopic();
+ subscriptionCoordinator.lockTopicAlteration(req.getTopicName(),
isTableModel);
boolean isOwnerLeaseRenewalBlocked = false;
try {
isOwnerLeaseRenewalBlocked =
@@ -1973,9 +1974,9 @@ public class ProcedureManager {
.setMessage(e.getMessage());
} finally {
if (isOwnerLeaseRenewalBlocked) {
- subscriptionCoordinator.unblockOwnerLeaseRenewal(req.getTopicName());
+ subscriptionCoordinator.unblockOwnerLeaseRenewal(req.getTopicName(),
isTableModel);
}
- subscriptionCoordinator.unlockTopicAlteration(req.getTopicName());
+ subscriptionCoordinator.unlockTopicAlteration(req.getTopicName(),
isTableModel);
}
}
@@ -2068,6 +2069,22 @@ public class ProcedureManager {
}
}
+ public TSStatus dropTopic(String topicName, boolean isTableModel) {
+ try {
+ DropTopicProcedure procedure = new DropTopicProcedure(topicName,
isTableModel);
+ executor.submitProcedure(procedure);
+ TSStatus status = waitingProcedureFinished(procedure);
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return status;
+ } else {
+ return new TSStatus(TSStatusCode.DROP_TOPIC_ERROR.getStatusCode())
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
+ }
+ } catch (Exception e) {
+ return new
TSStatus(TSStatusCode.DROP_TOPIC_ERROR.getStatusCode()).setMessage(e.getMessage());
+ }
+ }
+
public TSStatus topicMetaSync() {
try {
TopicMetaSyncProcedure procedure = new TopicMetaSyncProcedure();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
index 1d988eabc93..72db41afe8e 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
@@ -51,6 +51,7 @@ import
org.apache.iotdb.mpp.rpc.thrift.TPushTopicOwnerLeaseReq;
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
@@ -82,16 +83,18 @@ public class SubscriptionCoordinator {
private final SubscriptionMetaSyncer subscriptionMetaSyncer;
private final SubscriptionOwnerLeaseSyncer subscriptionOwnerLeaseSyncer;
- // Serialize client ALTER TOPIC requests per topic. Besides preventing
ordinary partial updates
- // from being built from the same snapshot, this also ensures that only one
owner-transfer
- // request can own the renewal-block entry for a topic at a time.
- private final Map<String, TopicAlterationLock> topicAlterationLocks = new
HashMap<>();
+ // Serialize client ALTER TOPIC requests per model-qualified topic. Besides
preventing ordinary
+ // partial updates from being built from the same snapshot, this also
ensures that only one
+ // owner-transfer request can own the renewal-block entry for a topic at a
time.
+ private final Map<TopicModelKey, TopicAlterationLock> topicAlterationLocks =
new HashMap<>();
- // topicName -> blockSinceMs (ConfigNode local clock when owner-lease
renewal was stopped for an
- // in-flight owner transfer). Used to skip renewal and to bound the
admission wait.
- private final Map<String, Long> blockedOwnerLeaseRenewalTopics =
+ // model-qualified topic -> blockSinceMs (ConfigNode local clock when
owner-lease renewal was
+ // stopped for an in-flight owner transfer). Used to skip renewal and to
bound the admission wait.
+ private final Map<TopicModelKey, Long> blockedOwnerLeaseRenewalTopics =
Collections.synchronizedMap(new HashMap<>());
+ private record TopicModelKey(String topicName, boolean isTableModel) {}
+
private static class TopicAlterationLock {
private final ReentrantLock lock = new ReentrantLock(true);
@@ -148,25 +151,35 @@ public class SubscriptionCoordinator {
}
public void lockTopicAlteration(final String topicName) {
+ lockTopicAlteration(topicName, false);
+ }
+
+ public void lockTopicAlteration(final String topicName, final boolean
isTableModel) {
+ final TopicModelKey topicModelKey = new TopicModelKey(topicName,
isTableModel);
final TopicAlterationLock topicAlterationLock;
synchronized (topicAlterationLocks) {
topicAlterationLock =
- topicAlterationLocks.computeIfAbsent(topicName, ignored -> new
TopicAlterationLock());
+ topicAlterationLocks.computeIfAbsent(topicModelKey, ignored -> new
TopicAlterationLock());
topicAlterationLock.referenceCount++;
}
topicAlterationLock.lock.lock();
}
public void unlockTopicAlteration(final String topicName) {
+ unlockTopicAlteration(topicName, false);
+ }
+
+ public void unlockTopicAlteration(final String topicName, final boolean
isTableModel) {
+ final TopicModelKey topicModelKey = new TopicModelKey(topicName,
isTableModel);
final TopicAlterationLock topicAlterationLock;
synchronized (topicAlterationLocks) {
- topicAlterationLock = topicAlterationLocks.get(topicName);
+ topicAlterationLock = topicAlterationLocks.get(topicModelKey);
}
topicAlterationLock.lock.unlock();
synchronized (topicAlterationLocks) {
if (--topicAlterationLock.referenceCount == 0) {
- topicAlterationLocks.remove(topicName, topicAlterationLock);
+ topicAlterationLocks.remove(topicModelKey, topicAlterationLock);
}
}
}
@@ -221,7 +234,9 @@ public class SubscriptionCoordinator {
}
public boolean blockOwnerLeaseRenewalIfOwnerTransfer(TAlterTopicReq req) {
- final TopicMeta currentTopicMeta =
subscriptionInfo.deepCopyTopicMeta(req.getTopicName());
+ final TopicMeta currentTopicMeta =
+ subscriptionInfo.deepCopyTopicMeta(
+ req.getTopicName(), new
TopicConfig(req.getTopicAttributes()).isTableTopic());
final TopicMeta updatedTopicMeta =
Objects.isNull(currentTopicMeta)
? null
@@ -232,12 +247,18 @@ public class SubscriptionCoordinator {
return false;
}
- blockedOwnerLeaseRenewalTopics.put(req.getTopicName(),
System.currentTimeMillis());
+ blockedOwnerLeaseRenewalTopics.put(
+ new TopicModelKey(req.getTopicName(),
currentTopicMeta.visibleUnderTableModel()),
+ System.currentTimeMillis());
return true;
}
public void unblockOwnerLeaseRenewal(String topicName) {
- blockedOwnerLeaseRenewalTopics.remove(topicName);
+ unblockOwnerLeaseRenewal(topicName, false);
+ }
+
+ public void unblockOwnerLeaseRenewal(final String topicName, final boolean
isTableModel) {
+ blockedOwnerLeaseRenewalTopics.remove(new TopicModelKey(topicName,
isTableModel));
}
/**
@@ -250,7 +271,9 @@ public class SubscriptionCoordinator {
*/
public TopicMeta buildAlteredTopicMetaAfterOwnerLeaseExpired(TAlterTopicReq
req)
throws InterruptedException {
- final TopicMeta currentTopicMeta =
subscriptionInfo.deepCopyTopicMeta(req.getTopicName());
+ final TopicMeta currentTopicMeta =
+ subscriptionInfo.deepCopyTopicMeta(
+ req.getTopicName(), new
TopicConfig(req.getTopicAttributes()).isTableTopic());
final TopicMeta updatedTopicMeta =
Objects.isNull(currentTopicMeta)
? null
@@ -268,7 +291,8 @@ public class SubscriptionCoordinator {
return updatedTopicMeta;
}
- waitForOwnerLeaseExpiration(req.getTopicName(), leaseDurationMs);
+ waitForOwnerLeaseExpiration(
+ req.getTopicName(), currentTopicMeta.visibleUnderTableModel(),
leaseDurationMs);
// Another alteration may have completed while this owner transfer was
waiting for the old
// lease to drain. Rebuild from the latest TopicMeta so that this request
only applies its own
@@ -278,7 +302,14 @@ public class SubscriptionCoordinator {
void waitForOwnerLeaseExpiration(final String topicName, final long
leaseDurationMs)
throws InterruptedException {
- final Long blockSinceMs = blockedOwnerLeaseRenewalTopics.get(topicName);
+ waitForOwnerLeaseExpiration(topicName, false, leaseDurationMs);
+ }
+
+ void waitForOwnerLeaseExpiration(
+ final String topicName, final boolean isTableModel, final long
leaseDurationMs)
+ throws InterruptedException {
+ final Long blockSinceMs =
+ blockedOwnerLeaseRenewalTopics.get(new TopicModelKey(topicName,
isTableModel));
if (Objects.isNull(blockSinceMs)) {
return;
}
@@ -303,13 +334,18 @@ public class SubscriptionCoordinator {
* let its local lease expire and fence the owner (fail-closed).
*/
public void pushTopicOwnerLeasesToDataNodes() {
- final Set<String> blockedTopicNames;
+ final Set<String> blockedTreeTopicNames = new HashSet<>();
+ final Set<String> blockedTableTopicNames = new HashSet<>();
synchronized (blockedOwnerLeaseRenewalTopics) {
- blockedTopicNames = new
HashSet<>(blockedOwnerLeaseRenewalTopics.keySet());
+ for (final TopicModelKey topicModelKey :
blockedOwnerLeaseRenewalTopics.keySet()) {
+ (topicModelKey.isTableModel() ? blockedTableTopicNames :
blockedTreeTopicNames)
+ .add(topicModelKey.topicName());
+ }
}
final List<TTopicOwnerLeaseEntry> ownerLeases =
- subscriptionInfo.collectTopicOwnerLeaseEntries(blockedTopicNames);
+ subscriptionInfo.collectTopicOwnerLeaseEntries(
+ blockedTreeTopicNames, blockedTableTopicNames);
if (ownerLeases.isEmpty()) {
return;
}
@@ -339,7 +375,7 @@ public class SubscriptionCoordinator {
String.format(
"Failed to drop topic %s. Failures: %s does not exist.",
topicName, topicName));
}
- return configManager.getProcedureManager().dropTopic(topicName);
+ return configManager.getProcedureManager().dropTopic(topicName,
req.isTableModel);
}
public TShowTopicResp showTopic(TShowTopicReq req) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
index 12b25159fa9..6e26f20c0d8 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
@@ -218,9 +218,11 @@ public class SubscriptionInfo implements SnapshotProcessor
{
private boolean checkBeforeCreateTopicInternal(TCreateTopicReq
createTopicReq)
throws SubscriptionException {
- validateTopicConfig(new
TopicConfig(safeTopicAttributes(createTopicReq.getTopicAttributes())));
+ final TopicConfig topicConfig =
+ new
TopicConfig(safeTopicAttributes(createTopicReq.getTopicAttributes()));
+ validateTopicConfig(topicConfig);
- if (!isTopicExisted(createTopicReq.getTopicName())) {
+ if (!isTopicExisted(createTopicReq.getTopicName(),
topicConfig.isTableTopic())) {
return true;
}
@@ -237,30 +239,36 @@ public class SubscriptionInfo implements
SnapshotProcessor {
}
public void validateBeforeDroppingTopic(String topicName) throws
SubscriptionException {
+ validateBeforeDroppingTopic(topicName, false);
+ }
+
+ public void validateBeforeDroppingTopic(String topicName, boolean
isTableModel)
+ throws SubscriptionException {
acquireReadLock();
try {
- checkBeforeDropTopicInternal(topicName);
+ checkBeforeDropTopicInternal(topicName, isTableModel);
} finally {
releaseReadLock();
}
}
- private void checkBeforeDropTopicInternal(String topicName) throws
SubscriptionException {
+ private void checkBeforeDropTopicInternal(String topicName, boolean
isTableModel)
+ throws SubscriptionException {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
ConfigNodeMessages.CHECK_BEFORE_DROPPING_TOPIC_TOPIC_EXISTS,
topicName,
- isTopicExisted(topicName));
+ isTopicExisted(topicName, isTableModel));
}
- TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName);
+ TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName,
isTableModel);
if (Objects.isNull(topicMeta)) {
// DO NOTHING HERE!
// No matter whether the topic exists, we allow the drop operation
// executed on all nodes to ensure the consistency.
return;
} else {
- if
(!consumerGroupMetaKeeper.isTopicSubscribedByConsumerGroup(topicName)) {
+ if (!consumerGroupMetaKeeper.isTopicSubscribedByConsumerGroup(topicName,
isTableModel)) {
return;
}
}
@@ -311,8 +319,10 @@ public class SubscriptionInfo implements SnapshotProcessor
{
private void checkBeforeAlteringTopicInternal(TopicMeta topicMeta) throws
SubscriptionException {
validateTopicConfig(topicMeta.getConfig());
- if (isTopicExisted(topicMeta.getTopicName())) {
- final TopicMeta existedTopicMeta =
topicMetaKeeper.getTopicMeta(topicMeta.getTopicName());
+ final boolean isTableModel = topicMeta.visibleUnderTableModel();
+ if (isTopicExisted(topicMeta.getTopicName(), isTableModel)) {
+ final TopicMeta existedTopicMeta =
+ topicMetaKeeper.getTopicMeta(topicMeta.getTopicName(), isTableModel);
validateUnsupportedHotUpdatedTopicConfig(
topicMeta.getTopicName(), existedTopicMeta.getConfig(),
topicMeta.getConfig());
return;
@@ -616,6 +626,15 @@ public class SubscriptionInfo implements SnapshotProcessor
{
}
}
+ public TopicMeta getTopicMeta(String topicName, boolean isTableModel) {
+ acquireReadLock();
+ try {
+ return topicMetaKeeper.getTopicMeta(topicName, isTableModel);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
public Iterable<TopicMeta> getAllTopicMeta() {
acquireReadLock();
try {
@@ -636,12 +655,32 @@ public class SubscriptionInfo implements
SnapshotProcessor {
}
}
+ public TopicMeta deepCopyTopicMeta(String topicName, boolean isTableModel) {
+ acquireReadLock();
+ try {
+ return topicMetaKeeper.containsTopicMeta(topicName, isTableModel)
+ ? topicMetaKeeper.getTopicMeta(topicName, isTableModel).deepCopy()
+ : null;
+ } finally {
+ releaseReadLock();
+ }
+ }
+
public TopicMeta deepCopyTopicMetaWithUpdatedAttributes(
String topicName, Map<String, String> updatedAttributes) {
+ final boolean isTableModel =
+ new TopicConfig(safeTopicAttributes(updatedAttributes)).isTableTopic();
+ return deepCopyTopicMetaWithUpdatedAttributes(topicName,
updatedAttributes, isTableModel);
+ }
+
+ public TopicMeta deepCopyTopicMetaWithUpdatedAttributes(
+ String topicName, Map<String, String> updatedAttributes, boolean
isTableModel) {
acquireReadLock();
try {
- return topicMetaKeeper.containsTopicMeta(topicName)
- ?
topicMetaKeeper.getTopicMeta(topicName).deepCopyWithUpdatedAttributes(updatedAttributes)
+ return topicMetaKeeper.containsTopicMeta(topicName, isTableModel)
+ ? topicMetaKeeper
+ .getTopicMeta(topicName, isTableModel)
+ .deepCopyWithUpdatedAttributes(updatedAttributes)
: null;
} finally {
releaseReadLock();
@@ -657,21 +696,28 @@ public class SubscriptionInfo implements
SnapshotProcessor {
*/
public List<TTopicOwnerLeaseEntry> collectTopicOwnerLeaseEntries(
final Set<String> blockedTopicNames) {
+ return collectTopicOwnerLeaseEntries(blockedTopicNames, blockedTopicNames);
+ }
+
+ public List<TTopicOwnerLeaseEntry> collectTopicOwnerLeaseEntries(
+ final Set<String> blockedTreeTopicNames, final Set<String>
blockedTableTopicNames) {
acquireReadLock();
try {
final List<TTopicOwnerLeaseEntry> entries = new ArrayList<>();
for (final TopicMeta topicMeta : topicMetaKeeper.getAllTopicMeta()) {
if (!topicMeta.isOwnerFencingEnabled()
|| Objects.isNull(topicMeta.getOwnerLeaseDurationMs())
- || blockedTopicNames.contains(topicMeta.getTopicName())) {
+ || (topicMeta.visibleUnderTableModel() ? blockedTableTopicNames :
blockedTreeTopicNames)
+ .contains(topicMeta.getTopicName())) {
continue;
}
entries.add(
new TTopicOwnerLeaseEntry(
- topicMeta.getTopicName(),
- topicMeta.getOwnerId(),
- topicMeta.getOwnerEpoch(),
- topicMeta.getOwnerLeaseDurationMs()));
+ topicMeta.getTopicName(),
+ topicMeta.getOwnerId(),
+ topicMeta.getOwnerEpoch(),
+ topicMeta.getOwnerLeaseDurationMs())
+ .setIsTableModel(topicMeta.visibleUnderTableModel()));
}
return entries;
} finally {
@@ -711,15 +757,17 @@ public class SubscriptionInfo implements
SnapshotProcessor {
}
private TSStatus alterTopicInternal(final AlterTopicPlan plan) {
+ final boolean isTableModel = plan.getTopicMeta().visibleUnderTableModel();
try {
TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(plan.getTopicMeta().getTopicName()),
plan.getTopicMeta());
+ topicMetaKeeper.getTopicMeta(plan.getTopicMeta().getTopicName(),
isTableModel),
+ plan.getTopicMeta());
} catch (final IllegalArgumentException e) {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_OWNER_EPOCH_CONFLICT.getStatusCode())
.setMessage(e.getMessage());
}
- topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName());
+ topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName(),
isTableModel);
topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(),
plan.getTopicMeta());
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
@@ -752,7 +800,11 @@ public class SubscriptionInfo implements SnapshotProcessor
{
public TSStatus dropTopic(DropTopicPlan plan) {
acquireWriteLock();
try {
- topicMetaKeeper.removeTopicMeta(plan.getTopicName());
+ if (plan.isTableModelSet()) {
+ topicMetaKeeper.removeTopicMeta(plan.getTopicName(),
plan.isTableModel());
+ } else {
+ topicMetaKeeper.removeTopicMeta(plan.getTopicName());
+ }
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} finally {
releaseWriteLock();
@@ -1087,7 +1139,8 @@ public class SubscriptionInfo implements
SnapshotProcessor {
continue;
}
for (String consumerGroupId :
-
consumerGroupMetaKeeper.getSubscribedConsumerGroupIds(topicMeta.getTopicName()))
{
+ consumerGroupMetaKeeper.getSubscribedConsumerGroupIds(
+ topicMeta.getTopicName(), topicMeta.visibleUnderTableModel())) {
Set<String> subscribedConsumerIDs =
consumerGroupMetaKeeper.getConsumersSubscribingTopic(
consumerGroupId, topicMeta.getTopicName());
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index 5c020ebf2ec..cf1883b447f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -102,6 +102,7 @@ import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@@ -768,10 +769,21 @@ public class ConfigNodeProcedureEnv {
}
public List<TSStatus> dropSingleTopicOnDataNode(String topicNameToDrop) {
+ return dropSingleTopicOnDataNode(topicNameToDrop, null);
+ }
+
+ public List<TSStatus> dropSingleTopicOnDataNode(String topicNameToDrop,
boolean isTableModel) {
+ return dropSingleTopicOnDataNode(topicNameToDrop,
Boolean.valueOf(isTableModel));
+ }
+
+ private List<TSStatus> dropSingleTopicOnDataNode(String topicNameToDrop,
Boolean isTableModel) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushSingleTopicMetaReq request =
new TPushSingleTopicMetaReq().setTopicNameToDrop(topicNameToDrop);
+ if (Objects.nonNull(isTableModel)) {
+ request.setIsTableModel(isTableModel);
+ }
final DataNodeAsyncRequestContext<TPushSingleTopicMetaReq,
TPushTopicMetaResp> clientHandler =
new DataNodeAsyncRequestContext<>(
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java
index a81661bff2f..04e29edbb94 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java
@@ -115,7 +115,8 @@ public class CreateSubscriptionProcedure extends
AbstractOperateSubscriptionAndP
// Construct CreatePipeProcedureV2s (for non-consensus topics)
for (final String topicName : subscribeReq.getTopicNames()) {
- final TopicMeta topicMeta =
subscriptionInfo.get().deepCopyTopicMeta(topicName);
+ final TopicMeta topicMeta =
+ subscriptionInfo.get().deepCopyTopicMeta(topicName,
subscribeReq.isTableModel);
final String topicMode = topicMeta.getConfig().getMode();
final boolean isConsensusBasedTopic =
topicMeta.getConfig().isIncrementalMode();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
index 321ebc9bb96..b4c16985aef 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
@@ -106,7 +106,8 @@ public class DropSubscriptionProcedure extends
AbstractOperateSubscriptionAndPip
for (final String topic : unsubscribeReq.getTopicNames()) {
if (topicsUnsubByGroup.contains(topic)) {
- final TopicMeta topicMeta =
subscriptionInfo.get().deepCopyTopicMeta(topic);
+ final TopicMeta topicMeta =
+ subscriptionInfo.get().deepCopyTopicMeta(topic,
unsubscribeReq.isTableModel);
final String topicMode = topicMeta.getConfig().getMode();
final boolean isConsensusBasedTopic =
topicMeta.getConfig().isIncrementalMode();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
index 6e9c94dcca4..1f6cc40cb09 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
@@ -124,7 +124,11 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
public boolean executeFromValidate(ConfigNodeProcedureEnv env) throws
SubscriptionException {
LOGGER.info(ProcedureMessages.ALTERTOPICPROCEDURE_EXECUTEFROMVALIDATE);
- existedTopicMeta =
subscriptionInfo.get().deepCopyTopicMeta(updatedTopicMeta.getTopicName());
+ existedTopicMeta =
+ subscriptionInfo
+ .get()
+ .deepCopyTopicMeta(
+ updatedTopicMeta.getTopicName(),
updatedTopicMeta.visibleUnderTableModel());
if (Objects.nonNull(updatedTopicAttributes) &&
Objects.nonNull(existedTopicMeta)) {
updatedTopicMeta =
existedTopicMeta.deepCopyWithUpdatedAttributes(updatedTopicAttributes);
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/CreateTopicProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/CreateTopicProcedure.java
index 08ae98a5780..e525ce97479 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/CreateTopicProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/CreateTopicProcedure.java
@@ -137,7 +137,8 @@ public class CreateTopicProcedure extends
AbstractOperateSubscriptionProcedure {
response =
env.getConfigManager()
.getConsensusManager()
- .write(new DropTopicPlan(topicMeta.getTopicName()));
+ .write(
+ new DropTopicPlan(topicMeta.getTopicName(),
topicMeta.visibleUnderTableModel()));
} catch (ConsensusException e) {
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE,
e);
response =
@@ -157,7 +158,8 @@ public class CreateTopicProcedure extends
AbstractOperateSubscriptionProcedure {
throws SubscriptionException {
LOGGER.info(ProcedureMessages.CREATETOPICPROCEDURE_ROLLBACKFROMCREATEONDATANODES,
topicMeta);
- final List<TSStatus> statuses =
env.dropSingleTopicOnDataNode(topicMeta.getTopicName());
+ final List<TSStatus> statuses =
+ env.dropSingleTopicOnDataNode(topicMeta.getTopicName(),
topicMeta.visibleUnderTableModel());
if (RpcUtils.squashResponseStatusList(statuses).getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
// throw exception instead of logging warn, do not rely on metadata
synchronization
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
index 8a17f99045a..8a95811fce5 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
@@ -47,6 +47,8 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
private static final Logger LOGGER =
LoggerFactory.getLogger(DropTopicProcedure.class);
private String topicName;
+ private boolean isTableModel;
+ private boolean isTableModelSet;
public DropTopicProcedure() {
super();
@@ -57,6 +59,12 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
this.topicName = topicName;
}
+ public DropTopicProcedure(String topicName, boolean isTableModel) {
+ this(topicName);
+ this.isTableModel = isTableModel;
+ this.isTableModelSet = true;
+ }
+
@Override
protected SubscriptionOperation getOperation() {
return SubscriptionOperation.DROP_TOPIC;
@@ -66,7 +74,11 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
protected boolean executeFromValidate(ConfigNodeProcedureEnv env) throws
SubscriptionException {
LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_EXECUTEFROMVALIDATE,
topicName);
- subscriptionInfo.get().validateBeforeDroppingTopic(topicName);
+ if (isTableModelSet) {
+ subscriptionInfo.get().validateBeforeDroppingTopic(topicName,
isTableModel);
+ } else {
+ subscriptionInfo.get().validateBeforeDroppingTopic(topicName);
+ }
return true;
}
@@ -77,7 +89,13 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
TSStatus response;
try {
- response = env.getConfigManager().getConsensusManager().write(new
DropTopicPlan(topicName));
+ response =
+ env.getConfigManager()
+ .getConsensusManager()
+ .write(
+ isTableModelSet
+ ? new DropTopicPlan(topicName, isTableModel)
+ : new DropTopicPlan(topicName));
} catch (ConsensusException e) {
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE,
e);
response =
@@ -94,7 +112,10 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_EXECUTEFROMOPERATEONDATANODES,
topicName);
- final List<TSStatus> statuses = env.dropSingleTopicOnDataNode(topicName);
+ final List<TSStatus> statuses =
+ isTableModelSet
+ ? env.dropSingleTopicOnDataNode(topicName, isTableModel)
+ : env.dropSingleTopicOnDataNode(topicName);
if (RpcUtils.squashResponseStatusList(statuses).getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
// throw exception instead of logging warn, do not rely on metadata
synchronization
@@ -125,6 +146,9 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
super.serialize(stream);
ReadWriteIOUtils.write(topicName, stream);
+ if (isTableModelSet) {
+ ReadWriteIOUtils.write(isTableModel, stream);
+ }
}
@Override
@@ -132,6 +156,10 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
super.deserialize(byteBuffer);
topicName = ReadWriteIOUtils.readString(byteBuffer);
+ isTableModelSet = byteBuffer.hasRemaining();
+ if (isTableModelSet) {
+ isTableModel = ReadWriteIOUtils.readBool(byteBuffer);
+ }
}
@Override
@@ -146,11 +174,14 @@ public class DropTopicProcedure extends
AbstractOperateSubscriptionProcedure {
return Objects.equals(getProcId(), that.getProcId())
&& Objects.equals(getCurrentState(), that.getCurrentState())
&& getCycles() == that.getCycles()
- && Objects.equals(topicName, that.topicName);
+ && Objects.equals(topicName, that.topicName)
+ && isTableModel == that.isTableModel
+ && isTableModelSet == that.isTableModelSet;
}
@Override
public int hashCode() {
- return Objects.hash(getProcId(), getCurrentState(), getCycles(),
topicName);
+ return Objects.hash(
+ getProcId(), getCurrentState(), getCycles(), topicName, isTableModel,
isTableModelSet);
}
}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index 35537a2cea2..0e858589803 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -1294,7 +1294,13 @@ public class ConfigPhysicalPlanSerDeTest {
final DropTopicPlan dropTopicPlan = new DropTopicPlan("test_topic");
final DropTopicPlan dropTopicPlan1 =
(DropTopicPlan)
ConfigPhysicalPlan.Factory.create(dropTopicPlan.serializeToByteBuffer());
- Assert.assertEquals(dropTopicPlan.getTopicName(),
dropTopicPlan1.getTopicName());
+ Assert.assertEquals(dropTopicPlan, dropTopicPlan1);
+
+ final DropTopicPlan tableDropTopicPlan = new DropTopicPlan("test_topic",
true);
+ final DropTopicPlan tableDropTopicPlan1 =
+ (DropTopicPlan)
+
ConfigPhysicalPlan.Factory.create(tableDropTopicPlan.serializeToByteBuffer());
+ Assert.assertEquals(tableDropTopicPlan, tableDropTopicPlan1);
}
@Test
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
index 12b1b3dd2ae..6bd896fd2f6 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
@@ -52,33 +52,33 @@ public class SubscriptionCoordinatorTest {
final CountDownLatch sameTopicLockAcquired = new CountDownLatch(1);
boolean topic1LockHeld = true;
- coordinator.lockTopicAlteration("topic1");
+ coordinator.lockTopicAlteration("topic1", false);
try {
final Future<?> sameTopicAlteration =
executor.submit(
() -> {
sameTopicAttemptStarted.countDown();
- coordinator.lockTopicAlteration("topic1");
+ coordinator.lockTopicAlteration("topic1", false);
try {
sameTopicLockAcquired.countDown();
} finally {
- coordinator.unlockTopicAlteration("topic1");
+ coordinator.unlockTopicAlteration("topic1", false);
}
});
Assert.assertTrue(sameTopicAttemptStarted.await(5, TimeUnit.SECONDS));
Assert.assertFalse(sameTopicLockAcquired.await(100,
TimeUnit.MILLISECONDS));
- coordinator.lockTopicAlteration("topic2");
- coordinator.unlockTopicAlteration("topic2");
+ coordinator.lockTopicAlteration("topic1", true);
+ coordinator.unlockTopicAlteration("topic1", true);
- coordinator.unlockTopicAlteration("topic1");
+ coordinator.unlockTopicAlteration("topic1", false);
topic1LockHeld = false;
Assert.assertTrue(sameTopicLockAcquired.await(5, TimeUnit.SECONDS));
sameTopicAlteration.get(5, TimeUnit.SECONDS);
} finally {
if (topic1LockHeld) {
- coordinator.unlockTopicAlteration("topic1");
+ coordinator.unlockTopicAlteration("topic1", false);
}
executor.shutdownNow();
}
@@ -108,8 +108,11 @@ public class SubscriptionCoordinatorTest {
new SubscriptionCoordinator(Mockito.mock(ConfigManager.class),
subscriptionInfo) {
@Override
void waitForOwnerLeaseExpiration(
- final String waitingTopicName, final long
waitingLeaseDurationMs) {
+ final String waitingTopicName,
+ final boolean isTableModel,
+ final long waitingLeaseDurationMs) {
Assert.assertEquals(topicName, waitingTopicName);
+ Assert.assertFalse(isTableModel);
Assert.assertEquals(ownerLeaseDurationMs, waitingLeaseDurationMs);
final Map<String, String> updatedAttributes = new HashMap<>();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
index f87afe129e5..170bbe2d3e1 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.confignode.persistence.subscription;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import
org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
@@ -27,6 +28,7 @@ import
org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
import
org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan;
import
org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan;
import
org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.subscription.topic.DropTopicPlan;
import
org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp;
@@ -47,6 +49,39 @@ import java.util.UUID;
public class SubscriptionInfoTest {
+ @Test
+ public void testSameNameTreeAndTableTopicsCanBeManagedIndependently() {
+ final String topicName = "same-name-topic";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final TopicMeta treeTopicMeta =
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE));
+ final TopicMeta tableTopicMeta =
+ new TopicMeta(
+ topicName,
+ 2L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new
CreateTopicPlan(treeTopicMeta)).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new
CreateTopicPlan(tableTopicMeta)).getCode());
+ Assert.assertSame(treeTopicMeta, subscriptionInfo.getTopicMeta(topicName,
false));
+ Assert.assertSame(tableTopicMeta, subscriptionInfo.getTopicMeta(topicName,
true));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.dropTopic(new DropTopicPlan(topicName,
false)).getCode());
+ Assert.assertFalse(subscriptionInfo.isTopicExisted(topicName, false));
+ Assert.assertTrue(subscriptionInfo.isTopicExisted(topicName, true));
+ }
+
@Test
public void testAlterTopicRejectsOwnerEpochRollback() {
final String topicName = "topic-" + UUID.randomUUID();
@@ -131,6 +166,37 @@ public class SubscriptionInfoTest {
Assert.assertEquals(ownerLeaseDurationMs, entry.getLeaseRemainingMs());
}
+ @Test
+ public void testCollectTopicOwnerLeaseEntriesIsolatedByModel() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final long ownerLeaseDurationMs = 60000;
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final TopicMeta treeTopicMeta =
+ createTopicMeta(topicName, "tree-owner", 5L, ownerLeaseDurationMs);
+ final TopicMeta tableTopicMeta =
+ createTopicMeta(topicName, "table-owner", 6L, ownerLeaseDurationMs);
+ tableTopicMeta
+ .getConfig()
+ .getAttribute()
+ .put(SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new
CreateTopicPlan(treeTopicMeta)).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new
CreateTopicPlan(tableTopicMeta)).getCode());
+
+ final List<TTopicOwnerLeaseEntry> entries =
+ subscriptionInfo.collectTopicOwnerLeaseEntries(
+ Collections.singleton(topicName), Collections.emptySet());
+
+ Assert.assertEquals(1, entries.size());
+ Assert.assertEquals(topicName, entries.get(0).getTopicName());
+ Assert.assertEquals("table-owner", entries.get(0).getOwnerId());
+ Assert.assertTrue(entries.get(0).isIsTableModel());
+ }
+
@Test
public void testAlterTopicOwnerAndShowTopicOwner() {
final String topicName = "topic-" + UUID.randomUUID();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java
index fc4352a7a00..741e3e344be 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java
@@ -165,7 +165,7 @@ public class CreateSubscriptionProcedureTest {
Mockito.when(
subscriptionInfo.isTopicSubscribedByConsumerGroup("test_topic",
"test_consumer_group"))
.thenReturn(false);
-
Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic")).thenReturn(topicMeta);
+ Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic",
false)).thenReturn(topicMeta);
final PipeTaskInfo pipeTaskInfo = Mockito.mock(PipeTaskInfo.class);
Mockito.when(pipeTaskInfo.checkBeforeCreatePipe(Mockito.any(TCreatePipeReq.class)))
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
index 6536c040f10..1bf74f8db24 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
@@ -145,7 +145,7 @@ public class DropSubscriptionProcedureTest {
.thenReturn(consumerGroupMeta);
Mockito.when(subscriptionInfo.deepCopyConsumerGroupMeta("test_consumer_group"))
.thenAnswer(invocation -> consumerGroupMeta.deepCopy());
- Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic"))
+ Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic", false))
.thenReturn(new TopicMeta("test_topic", 1, Collections.emptyMap()));
final PipeTaskInfo pipeTaskInfo = Mockito.mock(PipeTaskInfo.class);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java
index 9b9e2fc9538..0ad614408a3 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java
@@ -49,4 +49,23 @@ public class DropTopicProcedureTest {
fail();
}
}
+
+ @Test
+ public void serializeDeserializeTableModelTest() {
+ PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream);
+
+ DropTopicProcedure proc = new DropTopicProcedure("test_topic", true);
+
+ try {
+ proc.serialize(outputStream);
+ ByteBuffer buffer =
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
+ DropTopicProcedure proc2 = (DropTopicProcedure)
ProcedureFactory.getInstance().create(buffer);
+
+ assertEquals(proc, proc2);
+ } catch (Exception e) {
+ fail();
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 28d75ba4f49..a06d10413fd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -1449,7 +1449,11 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
try {
final TPushTopicMetaRespExceptionMessage exceptionMessage;
if (req.isSetTopicNameToDrop()) {
- exceptionMessage =
SubscriptionAgent.topic().handleDropTopic(req.getTopicNameToDrop());
+ exceptionMessage =
+ req.isSetIsTableModel()
+ ? SubscriptionAgent.topic()
+ .handleDropTopic(req.getTopicNameToDrop(),
req.isIsTableModel())
+ :
SubscriptionAgent.topic().handleDropTopic(req.getTopicNameToDrop());
} else if (req.isSetTopicMeta()) {
exceptionMessage =
SubscriptionAgent.topic()
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 019253774d5..da649e75b63 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
@@ -105,7 +105,8 @@ public class SubscriptionBrokerAgent {
final Map<String, TopicProgress> progressByTopic) {
final String consumerGroupId = consumerConfig.getConsumerGroupId();
final String consumerId = consumerConfig.getConsumerId();
- final List<String> unsupportedConsensusTopics =
getUnsupportedConsensusTopics(topicNames);
+ final List<String> unsupportedConsensusTopics =
+ getUnsupportedConsensusTopics(consumerGroupId, topicNames);
if (!unsupportedConsensusTopics.isEmpty()) {
final String errorMessage =
buildUnsupportedConsensusRuntimeMessage(
@@ -238,7 +239,8 @@ public class SubscriptionBrokerAgent {
continue;
}
final String topicName = commitContext.getTopicName();
- if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName))
{
+ if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName, isTableModel(consumerConfig))) {
continue;
}
final String regionId = commitContext.getRegionId();
@@ -347,7 +349,8 @@ public class SubscriptionBrokerAgent {
private ConsensusSubscriptionBroker getConsensusBrokerForSeekOrNoOp(
final String consumerGroupId, final String topicName, final String
operation) {
- if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName)) {
+ if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName,
SubscriptionAgent.consumer().isTableModel(consumerGroupId))) {
final String errorMessage =
String.format(
"Subscription: %s is only supported for consensus-based
subscriptions, "
@@ -394,20 +397,26 @@ public class SubscriptionBrokerAgent {
}
}
- private List<String> getUnsupportedConsensusTopics(final Set<String>
topicNames) {
+ private List<String> getUnsupportedConsensusTopics(
+ final String consumerGroupId, final Set<String> topicNames) {
if (DataRegionConsensusImpl.getInstance() instanceof IoTConsensus) {
return Collections.emptyList();
}
final List<String> unsupportedConsensusTopics = new ArrayList<>();
for (final String topicName : topicNames) {
- if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName)) {
+ if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName,
SubscriptionAgent.consumer().isTableModel(consumerGroupId))) {
unsupportedConsensusTopics.add(topicName);
}
}
return unsupportedConsensusTopics;
}
+ private static boolean isTableModel(final ConsumerConfig consumerConfig) {
+ return
SubscriptionAgent.consumer().isTableModel(consumerConfig.getConsumerGroupId());
+ }
+
private String buildUnsupportedConsensusRuntimeMessage(
final String consumerGroupId, final String topicName, final String
operation) {
return buildUnsupportedConsensusRuntimeMessage(
@@ -656,14 +665,23 @@ public class SubscriptionBrokerAgent {
return queue;
}
- public void refreshConsensusQueueOrderMode(final String topicName, final
String orderMode) {
+ public void refreshConsensusQueueOrderMode(
+ final String topicName, final boolean isTableModel, final String
orderMode) {
LOGGER.info(
DataNodePipeMessages
.PIPE_LOG_SUBSCRIPTIONBROKERAGENT_REFRESHING_CONSENSUS_QUEUE_ORDER_1886704D,
topicName,
orderMode);
- for (final ConsensusSubscriptionBroker broker :
getBrokers(ConsensusSubscriptionBroker.class)) {
- broker.refreshConsensusQueueOrderMode(topicName, orderMode);
+ for (final Map.Entry<String, List<ISubscriptionBroker>> entry :
+ consumerGroupIdToBrokers.entrySet()) {
+ if (SubscriptionAgent.consumer().isTableModel(entry.getKey()) !=
isTableModel) {
+ continue;
+ }
+ final ConsensusSubscriptionBroker broker =
+ getBroker(entry.getValue(), ConsensusSubscriptionBroker.class);
+ if (Objects.nonNull(broker)) {
+ broker.refreshConsensusQueueOrderMode(topicName, orderMode);
+ }
}
}
@@ -691,13 +709,24 @@ public class SubscriptionBrokerAgent {
}
public ColumnFilterMatcher getColumnFilterMatcher(final String topicName) {
+ return getColumnFilterMatcher(topicName, true);
+ }
+
+ public ColumnFilterMatcher getColumnFilterMatcher(
+ final String topicName, final boolean isTableModel) {
+ if (!isTableModel) {
+ return ColumnFilterMatcher.matchAll();
+ }
+
final ColumnFilterMatcher matcher =
topicNameToColumnFilterMatcher.get(topicName);
if (Objects.nonNull(matcher)) {
return matcher;
}
final TopicConfig topicConfig =
-
SubscriptionAgent.topic().getTopicConfigs(Collections.singleton(topicName)).get(topicName);
+ SubscriptionAgent.topic()
+ .getTopicConfigs(Collections.singleton(topicName), true)
+ .get(topicName);
if (Objects.isNull(topicConfig)) {
return ColumnFilterMatcher.matchAll();
}
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 22da7a6336b..a268f515a6a 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
@@ -103,7 +103,9 @@ public class SubscriptionConsumerAgent {
if (Objects.isNull(metaInAgent)) {
SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId);
ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
- consumerGroupId, metaFromCoordinator.getSubscribedTopicNames());
+ consumerGroupId,
+ metaFromCoordinator.getSubscribedTopicNames(),
+ metaFromCoordinator.visibleUnder(true));
consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId,
metaFromCoordinator);
return;
}
@@ -128,7 +130,9 @@ public class SubscriptionConsumerAgent {
}
ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions(
- consumerGroupId, metaFromCoordinator.getSubscribedTopicNames());
+ consumerGroupId,
+ metaFromCoordinator.getSubscribedTopicNames(),
+ metaFromCoordinator.visibleUnder(true));
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId,
metaFromCoordinator);
// no need to create broker manually
@@ -141,7 +145,8 @@ public class SubscriptionConsumerAgent {
final Set<String> pipeTopicsUnsubByGroup = new LinkedHashSet<>();
final Set<String> consensusTopicsUnsubByGroup = new LinkedHashSet<>();
for (final String topicName : topicsUnsubByGroup) {
- if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName)) {
+ if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName, metaFromCoordinator.visibleUnder(true))) {
consensusTopicsUnsubByGroup.add(topicName);
continue;
}
@@ -277,6 +282,17 @@ public class SubscriptionConsumerAgent {
}
}
+ public boolean isTableModel(final String consumerGroupId) {
+ acquireReadLock();
+ try {
+ final ConsumerGroupMeta consumerGroupMeta =
+ consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId);
+ return Objects.nonNull(consumerGroupMeta) &&
consumerGroupMeta.visibleUnder(true);
+ } 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/agent/SubscriptionTopicAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
index 43475e79afe..3f69633f499 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.subscription.agent;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
@@ -98,19 +99,22 @@ public class SubscriptionTopicAgent {
private void handleSingleTopicMetaChangesInternal(final TopicMeta
metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
- final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName);
+ final boolean isTableModel = metaFromCoordinator.visibleUnderTableModel();
+ final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName,
isTableModel);
TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator);
- topicMetaKeeper.removeTopicMeta(topicName);
+ topicMetaKeeper.removeTopicMeta(topicName, isTableModel);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) {
SubscriptionAgent.broker().refreshColumnFilter(topicName,
metaFromCoordinator.getConfig());
- } else if (!metaFromCoordinator.getConfig().isTableTopic()) {
+ } else if (!metaFromCoordinator.getConfig().isTableTopic()
+ && !topicMetaKeeper.containsTopicMeta(topicName, true)) {
// ConfigNode rejects column-filter on tree topics. Drop defensively in
case stale or replayed
// topic metadata reaches this DataNode after a table-topic to
tree-topic transition.
SubscriptionAgent.broker().dropColumnFilter(topicName);
}
SubscriptionAgent.broker()
- .refreshConsensusQueueOrderMode(topicName,
metaFromCoordinator.getConfig().getOrderMode());
+ .refreshConsensusQueueOrderMode(
+ topicName, isTableModel,
metaFromCoordinator.getConfig().getOrderMode());
}
static boolean shouldRefreshColumnFilter(final TopicMeta oldMeta, final
TopicMeta newMeta) {
@@ -185,9 +189,19 @@ public class SubscriptionTopicAgent {
}
public TPushTopicMetaRespExceptionMessage handleDropTopic(final String
topicName) {
+ return handleDropTopic(topicName, null);
+ }
+
+ public TPushTopicMetaRespExceptionMessage handleDropTopic(
+ final String topicName, final boolean isTableModel) {
+ return handleDropTopic(topicName, Boolean.valueOf(isTableModel));
+ }
+
+ private TPushTopicMetaRespExceptionMessage handleDropTopic(
+ final String topicName, final Boolean isTableModel) {
acquireWriteLock();
try {
- handleDropTopicInternal(topicName);
+ handleDropTopicInternal(topicName, isTableModel);
return null;
} catch (final Exception e) {
LOGGER.warn(DataNodeMiscMessages.EXCEPTION_DROPPING_TOPIC, topicName, e);
@@ -200,9 +214,19 @@ public class SubscriptionTopicAgent {
}
}
- private void handleDropTopicInternal(final String topicName) {
- topicMetaKeeper.removeTopicMeta(topicName);
- SubscriptionAgent.broker().dropColumnFilter(topicName);
+ private void handleDropTopicInternal(final String topicName, final Boolean
isTableModel) {
+ final TopicMeta topicMeta =
+ Objects.isNull(isTableModel)
+ ? topicMetaKeeper.getTopicMeta(topicName)
+ : topicMetaKeeper.getTopicMeta(topicName, isTableModel);
+ if (Objects.isNull(isTableModel)) {
+ topicMetaKeeper.removeTopicMeta(topicName);
+ } else {
+ topicMetaKeeper.removeTopicMeta(topicName, isTableModel);
+ }
+ if (Objects.nonNull(topicMeta) && topicMeta.visibleUnderTableModel()) {
+ SubscriptionAgent.broker().dropColumnFilter(topicName);
+ }
}
public boolean isTopicExisted(final String topicName) {
@@ -215,11 +239,15 @@ public class SubscriptionTopicAgent {
}
public String getTopicFormat(final String topicName) {
+ return getTopicFormat(topicName, false);
+ }
+
+ public String getTopicFormat(final String topicName, final boolean
isTableModel) {
acquireReadLock();
try {
- return topicMetaKeeper.containsTopicMeta(topicName)
+ return topicMetaKeeper.containsTopicMeta(topicName, isTableModel)
? topicMetaKeeper
- .getTopicMeta(topicName)
+ .getTopicMeta(topicName, isTableModel)
.getConfig()
.getStringOrDefault(TopicConstant.FORMAT_KEY,
TopicConstant.FORMAT_DEFAULT_VALUE)
: null;
@@ -229,10 +257,14 @@ public class SubscriptionTopicAgent {
}
public String getTopicMode(final String topicName) {
+ return getTopicMode(topicName, false);
+ }
+
+ public String getTopicMode(final String topicName, final boolean
isTableModel) {
acquireReadLock();
try {
- return topicMetaKeeper.containsTopicMeta(topicName)
- ? topicMetaKeeper.getTopicMeta(topicName).getConfig().getMode()
+ return topicMetaKeeper.containsTopicMeta(topicName, isTableModel)
+ ? topicMetaKeeper.getTopicMeta(topicName,
isTableModel).getConfig().getMode()
: null;
} finally {
releaseReadLock();
@@ -240,23 +272,32 @@ public class SubscriptionTopicAgent {
}
public String getTopicOrderMode(final String topicName) {
+ return getTopicOrderMode(topicName, false);
+ }
+
+ public String getTopicOrderMode(final String topicName, final boolean
isTableModel) {
acquireReadLock();
try {
- return
topicMetaKeeper.getTopicMeta(topicName).getConfig().getOrderMode();
+ return topicMetaKeeper.getTopicMeta(topicName,
isTableModel).getConfig().getOrderMode();
} finally {
releaseReadLock();
}
}
public Map<String, TopicConfig> getTopicConfigs(final Set<String>
topicNames) {
+ return getTopicConfigs(topicNames, false);
+ }
+
+ public Map<String, TopicConfig> getTopicConfigs(
+ final Set<String> topicNames, final boolean isTableModel) {
acquireReadLock();
try {
return topicNames.stream()
- .filter(topicMetaKeeper::containsTopicMeta)
+ .filter(topicName -> topicMetaKeeper.containsTopicMeta(topicName,
isTableModel))
.collect(
Collectors.toMap(
topicName -> topicName,
- topicName ->
topicMetaKeeper.getTopicMeta(topicName).getConfig()));
+ topicName -> topicMetaKeeper.getTopicMeta(topicName,
isTableModel).getConfig()));
} finally {
releaseReadLock();
}
@@ -265,7 +306,8 @@ public class SubscriptionTopicAgent {
public TSStatus checkTopicOwner(final ConsumerConfig consumerConfig, final
String topicName) {
acquireReadLock();
try {
- final TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName);
+ final TopicMeta topicMeta =
+ topicMetaKeeper.getTopicMeta(topicName,
isTableModel(consumerConfig));
if (Objects.isNull(topicMeta) || !topicMeta.isOwnerFencingEnabled()) {
return RpcUtils.SUCCESS_STATUS;
}
@@ -340,7 +382,10 @@ public class SubscriptionTopicAgent {
acquireWriteLock();
try {
for (final TTopicOwnerLeaseEntry lease : ownerLeases) {
- final TopicMeta topicMeta =
topicMetaKeeper.getTopicMeta(lease.getTopicName());
+ final TopicMeta topicMeta =
+ lease.isSetIsTableModel()
+ ? topicMetaKeeper.getTopicMeta(lease.getTopicName(),
lease.isIsTableModel())
+ : topicMetaKeeper.getTopicMeta(lease.getTopicName());
if (Objects.isNull(topicMeta)) {
continue;
}
@@ -351,4 +396,8 @@ public class SubscriptionTopicAgent {
releaseWriteLock();
}
}
+
+ private static boolean isTableModel(final ConsumerConfig consumerConfig) {
+ return
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(consumerConfig.getSqlDialect());
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
index dac6bc5261c..1115c465c15 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
@@ -102,7 +102,7 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
@Override
public boolean acceptsTopic(final String topicName) {
return Objects.nonNull(topicName)
- && !ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName);
+ && !ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName,
isTableModel());
}
//////////////////////////// provided for SubscriptionBrokerAgent
////////////////////////////
@@ -436,7 +436,7 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
brokerId);
return;
}
- final String topicFormat =
SubscriptionAgent.topic().getTopicFormat(topicName);
+ final String topicFormat =
SubscriptionAgent.topic().getTopicFormat(topicName, isTableModel());
final SubscriptionPrefetchingQueue prefetchingQueue;
if (TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equals(topicFormat)) {
prefetchingQueue =
@@ -465,7 +465,7 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
public void updateCompletedTopicNames(final String topicName) {
// mark topic name completed only for topic of snapshot mode
if (SubscriptionAgent.topic()
- .getTopicMode(topicName)
+ .getTopicMode(topicName, isTableModel())
.equals(TopicConstant.MODE_SNAPSHOT_VALUE)) {
completedTopicNames.put(topicName, topicName);
}
@@ -487,7 +487,9 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
prefetchingQueue.markClosed();
// mark topic name completed only for topic of snapshot mode
- if
(SubscriptionAgent.topic().getTopicMode(topicName).equals(TopicConstant.MODE_SNAPSHOT_VALUE)
+ if (SubscriptionAgent.topic()
+ .getTopicMode(topicName, isTableModel())
+ .equals(TopicConstant.MODE_SNAPSHOT_VALUE)
&& prefetchingQueue.isCompleted()) {
completedTopicNames.put(topicName, topicName);
}
@@ -508,6 +510,10 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
brokerId);
}
+ private boolean isTableModel() {
+ return SubscriptionAgent.consumer().isTableModel(brokerId);
+ }
+
@Override
public void unbind(final String topicName) {
unbindPrefetchingQueue(topicName);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
index e069b6a51cb..0bfccd3a78b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
@@ -153,6 +153,10 @@ public abstract class SubscriptionPrefetchingQueue {
return topicName;
}
+ public String getConsumerGroupId() {
+ return brokerId;
+ }
+
protected void cleanUpInternal() {
// clean up events in batches
batches.cleanUp();
@@ -675,7 +679,10 @@ public abstract class SubscriptionPrefetchingQueue {
private boolean canPassThroughTsFile(final PipeTsFileInsertionEvent event) {
return PipeEventCollector.canSkipParsing4TsFileEvent(event)
&& (!event.isTableModelEvent()
- ||
SubscriptionAgent.broker().getColumnFilterMatcher(topicName).isMatchAll());
+ || SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ topicName,
SubscriptionAgent.consumer().isTableModel(brokerId))
+ .isMatchAll());
}
private RetryableState onRetryableTabletInsertionEvent(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
index 317f1f61ee9..0b27de77b12 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
@@ -677,7 +677,7 @@ public class ConsensusLogToTabletConverter {
private ColumnFilterMatcher getColumnFilterMatcher() {
return Objects.nonNull(topicName)
- ? SubscriptionAgent.broker().getColumnFilterMatcher(topicName)
+ ? SubscriptionAgent.broker().getColumnFilterMatcher(topicName, true)
: fallbackColumnFilterMatcher;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index 879346dad65..f535328cdca 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -2078,7 +2078,10 @@ public class ConsensusPrefetchingQueue {
SubscriptionPollResponseType.TABLETS.getType(),
payload,
commitContext,
-
SubscriptionAgent.broker().getColumnFilterMatcher(topicName).isTimeSelected(),
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ topicName,
SubscriptionAgent.consumer().isTableModel(consumerGroupId))
+ .isTimeSelected(),
getTimeSelectedByTable(converter.getDatabaseName(), tablets));
// Install the ownership record before exposing the event to concurrent
poll/ack threads.
@@ -2104,7 +2107,9 @@ public class ConsensusPrefetchingQueue {
return Collections.emptyMap();
}
final ColumnFilterMatcher matcher =
- SubscriptionAgent.broker().getColumnFilterMatcher(topicName);
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ topicName,
SubscriptionAgent.consumer().isTableModel(consumerGroupId));
final Map<String, Boolean> tableMap = new HashMap<>();
for (final Tablet tablet : tablets) {
if (Objects.nonNull(tablet) && Objects.nonNull(tablet.getTableName())) {
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 21fbd317241..bb512922248 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
@@ -159,13 +159,15 @@ public class ConsensusSubscriptionSetupHandler {
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)) {
+ if (!isConsensusBasedTopic(topicName, isTableModel)) {
continue;
}
try {
final Map<String, TopicConfig> topicConfigs =
-
SubscriptionAgent.topic().getTopicConfigs(java.util.Collections.singleton(topicName));
+ SubscriptionAgent.topic()
+ .getTopicConfigs(java.util.Collections.singleton(topicName),
isTableModel);
final TopicConfig topicConfig = topicConfigs.get(topicName);
if (topicConfig == null) {
continue;
@@ -292,8 +294,12 @@ public class ConsensusSubscriptionSetupHandler {
}
public static boolean isConsensusBasedTopic(final String topicName) {
+ return isConsensusBasedTopic(topicName, false);
+ }
+
+ public static boolean isConsensusBasedTopic(final String topicName, final
boolean isTableModel) {
try {
- final String topicMode =
SubscriptionAgent.topic().getTopicMode(topicName);
+ final String topicMode =
SubscriptionAgent.topic().getTopicMode(topicName, isTableModel);
final boolean result =
TopicConstant.MODE_INCREMENTAL_VALUE.equalsIgnoreCase(topicMode);
LOGGER.debug(
DataNodePipeMessages.PIPE_LOG_ISCONSENSUSBASEDTOPIC_CHECK_FOR_TOPIC_MODE_RESULT_19EFA0F9,
@@ -311,8 +317,9 @@ public class ConsensusSubscriptionSetupHandler {
}
}
- private static boolean isConsensusBasedTopicRequired(final String topicName)
{
- final String topicMode = SubscriptionAgent.topic().getTopicMode(topicName);
+ private static boolean isConsensusBasedTopicRequired(
+ final String topicName, final boolean isTableModel) {
+ final String topicMode = SubscriptionAgent.topic().getTopicMode(topicName,
isTableModel);
if (Objects.isNull(topicMode)) {
throw new SubscriptionException(
String.format(
@@ -325,6 +332,12 @@ public class ConsensusSubscriptionSetupHandler {
public static void setupConsensusSubscriptions(
final String consumerGroupId, final Set<String> topicNames) {
+ setupConsensusSubscriptions(
+ consumerGroupId, topicNames,
SubscriptionAgent.consumer().isTableModel(consumerGroupId));
+ }
+
+ public static void setupConsensusSubscriptions(
+ 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();
@@ -359,9 +372,10 @@ public class ConsensusSubscriptionSetupHandler {
setupConsensusTopics(
consumerGroupId,
topicNames,
- ConsensusSubscriptionSetupHandler::isConsensusBasedTopicRequired,
+ topicName -> isConsensusBasedTopicRequired(topicName, isTableModel),
topicName ->
- setupConsensusQueueForTopic(consumerGroupId, topicName,
ioTConsensus, commitManager),
+ setupConsensusQueueForTopic(
+ consumerGroupId, topicName, isTableModel, ioTConsensus,
commitManager),
attemptedTopicNames ->
rollbackConsensusSubscriptionSetup(
consumerGroupId, attemptedTopicNames, commitManager,
setupSnapshot));
@@ -456,13 +470,15 @@ public class ConsensusSubscriptionSetupHandler {
private static void setupConsensusQueueForTopic(
final String consumerGroupId,
final String topicName,
+ final boolean isTableModel,
final IoTConsensus ioTConsensus,
final ConsensusSubscriptionCommitManager commitManager) {
final int myNodeId = IOTDB_CONFIG.getDataNodeId();
// Get topic config for building the converter
final Map<String, TopicConfig> topicConfigs =
-
SubscriptionAgent.topic().getTopicConfigs(java.util.Collections.singleton(topicName));
+ SubscriptionAgent.topic()
+ .getTopicConfigs(java.util.Collections.singleton(topicName),
isTableModel);
final TopicConfig topicConfig = topicConfigs.get(topicName);
if (topicConfig == null) {
throw new SubscriptionException(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
index acbc1fc09d8..67321b04651 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
@@ -259,7 +259,9 @@ public class SubscriptionPipeTabletEventBatch extends
SubscriptionPipeEventBatch
final TopicConfig topicConfig =
SubscriptionAgent.topic()
-
.getTopicConfigs(Collections.singleton(prefetchingQueue.getTopicName()))
+ .getTopicConfigs(
+ Collections.singleton(prefetchingQueue.getTopicName()),
+
SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()))
.get(prefetchingQueue.getTopicName());
if (Objects.isNull(topicConfig)) {
return false;
@@ -331,7 +333,10 @@ public class SubscriptionPipeTabletEventBatch extends
SubscriptionPipeEventBatch
}
final ColumnFilterMatcher matcher =
-
SubscriptionAgent.broker().getColumnFilterMatcher(prefetchingQueue.getTopicName());
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ prefetchingQueue.getTopicName(),
+
SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()));
final List<Tablet> prunedTablets = new ArrayList<>(tablets.right.size());
for (final Tablet tablet : tablets.right) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
index a8735b59edd..7a27f464ebb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java
@@ -158,7 +158,10 @@ public class SubscriptionPipeTsFileEventBatch extends
SubscriptionPipeEventBatch
private Tablet pruneTableModelTablet(final String databaseName, final Tablet
tablet) {
final ColumnFilterMatcher matcher =
-
SubscriptionAgent.broker().getColumnFilterMatcher(prefetchingQueue.getTopicName());
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ prefetchingQueue.getTopicName(),
+
SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId()));
return TabletColumnPruner.pruneTableModelTablet(tablet, databaseName,
matcher);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
index 91c136a4f42..1392c0cbb28 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java
@@ -96,7 +96,10 @@ public class SubscriptionEventTabletResponse extends
SubscriptionEventExtendable
this.commitContext = commitContext;
this.rootCommitContext = rootCommitContext;
this.columnFilterMatcher =
-
SubscriptionAgent.broker().getColumnFilterMatcher(queue.getTopicName());
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ queue.getTopicName(),
+
SubscriptionAgent.consumer().isTableModel(queue.getConsumerGroupId()));
init();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java
index dd047384ff5..9ddeca25c4c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java
@@ -79,7 +79,10 @@ public class SubscriptionEventTsFileResponse extends
SubscriptionEventExtendable
this.databaseName = databaseName;
this.commitContext = commitContext;
this.columnFilterMatcher =
-
SubscriptionAgent.broker().getColumnFilterMatcher(commitContext.getTopicName());
+ SubscriptionAgent.broker()
+ .getColumnFilterMatcher(
+ commitContext.getTopicName(),
+
SubscriptionAgent.consumer().isTableModel(commitContext.getConsumerGroupId()));
this.timeSelectedByTable =
columnFilterMatcher.getTimeSelectedByTable(databaseName);
init();
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 c7c791ff128..268bb8df119 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
@@ -373,7 +373,8 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
// fetch subscribed topics
final Map<String, TopicConfig> topics =
- SubscriptionAgent.topic().getTopicConfigs(subscribedTopicNames);
+ SubscriptionAgent.topic()
+ .getTopicConfigs(subscribedTopicNames,
isTableModel(consumerConfig));
// fetch available endpoints
final Map<Integer, TEndPoint> endPoints = new HashMap<>();
@@ -462,7 +463,8 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
.getTopicConfigs(
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
- consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId())));
+ consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId()),
+ isTableModel(consumerConfig)));
}
private TPipeSubscribeResp handlePipeSubscribeUnsubscribe(final
PipeSubscribeUnsubscribeReq req) {
@@ -508,7 +510,8 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
.getTopicConfigs(
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
- consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId())));
+ consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId()),
+ isTableModel(consumerConfig)));
}
private TPipeSubscribeResp handlePipeSubscribePoll(final
PipeSubscribePollReq req) {
@@ -658,7 +661,7 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
SubscriptionPrefetchingQueue.generatePrefetchingQueueId(
commitContext.getConsumerGroupId(),
commitContext.getTopicName());
if
(ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
- commitContext.getTopicName())) {
+ commitContext.getTopicName(),
isTableModel(consumerConfig))) {
ConsensusSubscriptionPrefetchingQueueMetrics.getInstance()
.mark(queueId, commitContext.getRegionId(), size);
} else {
@@ -1069,7 +1072,8 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
.getTopicConfigs(
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
- consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId()));
+ consumerConfig.getConsumerGroupId(),
consumerConfig.getConsumerId()),
+ isTableModel(consumerConfig));
// fetch topics should be unsubscribed
final List<String> topicNamesToUnsubscribe =
@@ -1274,4 +1278,8 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs(),
consumerConfig.getHeartbeatIntervalMs() *
HEARTBEAT_TIMEOUT_MULTIPLIER);
}
+
+ private static boolean isTableModel(final ConsumerConfig consumerConfig) {
+ return
SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(consumerConfig.getSqlDialect());
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtask.java
index 454ce2b80b6..6c26e0fab10 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtask.java
@@ -118,6 +118,7 @@ public class SubscriptionSinkSubtask extends
PipeSinkSubtask {
}
private boolean isConsensusDrivenTopic() {
- return ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName);
+ return ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName, SubscriptionAgent.consumer().isTableModel(consumerGroupId));
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
index 3a0f52e3f99..2fb50ec4a16 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
@@ -53,7 +53,9 @@ public class SubscriptionSinkSubtaskLifeCycle extends
PipeSinkSubtaskLifeCycle {
if (registeredTaskCount == 0) {
if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
- ((SubscriptionSinkSubtask) subtask).getTopicName())) {
+ ((SubscriptionSinkSubtask) subtask).getTopicName(),
+ SubscriptionAgent.consumer()
+ .isTableModel(((SubscriptionSinkSubtask)
subtask).getConsumerGroupId()))) {
SubscriptionAgent.broker().bindPrefetchingQueue((SubscriptionSinkSubtask)
subtask);
}
executor.register(subtask);
@@ -103,7 +105,8 @@ public class SubscriptionSinkSubtaskLifeCycle extends
PipeSinkSubtaskLifeCycle {
// when dropping the subscription.
final String consumerGroupId = ((SubscriptionSinkSubtask)
subtask).getConsumerGroupId();
final String topicName = ((SubscriptionSinkSubtask)
subtask).getTopicName();
- if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(topicName)) {
+ if (!ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName,
SubscriptionAgent.consumer().isTableModel(consumerGroupId))) {
SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId,
topicName);
}
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
index 010cf08f196..e8646e005a3 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -925,6 +925,7 @@ public final class PipeMessages {
public static final String
EXCEPTION_FAILED_REMOVE_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_75C319C3
=
"Failed to remove subscription from consumer group meta: consumer %s
does not exist in"
+ " consumer group %s";
+ public static final String
EXCEPTION_FAILED_TO_CREATE_CONSUMER_ARG_BECAUSE_INCONSISTENT_SQL_DIALECT_UNDER_THE_SAME_CONSUMER_GROUP_EXPECTED_ARG_ACTUAL_ARG_A7DA3FB9
= "Failed to create consumer %s because inconsistent sql dialect under the
same consumer group, expected %s, actual %s";
public static final String
EXCEPTION_PATH_PATTERN_ARG_NOT_VALID_SOURCE_ONLY_PREFIX_FULL_PATH_784778B8 =
"The path pattern %s is not valid for the source. Only prefix or full path is
allowed.";
public static final String
EXCEPTION_CAPTURE_TREE_CAN_NOT_SPECIFIED_FALSE_DOUBLE_LIVING_ENABLED_29A08445 =
"capture.tree can not be specified to false when double living is enabled";
public static final String
EXCEPTION_CAPTURE_TABLE_CAN_NOT_SPECIFIED_FALSE_DOUBLE_LIVING_ENABLED_8AEB8F7B
= "capture.table can not be specified to false when double living is enabled";
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
index 8ea19a43f89..c6da4c6082e 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -894,6 +894,7 @@ public final class PipeMessages {
public static final String
EXCEPTION_INVALID_REGION_PROGRESS_VALUE_LENGTH_6192D17F = "region progress
value 长度无效：%d";
public static final String
EXCEPTION_FAILED_ADD_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_EF08EE87
= "添加 subscription 到 consumer group meta 失败：consumer %s 不存在于 consumer group
%s";
public static final String
EXCEPTION_FAILED_REMOVE_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_75C319C3
= "从 consumer group meta 移除 subscription 失败：consumer %s 不存在于 consumer group
%s";
+ public static final String
EXCEPTION_FAILED_TO_CREATE_CONSUMER_ARG_BECAUSE_INCONSISTENT_SQL_DIALECT_UNDER_THE_SAME_CONSUMER_GROUP_EXPECTED_ARG_ACTUAL_ARG_A7DA3FB9
= "创建 consumer %s 失败，因为同一 consumer group 下的 SQL dialect 不一致，预期为 %s，实际为 %s";
public static final String
EXCEPTION_PATH_PATTERN_ARG_NOT_VALID_SOURCE_ONLY_PREFIX_FULL_PATH_784778B8 =
"路径模式 %s 对 source 无效。仅允许前缀或完整路径。";
public static final String
EXCEPTION_CAPTURE_TREE_CAN_NOT_SPECIFIED_FALSE_DOUBLE_LIVING_ENABLED_29A08445 =
"启用 double living 时，capture.tree 不能指定为 false";
public static final String
EXCEPTION_CAPTURE_TABLE_CAN_NOT_SPECIFIED_FALSE_DOUBLE_LIVING_ENABLED_8AEB8F7B
= "启用 double living 时，capture.table 不能指定为 false";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
index af07dffa2cc..c6153db8668 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.commons.subscription.meta.consumer;
import org.apache.iotdb.commons.i18n.PipeMessages;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.apache.thrift.annotation.Nullable;
@@ -163,6 +164,24 @@ public class ConsumerGroupMeta {
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
+
+ final String expectedSqlDialect =
existedConsumerMeta.getConfig().getSqlDialect();
+ final String actualSqlDialect = consumerMeta.getConfig().getSqlDialect();
+ final boolean isExpectedTableModel =
+
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(expectedSqlDialect);
+ final boolean isActualTableModel =
+
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(actualSqlDialect);
+ if (isExpectedTableModel != isActualTableModel) {
+ final String exceptionMessage =
+ String.format(
+ PipeMessages
+
.EXCEPTION_FAILED_TO_CREATE_CONSUMER_ARG_BECAUSE_INCONSISTENT_SQL_DIALECT_UNDER_THE_SAME_CONSUMER_GROUP_EXPECTED_ARG_ACTUAL_ARG_A7DA3FB9,
+ consumerMeta.getConsumerId(),
+ expectedSqlDialect,
+ actualSqlDialect);
+ LOGGER.warn(exceptionMessage);
+ throw new SubscriptionException(exceptionMessage);
+ }
}
public void addConsumer(final ConsumerMeta consumerMeta) {
@@ -196,6 +215,15 @@ public class ConsumerGroupMeta {
return consumerIdToConsumerMeta.get(consumerId);
}
+ public boolean visibleUnder(final boolean isTableModel) {
+ if (consumerIdToConsumerMeta.isEmpty()) {
+ return false;
+ }
+ final String sqlDialect =
+
consumerIdToConsumerMeta.values().iterator().next().getConfig().getSqlDialect();
+ return isTableModel ==
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(sqlDialect);
+ }
+
////////////////////////// subscription //////////////////////////
/** Get all topic names subscribed by this consumer group. */
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMetaKeeper.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMetaKeeper.java
index 34dbbbd761d..330ae2cf9ee 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMetaKeeper.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMetaKeeper.java
@@ -131,6 +131,17 @@ public class ConsumerGroupMetaKeeper {
.collect(Collectors.toSet());
}
+ public Set<String> getSubscribedConsumerGroupIds(
+ final String topicName, final boolean isTableModel) {
+ return consumerGroupIdToConsumerGroupMetaMap.entrySet().stream()
+ .filter(
+ entry ->
+ entry.getValue().visibleUnder(isTableModel)
+ &&
entry.getValue().isTopicSubscribedByConsumerGroup(topicName))
+ .map(Entry::getKey)
+ .collect(Collectors.toSet());
+ }
+
public boolean isTopicSubscribedByConsumerGroup(
final String topicName, final String consumerGroupId) {
return consumerGroupIdToConsumerGroupMetaMap.containsKey(consumerGroupId)
@@ -144,6 +155,15 @@ public class ConsumerGroupMetaKeeper {
.anyMatch(meta -> meta.isTopicSubscribedByConsumerGroup(topicName));
}
+ public boolean isTopicSubscribedByConsumerGroup(
+ final String topicName, final boolean isTableModel) {
+ return consumerGroupIdToConsumerGroupMetaMap.values().stream()
+ .anyMatch(
+ meta ->
+ meta.visibleUnder(isTableModel)
+ && meta.isTopicSubscribedByConsumerGroup(topicName));
+ }
+
///////////////////////////////// Snapshot
/////////////////////////////////
public void processTakeSnapshot(FileOutputStream fileOutputStream) throws
IOException {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerMeta.java
index 75cca9ccbcd..006bd9b39cb 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerMeta.java
@@ -74,6 +74,10 @@ public class ConsumerMeta {
return Objects.nonNull(getEncryptedPassword()) ? getEncryptedPassword() :
getPassword();
}
+ public ConsumerConfig getConfig() {
+ return config;
+ }
+
public ByteBuffer serialize() throws IOException {
PublicBAOS byteArrayOutputStream = new PublicBAOS();
DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMetaKeeper.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMetaKeeper.java
index 741f7c38058..07cd7465681 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMetaKeeper.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMetaKeeper.java
@@ -61,11 +61,16 @@ public class TopicMetaKeeper {
///////////////////////////////// TopicMeta
/////////////////////////////////
public void addTopicMeta(String topicName, TopicMeta topicMeta) {
- topicNameToTopicMetaMap.put(topicName, topicMeta);
+ topicNameToTopicMetaMap.put(generateTopicMetaKeeperKey(topicMeta),
topicMeta);
}
public TopicMeta getTopicMeta(String topicName) {
- return topicNameToTopicMetaMap.get(topicName);
+ final TopicMeta treeTopicMeta = getTopicMeta(topicName, false);
+ return Objects.nonNull(treeTopicMeta) ? treeTopicMeta :
getTopicMeta(topicName, true);
+ }
+
+ public TopicMeta getTopicMeta(String topicName, boolean isTableModel) {
+ return topicNameToTopicMetaMap.get(generateTopicMetaKeeperKey(topicName,
isTableModel));
}
public Iterable<TopicMeta> getAllTopicMeta() {
@@ -73,19 +78,22 @@ public class TopicMetaKeeper {
}
public void removeTopicMeta(String topicName) {
- topicNameToTopicMetaMap.remove(topicName);
+ final String treeModelKey = generateTopicMetaKeeperKey(topicName, false);
+ if (topicNameToTopicMetaMap.remove(treeModelKey) == null) {
+ topicNameToTopicMetaMap.remove(generateTopicMetaKeeperKey(topicName,
true));
+ }
+ }
+
+ public void removeTopicMeta(String topicName, boolean isTableModel) {
+ topicNameToTopicMetaMap.remove(generateTopicMetaKeeperKey(topicName,
isTableModel));
}
public boolean containsTopicMeta(String topicName) {
- return topicNameToTopicMetaMap.containsKey(topicName);
+ return containsTopicMeta(topicName, false) || containsTopicMeta(topicName,
true);
}
public boolean containsTopicMeta(String topicName, boolean isTableModel) {
- TopicMeta topicMeta = topicNameToTopicMetaMap.get(topicName);
- if (Objects.isNull(topicMeta)) {
- return false;
- }
- return topicMeta.visibleUnder(isTableModel);
+ return
topicNameToTopicMetaMap.containsKey(generateTopicMetaKeeperKey(topicName,
isTableModel));
}
public void clear() {
@@ -101,7 +109,7 @@ public class TopicMetaKeeper {
public void processTakeSnapshot(FileOutputStream fileOutputStream) throws
IOException {
ReadWriteIOUtils.write(topicNameToTopicMetaMap.size(), fileOutputStream);
for (Map.Entry<String, TopicMeta> entry :
topicNameToTopicMetaMap.entrySet()) {
- ReadWriteIOUtils.write(entry.getKey(), fileOutputStream);
+ ReadWriteIOUtils.write(entry.getValue().getTopicName(),
fileOutputStream);
entry.getValue().serialize(fileOutputStream);
}
}
@@ -112,7 +120,8 @@ public class TopicMetaKeeper {
final int size = ReadWriteIOUtils.readInt(fileInputStream);
for (int i = 0; i < size; i++) {
final String topicName = ReadWriteIOUtils.readString(fileInputStream);
- topicNameToTopicMetaMap.put(topicName,
TopicMeta.deserialize(fileInputStream));
+ final TopicMeta topicMeta = TopicMeta.deserialize(fileInputStream);
+ topicNameToTopicMetaMap.put(generateTopicMetaKeeperKey(topicMeta),
topicMeta);
}
}
@@ -139,4 +148,15 @@ public class TopicMetaKeeper {
public String toString() {
return "TopicMetaKeeper{" + "topicNameToTopicMetaMap=" +
topicNameToTopicMetaMap + '}';
}
+
+ ///////////////////////////////// Tree & Table Isolation
/////////////////////////////////
+
+ private static String generateTopicMetaKeeperKey(final TopicMeta topicMeta) {
+ return generateTopicMetaKeeperKey(topicMeta.getTopicName(),
topicMeta.visibleUnderTableModel());
+ }
+
+ private static String generateTopicMetaKeeperKey(
+ final String topicName, final boolean isTableModel) {
+ return (isTableModel ? "table:" : "tree:") + topicName;
+ }
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
index 9ef7191ba66..fc8d0122b2b 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
@@ -19,8 +19,10 @@
package org.apache.iotdb.commons.subscription.consumer;
+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.rpc.subscription.config.ConsumerConstant;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.junit.Assert;
@@ -89,4 +91,27 @@ public class ConsumerGroupDeSerTest {
consumerGroupMeta.getConsumersSubscribingTopic("test_topic").contains("test_consumer1"));
Assert.assertTrue(copiedConsumerGroupMeta.getConsumersSubscribingTopic("test_topic").isEmpty());
}
+
+ @Test
+ public void testConsumerGroupShouldRejectMixedSqlDialects() {
+ final Map<String, String> treeConsumerAttributes = new HashMap<>();
+ treeConsumerAttributes.put("username", "user");
+ treeConsumerAttributes.put("password", "password");
+ treeConsumerAttributes.put(
+ ConsumerConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE);
+
+ final Map<String, String> tableConsumerAttributes = new
HashMap<>(treeConsumerAttributes);
+ tableConsumerAttributes.put(
+ ConsumerConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE);
+
+ final ConsumerGroupMeta consumerGroupMeta =
+ new ConsumerGroupMeta(
+ "test_consumer_group", 1, new ConsumerMeta("tree_consumer", 1,
treeConsumerAttributes));
+
+ Assert.assertThrows(
+ SubscriptionException.class,
+ () ->
+ consumerGroupMeta.checkAuthorityBeforeJoinConsumerGroup(
+ new ConsumerMeta("table_consumer", 2,
tableConsumerAttributes)));
+ }
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaKeeperTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaKeeperTest.java
new file mode 100644
index 00000000000..00592403109
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicMetaKeeperTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.commons.subscription.topic;
+
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+
+public class TopicMetaKeeperTest {
+
+ @Test
+ public void testSameNameTreeAndTableTopicsCanCoexist() {
+ final String topicName = "topic";
+ final TopicMeta treeTopicMeta =
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE));
+ final TopicMeta tableTopicMeta =
+ new TopicMeta(
+ topicName,
+ 2L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE));
+ final TopicMetaKeeper keeper = new TopicMetaKeeper();
+
+ keeper.addTopicMeta(topicName, treeTopicMeta);
+ keeper.addTopicMeta(topicName, tableTopicMeta);
+
+ Assert.assertSame(treeTopicMeta, keeper.getTopicMeta(topicName, false));
+ Assert.assertSame(tableTopicMeta, keeper.getTopicMeta(topicName, true));
+ Assert.assertSame(treeTopicMeta, keeper.getTopicMeta(topicName));
+ Assert.assertEquals(2, sizeOf(keeper.getAllTopicMeta()));
+
+ keeper.removeTopicMeta(topicName, false);
+ Assert.assertFalse(keeper.containsTopicMeta(topicName, false));
+ Assert.assertTrue(keeper.containsTopicMeta(topicName, true));
+ Assert.assertSame(tableTopicMeta, keeper.getTopicMeta(topicName));
+ }
+
+ @Test
+ public void testSnapshotPreservesSameNameTreeAndTableTopics() throws
Exception {
+ final String topicName = "topic";
+ final TopicMetaKeeper keeper = new TopicMetaKeeper();
+ keeper.addTopicMeta(
+ topicName,
+ new TopicMeta(
+ topicName,
+ 1L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE)));
+ keeper.addTopicMeta(
+ topicName,
+ new TopicMeta(
+ topicName,
+ 2L,
+ Collections.singletonMap(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE)));
+
+ final Path snapshotFile = Files.createTempFile("topic-meta-keeper",
".snapshot");
+ try {
+ try (final FileOutputStream outputStream = new
FileOutputStream(snapshotFile.toFile())) {
+ keeper.processTakeSnapshot(outputStream);
+ }
+
+ try (final FileInputStream inputStream = new
FileInputStream(snapshotFile.toFile())) {
+ Assert.assertEquals(2, ReadWriteIOUtils.readInt(inputStream));
+ for (int i = 0; i < 2; i++) {
+ Assert.assertEquals(topicName,
ReadWriteIOUtils.readString(inputStream));
+ TopicMeta.deserialize(inputStream);
+ }
+ }
+
+ final TopicMetaKeeper restoredKeeper = new TopicMetaKeeper();
+ try (final FileInputStream inputStream = new
FileInputStream(snapshotFile.toFile())) {
+ restoredKeeper.processLoadSnapshot(inputStream);
+ }
+ Assert.assertNotNull(restoredKeeper.getTopicMeta(topicName, false));
+ Assert.assertNotNull(restoredKeeper.getTopicMeta(topicName, true));
+ } finally {
+ Files.deleteIfExists(snapshotFile);
+ }
+ }
+
+ private static int sizeOf(final Iterable<TopicMeta> topicMetas) {
+ int size = 0;
+ for (final TopicMeta ignored : topicMetas) {
+ size++;
+ }
+ return size;
+ }
+}
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index cc5e0b7dc25..ea0f9467a2a 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -567,6 +567,7 @@ struct TPushTopicMetaReq {
struct TPushSingleTopicMetaReq {
1: optional binary topicMeta // Should not set both to null.
2: optional string topicNameToDrop
+ 3: optional bool isTableModel
}
struct TPushMultiTopicMetaReq {
@@ -594,6 +595,7 @@ struct TTopicOwnerLeaseEntry {
2: required string ownerId
3: required i64 ownerEpoch
4: required i64 leaseRemainingMs
+ 5: optional bool isTableModel
}
struct TPushTopicOwnerLeaseReq {
