Skip to content

Subscription: improve parsing logic when using JAVA SDK client & refactor subscription IT & intro getSubscribedTopicNames API #12721

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SteveYurongSu merged 3 commits into from
Jun 12, 2024
Merged

Subscription: improve parsing logic when using JAVA SDK client & refactor subscription IT & intro getSubscribedTopicNames API #12721

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3b3d6f0
basic apply patch
VGalaxies Jun 12, 2024
2865746
fixup! basic apply patch
VGalaxies Jun 12, 2024
f7b049e
fixup
VGalaxies Jun 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 34 additions & 14 deletions example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,13 @@
import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;

import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.expression.QueryExpression;
import org.apache.tsfile.read.query.dataset.QueryDataSet;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
Expand All @@ -46,11 +50,12 @@ public class SubscriptionSessionExample {
private static final int PORT = 6667;

private static final String TOPIC_1 = "topic1";
private static final String TOPIC_2 = "`topic2`";
private static final String TOPIC_2 = "`'topic2'`";

public static final long SLEEP_NS = 1_000_000_000L;
public static final long POLL_TIMEOUT_MS = 10_000L;
private static final long SLEEP_NS = 1_000_000_000L;
private static final long POLL_TIMEOUT_MS = 10_000L;
private static final int MAX_RETRY_TIMES = 3;
private static final int PARALLELISM = 8;

private static void prepareData() throws Exception {
// Open session
Expand All @@ -66,13 +71,13 @@ private static void prepareData() throws Exception {

// Insert some historical data
final long currentTime = System.currentTimeMillis();
for (int i = 0; i < 10000; ++i) {
for (int i = 0; i < 100; ++i) {
session.executeNonQueryStatement(
String.format("insert into root.db.d1(time, s1, s2) values (%s, 1, 2)", i));
session.executeNonQueryStatement(
String.format("insert into root.db.d2(time, s3, s4) values (%s, 3, 4)", currentTime + i));
String.format("insert into root.db.d2(time, s1, s2) values (%s, 3, 4)", currentTime + i));
session.executeNonQueryStatement(
String.format("insert into root.sg.d3(time, s5) values (%s, 5)", currentTime + 2 * i));
String.format("insert into root.sg.d3(time, s1) values (%s, 5)", currentTime + 2 * i));
}
session.executeNonQueryStatement("flush");

Expand Down Expand Up @@ -104,11 +109,16 @@ private static void dataQuery() throws Exception {
session = null;
}

private static void subscriptionExample1() throws Exception {
/** single pull consumer subscribe topic with path and time range */
private static void dataSubscription1() throws Exception {
// Create topics
try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
subscriptionSession.open();
subscriptionSession.createTopic(TOPIC_1);
final Properties config = new Properties();
config.put(TopicConstant.PATH_KEY, "root.db.d1.s1");
config.put(TopicConstant.START_TIME_KEY, 25);
config.put(TopicConstant.END_TIME_KEY, 75);
subscriptionSession.createTopic(TOPIC_1, config);
}

int retryCount = 0;
Expand Down Expand Up @@ -151,7 +161,8 @@ private static void subscriptionExample1() throws Exception {
consumer1.close();
}

private static void subscriptionExample2() throws Exception {
/** multi pull consumer subscribe topic with tsfile format */
private static void dataSubscription2() throws Exception {
try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
subscriptionSession.open();
final Properties config = new Properties();
Expand All @@ -160,7 +171,7 @@ private static void subscriptionExample2() throws Exception {
}

final List<Thread> threads = new ArrayList<>();
for (int i = 0; i < 8; ++i) {
for (int i = 0; i < PARALLELISM; ++i) {
final int idx = i;
final Thread thread =
new Thread(
Expand All @@ -187,7 +198,16 @@ private static void subscriptionExample2() throws Exception {
}
for (final SubscriptionMessage message : messages) {
try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
// do something...
final QueryDataSet dataSet =
reader.query(
QueryExpression.create(
Arrays.asList(
new Path("root.db.d2", "s2", true),
new Path("root.db.d3", "s1", true)),
null));
while (dataSet.hasNext()) {
System.out.println(dataSet.next());
}
}
}
consumer2.commitSync(messages);
Expand All @@ -208,8 +228,8 @@ private static void subscriptionExample2() throws Exception {

public static void main(final String[] args) throws Exception {
prepareData();
dataQuery();
subscriptionExample1();
subscriptionExample2();
// dataQuery();
// dataSubscription1();
dataSubscription2();
}
}
49 changes: 49 additions & 0 deletions integration-test/src/test/java/org/apache/iotdb/subscription/it/AbstractSubscriptionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* 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.subscription.it;

import org.apache.iotdb.session.subscription.consumer.SubscriptionExecutorServiceManager;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.rules.TestName;
import org.junit.rules.TestRule;

public abstract class AbstractSubscriptionIT {

@Rule public TestName testName = new TestName();

@Rule public final TestRule skipOnSetUpFailure = new SkipOnSetUpFailure("setUp");

@Before
public void setUp() {
// set thread name
Thread.currentThread().setName(String.format("%s - main", testName.getMethodName()));

// set thread pools core size
SubscriptionExecutorServiceManager.setControlFlowExecutorCorePoolSize(1);
SubscriptionExecutorServiceManager.setUpstreamDataFlowExecutorCorePoolSize(1);
SubscriptionExecutorServiceManager.setDownstreamDataFlowExecutorCorePoolSize(1);
}

@After
public void tearDown() {}
}
19 changes: 16 additions & 3 deletions ...tion-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,24 @@

package org.apache.iotdb.subscription.it;

import org.awaitility.Awaitility;
import org.awaitility.core.ConditionFactory;

import java.util.concurrent.TimeUnit;

public class IoTDBSubscriptionITConstant {

public static final long AWAITILITY_POLL_DELAY_SECOND = 1L;
public static final long AWAITILITY_POLL_INTERVAL_SECOND = 2L;
public static final long AWAITILITY_AT_MOST_SECOND = 600L;
private static final long AWAITILITY_POLL_DELAY_SECOND = 1L;
private static final long AWAITILITY_POLL_INTERVAL_SECOND = 2L;
private static final long AWAITILITY_AT_MOST_SECOND = 600L;

public static final ConditionFactory AWAIT =
Awaitility.await()
.pollInSameThread()
.pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS)
.pollInterval(
IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS)
.atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS);

public static final long SLEEP_NS = 1_000_000_000L;
public static final long POLL_TIMEOUT_MS = 10_000L;
Expand Down
53 changes: 21 additions & 32 deletions .../it/local/IoTDBSubscriptionRestartIT.java → ...t/cluster/IoTDBSubscriptionRestartIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
* under the License.
*/

package org.apache.iotdb.subscription.it.local;
package org.apache.iotdb.subscription.it.cluster;

import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq;
Expand All @@ -36,40 +36,38 @@
import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
import org.apache.iotdb.subscription.it.SkipOnSetUpFailure;

import org.awaitility.Awaitility;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestRule;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT;
import static org.junit.Assert.fail;

@RunWith(IoTDBTestRunner.class)
@Category({ClusterIT.class})
public class IoTDBSubscriptionRestartIT {
public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {

private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionRestartIT.class);

@Rule public final TestRule skipOnSetUpFailure = new SkipOnSetUpFailure("setUp");

@Override
@Before
public void setUp() throws Exception {
public void setUp() {
super.setUp();

EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
Expand All @@ -82,8 +80,11 @@ public void setUp() throws Exception {
EnvFactory.getEnv().initClusterEnvironment(3, 3);
}

@Override
@After
public void tearDown() throws Exception {
public void tearDown() {
super.tearDown();

EnvFactory.getEnv().cleanClusterEnvironment();
}

Expand Down Expand Up @@ -195,18 +196,14 @@ public void testSubscriptionAfterRestartCluster() throws Exception {
} finally {
LOGGER.info("consumer exiting...");
}
});
},
String.format("%s - %s", testName.getMethodName(), consumer));
thread.start();

// Check timestamps size
try {
// Keep retrying if there are execution failures
Awaitility.await()
.pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS)
.pollInterval(
IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS)
.atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(100, timestamps.size()));
AWAIT.untilAsserted(() -> Assert.assertEquals(100, timestamps.size()));
} catch (final Exception e) {
e.printStackTrace();
fail(e.getMessage());
Expand Down Expand Up @@ -312,7 +309,8 @@ public void testSubscriptionAfterRestartDataNode() throws Exception {
} finally {
LOGGER.info("consumer exiting...");
}
});
},
String.format("%s - %s", testName.getMethodName(), consumer));
thread.start();

// Start DN 1 & DN 2
Expand Down Expand Up @@ -343,12 +341,7 @@ public void testSubscriptionAfterRestartDataNode() throws Exception {
// Check timestamps size
try {
// Keep retrying if there are execution failures
Awaitility.await()
.pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS)
.pollInterval(
IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS)
.atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(200, timestamps.size()));
AWAIT.untilAsserted(() -> Assert.assertEquals(200, timestamps.size()));
} catch (final Exception e) {
e.printStackTrace();
fail(e.getMessage());
Expand Down Expand Up @@ -443,7 +436,8 @@ public void testSubscriptionWhenConfigNodeLeaderChange() throws Exception {
} finally {
LOGGER.info("consumer exiting...");
}
});
},
String.format("%s - %s", testName.getMethodName(), consumer));
thread.start();

// Shutdown leader CN
Expand Down Expand Up @@ -486,12 +480,7 @@ public void testSubscriptionWhenConfigNodeLeaderChange() throws Exception {
// Check timestamps size
try {
// Keep retrying if there are execution failures
Awaitility.await()
.pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS)
.pollInterval(
IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS)
.atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(200, timestamps.size()));
AWAIT.untilAsserted(() -> Assert.assertEquals(200, timestamps.size()));
} catch (final Exception e) {
e.printStackTrace();
fail(e.getMessage());
Expand Down
24 changes: 9 additions & 15 deletions ...-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,29 +21,20 @@

import org.apache.iotdb.it.env.MultiEnvFactory;
import org.apache.iotdb.itbase.env.BaseEnv;
import org.apache.iotdb.session.subscription.consumer.SubscriptionExecutorServiceManager;
import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.rules.TestName;

abstract class AbstractSubscriptionDualIT {
abstract class AbstractSubscriptionDualIT extends AbstractSubscriptionIT {

protected BaseEnv senderEnv;
protected BaseEnv receiverEnv;

@Rule public TestName testName = new TestName();

@Override
@Before
public void setUp() {
// set thread name
Thread.currentThread().setName(String.format("%s - main", testName.getMethodName()));

// set thread pools core size
SubscriptionExecutorServiceManager.setControlFlowExecutorCorePoolSize(1);
SubscriptionExecutorServiceManager.setUpstreamDataFlowExecutorCorePoolSize(1);
SubscriptionExecutorServiceManager.setDownstreamDataFlowExecutorCorePoolSize(1);
super.setUp();

MultiEnvFactory.createEnv(2);
senderEnv = MultiEnvFactory.getEnv(0);
Expand All @@ -55,7 +46,7 @@ public void setUp() {
receiverEnv.initClusterEnvironment();
}

void setUpConfig() {
protected void setUpConfig() {
// enable auto create schema
senderEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true);
Expand All @@ -65,8 +56,11 @@ void setUpConfig() {
receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
}

@Override
@After
public final void tearDown() {
public void tearDown() {
super.tearDown();

senderEnv.cleanClusterEnvironment();
receiverEnv.cleanClusterEnvironment();
}
Expand Down
Loading