You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/06/25 11:41:20 UTC
[GitHub] [iotdb] qiaojialin commented on a diff in pull request #6378: [IOTDB-3525] Apply Producer-Consumer pattern to compaction submission
qiaojialin commented on code in PR #6378:
URL: https://github.com/apache/iotdb/pull/6378#discussion_r906662965
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionTaskSummary.java:
##########
@@ -20,13 +20,49 @@
/** The summary of one {@link AbstractCompactionTask} execution */
public class CompactionTaskSummary {
- private final boolean success;
+ private long timeCost = 0L;
+ private volatile boolean ran = false;
+ private volatile boolean finished = false;
+ private volatile boolean cancel = false;
+ private volatile boolean success = false;
Review Comment:
Use a Status like a StateMachine
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -72,16 +69,9 @@ public class CompactionTaskManager implements IService {
new FixedPriorityBlockingQueue<>(1024, new DefaultCompactionTaskComparatorImpl());
// <fullStorageGroupName,futureSet>, it is used to store all compaction tasks under each
Review Comment:
No more fullStorageGroupName, we only have StorageGroup and DataRegionId
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -72,16 +69,9 @@ public class CompactionTaskManager implements IService {
new FixedPriorityBlockingQueue<>(1024, new DefaultCompactionTaskComparatorImpl());
// <fullStorageGroupName,futureSet>, it is used to store all compaction tasks under each
// virtualStorageGroup
- private Map<String, Map<AbstractCompactionTask, Future<CompactionTaskSummary>>>
- storageGroupTasks = new HashMap<>();
-
- // The thread pool that periodically fetches and executes the compaction task from
- // candidateCompactionTaskQueue to taskExecutionPool. The default number of threads for this pool
- // is 1.
- private ScheduledExecutorService compactionTaskSubmissionThreadPool;
-
- private final long TASK_SUBMIT_INTERVAL =
- IoTDBDescriptor.getInstance().getConfig().getCompactionSubmissionIntervalInMs();
+ private final Map<String, Map<AbstractCompactionTask, Future<CompactionTaskSummary>>>
+ storageGroupTasks = new ConcurrentHashMap<>();
+ private AtomicInteger finishTaskNum = new AtomicInteger(0);
private final RateLimiter mergeWriteRateLimiter = RateLimiter.create(Double.MAX_VALUE);
Review Comment:
The RateLimiter is marked as unstable with @beta. Try to find a stable one later
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -72,16 +69,9 @@ public class CompactionTaskManager implements IService {
new FixedPriorityBlockingQueue<>(1024, new DefaultCompactionTaskComparatorImpl());
// <fullStorageGroupName,futureSet>, it is used to store all compaction tasks under each
// virtualStorageGroup
- private Map<String, Map<AbstractCompactionTask, Future<CompactionTaskSummary>>>
- storageGroupTasks = new HashMap<>();
-
- // The thread pool that periodically fetches and executes the compaction task from
- // candidateCompactionTaskQueue to taskExecutionPool. The default number of threads for this pool
- // is 1.
- private ScheduledExecutorService compactionTaskSubmissionThreadPool;
-
- private final long TASK_SUBMIT_INTERVAL =
- IoTDBDescriptor.getInstance().getConfig().getCompactionSubmissionIntervalInMs();
+ private final Map<String, Map<AbstractCompactionTask, Future<CompactionTaskSummary>>>
+ storageGroupTasks = new ConcurrentHashMap<>();
+ private AtomicInteger finishTaskNum = new AtomicInteger(0);
Review Comment:
```suggestion
private final AtomicInteger finishedTaskNum = new AtomicInteger(0);
```
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionWorker.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.db.engine.compaction;
+
+import org.apache.iotdb.db.engine.compaction.constant.CompactionTaskStatus;
+import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
+import org.apache.iotdb.db.service.metrics.recorder.CompactionMetricsRecorder;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class CompactionWorker implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger("COMPACTION");
+ private final int threadId;
+ private final FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue;
+
+ public CompactionWorker(
+ int threadId, FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue) {
+ this.threadId = threadId;
+ this.compactionTaskQueue = compactionTaskQueue;
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ AbstractCompactionTask task = compactionTaskQueue.take();
+ // add metrics
+ CompactionMetricsRecorder.recordTaskInfo(
+ task, CompactionTaskStatus.POLL_FROM_QUEUE, compactionTaskQueue.size());
+ if (task != null && task.checkValidAndSetMerging()) {
+ CompactionTaskSummary summary = task.getSummary();
+ CompactionTaskFuture future = new CompactionTaskFuture(summary);
+ CompactionTaskManager.getInstance().recordTask(task, future);
+ task.call();
Review Comment:
As AbstractCompactionTask is not submitted to some ThreadPool, it is no longer to implement the Callable interface.
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -155,21 +138,23 @@ public void waitAndStop(long milliseconds) {
public void waitAllCompactionFinish() {
Review Comment:
what's the behavior of this function? When others call this, will more compaction tasks be submitted? There are three candidateCompactionTaskQueue.clear(), which is unnormal.
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionWorker.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.db.engine.compaction;
+
+import org.apache.iotdb.db.engine.compaction.constant.CompactionTaskStatus;
+import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
+import org.apache.iotdb.db.service.metrics.recorder.CompactionMetricsRecorder;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class CompactionWorker implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger("COMPACTION");
+ private final int threadId;
+ private final FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue;
+
+ public CompactionWorker(
+ int threadId, FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue) {
+ this.threadId = threadId;
+ this.compactionTaskQueue = compactionTaskQueue;
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ AbstractCompactionTask task = compactionTaskQueue.take();
+ // add metrics
+ CompactionMetricsRecorder.recordTaskInfo(
+ task, CompactionTaskStatus.POLL_FROM_QUEUE, compactionTaskQueue.size());
+ if (task != null && task.checkValidAndSetMerging()) {
+ CompactionTaskSummary summary = task.getSummary();
+ CompactionTaskFuture future = new CompactionTaskFuture(summary);
+ CompactionTaskManager.getInstance().recordTask(task, future);
+ task.call();
+ }
+ } catch (InterruptedException e) {
+ log.error("CompactionThread-{} terminates because interruption", threadId);
+ return;
+ } catch (Throwable t) {
+ log.error("CompactionThread-{} meets exception", threadId, t);
+ }
+ }
+ }
+
+ class CompactionTaskFuture implements Future<CompactionTaskSummary> {
+ CompactionTaskSummary summary;
+
+ public CompactionTaskFuture(CompactionTaskSummary summary) {
+ this.summary = summary;
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ summary.cancel();
+ return true;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return summary.isCancel();
+ }
+
+ @Override
+ public boolean isDone() {
+ return summary.isFinished();
+ }
+
+ @Override
+ public CompactionTaskSummary get() throws InterruptedException, ExecutionException {
+ while (!summary.isFinished()) {
+ TimeUnit.MILLISECONDS.sleep(100);
+ }
+ return summary;
+ }
+
+ @Override
+ public CompactionTaskSummary get(long timeout, @NotNull TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ long perSleepTime = timeout > 100 ? timeout : 100;
Review Comment:
```suggestion
long perSleepTime = timeout < 100 ? timeout : 100;
```
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java:
##########
@@ -85,10 +88,11 @@ public CompactionTaskSummary call() throws Exception {
} finally {
this.currentTaskNum.decrementAndGet();
timeCost = System.currentTimeMillis() - startTime;
+ summary.finish(isSuccess, timeCost);
+ LOGGER.error("{} finish", this);
Review Comment:
info?
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -72,16 +69,9 @@ public class CompactionTaskManager implements IService {
new FixedPriorityBlockingQueue<>(1024, new DefaultCompactionTaskComparatorImpl());
Review Comment:
final
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -155,21 +138,23 @@ public void waitAndStop(long milliseconds) {
public void waitAllCompactionFinish() {
long sleepingStartTime = 0;
if (taskExecutionPool != null) {
- while (taskExecutionPool.getActiveCount() > 0 || taskExecutionPool.getQueue().size() > 0) {
- // wait
- try {
- Thread.sleep(200);
- sleepingStartTime += 200;
- if (sleepingStartTime % 10000 == 0) {
- logger.warn(
- "Has waiting {} seconds for all compaction task finish", sleepingStartTime / 1000);
- }
- if (sleepingStartTime >= MAX_WAITING_TIME) {
- return;
+ candidateCompactionTaskQueue.clear();
+ while (true) {
+ int totalSize = 0;
+ for (Map<AbstractCompactionTask, Future<CompactionTaskSummary>> taskMap :
+ storageGroupTasks.values()) {
+ totalSize += taskMap.size();
+ }
+ if (totalSize > 0) {
+ try {
+ Thread.sleep(10);
Review Comment:
```suggestion
Thread.sleep(100);
```
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java:
##########
@@ -47,8 +47,10 @@ public abstract class AbstractCompactionTask implements Callable<CompactionTaskS
protected long timeCost = 0L;
protected volatile boolean ran = false;
protected volatile boolean finished = false;
+ protected volatile boolean cancel = false;
Review Comment:
duplicated with CompactionTaskSummary
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -440,4 +374,18 @@ public void restart() throws InterruptedException {
public void clearCandidateQueue() {
candidateCompactionTaskQueue.clear();
}
+
+ @TestOnly
+ public Future<CompactionTaskSummary> getCompactionTaskFutureMayBlock(AbstractCompactionTask task)
+ throws InterruptedException {
+ String storageGroup = task.getFullStorageGroupName();
+ long startTime = System.currentTimeMillis();
+ while (!storageGroupTasks.containsKey(storageGroup)
+ || !storageGroupTasks.get(storageGroup).containsKey(task)) {
+ if (System.currentTimeMillis() - startTime > 20_000) {
+ throw new InterruptedException("Timeout when waiting for task future");
Review Comment:
Throw an ExecutionTimeoutException
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -440,4 +374,18 @@ public void restart() throws InterruptedException {
public void clearCandidateQueue() {
candidateCompactionTaskQueue.clear();
}
+
+ @TestOnly
+ public Future<CompactionTaskSummary> getCompactionTaskFutureMayBlock(AbstractCompactionTask task)
+ throws InterruptedException {
+ String storageGroup = task.getFullStorageGroupName();
+ long startTime = System.currentTimeMillis();
+ while (!storageGroupTasks.containsKey(storageGroup)
+ || !storageGroupTasks.get(storageGroup).containsKey(task)) {
+ if (System.currentTimeMillis() - startTime > 20_000) {
Review Comment:
Add a sleep in this while{}
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -155,21 +138,23 @@ public void waitAndStop(long milliseconds) {
public void waitAllCompactionFinish() {
long sleepingStartTime = 0;
if (taskExecutionPool != null) {
- while (taskExecutionPool.getActiveCount() > 0 || taskExecutionPool.getQueue().size() > 0) {
- // wait
- try {
- Thread.sleep(200);
- sleepingStartTime += 200;
- if (sleepingStartTime % 10000 == 0) {
- logger.warn(
- "Has waiting {} seconds for all compaction task finish", sleepingStartTime / 1000);
- }
- if (sleepingStartTime >= MAX_WAITING_TIME) {
- return;
+ candidateCompactionTaskQueue.clear();
+ while (true) {
+ int totalSize = 0;
+ for (Map<AbstractCompactionTask, Future<CompactionTaskSummary>> taskMap :
+ storageGroupTasks.values()) {
+ totalSize += taskMap.size();
Review Comment:
runningCompactionTasks
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionWorker.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.db.engine.compaction;
+
+import org.apache.iotdb.db.engine.compaction.constant.CompactionTaskStatus;
+import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
+import org.apache.iotdb.db.service.metrics.recorder.CompactionMetricsRecorder;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class CompactionWorker implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger("COMPACTION");
+ private final int threadId;
+ private final FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue;
+
+ public CompactionWorker(
+ int threadId, FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue) {
+ this.threadId = threadId;
+ this.compactionTaskQueue = compactionTaskQueue;
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ AbstractCompactionTask task = compactionTaskQueue.take();
+ // add metrics
+ CompactionMetricsRecorder.recordTaskInfo(
+ task, CompactionTaskStatus.POLL_FROM_QUEUE, compactionTaskQueue.size());
+ if (task != null && task.checkValidAndSetMerging()) {
+ CompactionTaskSummary summary = task.getSummary();
+ CompactionTaskFuture future = new CompactionTaskFuture(summary);
+ CompactionTaskManager.getInstance().recordTask(task, future);
+ task.call();
+ }
+ } catch (InterruptedException e) {
+ log.error("CompactionThread-{} terminates because interruption", threadId);
+ return;
+ } catch (Throwable t) {
+ log.error("CompactionThread-{} meets exception", threadId, t);
+ }
+ }
+ }
+
+ class CompactionTaskFuture implements Future<CompactionTaskSummary> {
Review Comment:
You could use SettableFuture.create(), without implement too much functions.
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java:
##########
@@ -440,4 +374,18 @@ public void restart() throws InterruptedException {
public void clearCandidateQueue() {
candidateCompactionTaskQueue.clear();
}
+
+ @TestOnly
+ public Future<CompactionTaskSummary> getCompactionTaskFutureMayBlock(AbstractCompactionTask task)
+ throws InterruptedException {
+ String storageGroup = task.getFullStorageGroupName();
Review Comment:
```suggestion
String regionWithSG = task.getRegionWithSG();
```
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionWorker.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.db.engine.compaction;
+
+import org.apache.iotdb.db.engine.compaction.constant.CompactionTaskStatus;
+import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
+import org.apache.iotdb.db.service.metrics.recorder.CompactionMetricsRecorder;
+import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class CompactionWorker implements Runnable {
+ private static final Logger log = LoggerFactory.getLogger("COMPACTION");
+ private final int threadId;
+ private final FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue;
+
+ public CompactionWorker(
+ int threadId, FixedPriorityBlockingQueue<AbstractCompactionTask> compactionTaskQueue) {
+ this.threadId = threadId;
+ this.compactionTaskQueue = compactionTaskQueue;
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ AbstractCompactionTask task = compactionTaskQueue.take();
+ // add metrics
+ CompactionMetricsRecorder.recordTaskInfo(
+ task, CompactionTaskStatus.POLL_FROM_QUEUE, compactionTaskQueue.size());
+ if (task != null && task.checkValidAndSetMerging()) {
+ CompactionTaskSummary summary = task.getSummary();
+ CompactionTaskFuture future = new CompactionTaskFuture(summary);
+ CompactionTaskManager.getInstance().recordTask(task, future);
+ task.call();
+ }
+ } catch (InterruptedException e) {
+ log.error("CompactionThread-{} terminates because interruption", threadId);
+ return;
+ } catch (Throwable t) {
+ log.error("CompactionThread-{} meets exception", threadId, t);
+ }
+ }
+ }
+
+ class CompactionTaskFuture implements Future<CompactionTaskSummary> {
+ CompactionTaskSummary summary;
+
+ public CompactionTaskFuture(CompactionTaskSummary summary) {
+ this.summary = summary;
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ summary.cancel();
+ return true;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return summary.isCancel();
+ }
+
+ @Override
+ public boolean isDone() {
+ return summary.isFinished();
+ }
+
+ @Override
+ public CompactionTaskSummary get() throws InterruptedException, ExecutionException {
+ while (!summary.isFinished()) {
+ TimeUnit.MILLISECONDS.sleep(100);
+ }
+ return summary;
+ }
+
+ @Override
+ public CompactionTaskSummary get(long timeout, @NotNull TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ long perSleepTime = timeout > 100 ? timeout : 100;
+ long totalSleepTime = 0L;
+ while (!summary.isFinished()) {
+ if (totalSleepTime > timeout) {
Review Comment:
```suggestion
if (totalSleepTime >= timeout) {
```
Or you will wait double time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org