You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/07/04 06:48:37 UTC
[iotdb] branch master updated: Style fix: fix warnings, sonar bugs, smells, lower cases of first letter in pipe/rpc module (#10421)
This is an automated email from the ASF dual-hosted git repository.
rong 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 13b28ddaf1c Style fix: fix warnings, sonar bugs, smells, lower cases of first letter in pipe/rpc module (#10421)
13b28ddaf1c is described below
commit 13b28ddaf1c4bb07ffd26431ee130c3747b745fa
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Jul 4 14:48:31 2023 +0800
Style fix: fix warnings, sonar bugs, smells, lower cases of first letter in pipe/rpc module (#10421)
---
.../PipeAttributeNotProvidedException.java | 2 +-
.../manager/pipe/runtime/PipeHeartbeatParser.java | 6 +-
.../manager/pipe/runtime/PipeMetaSyncer.java | 4 +-
.../confignode/persistence/pipe/PipeTaskInfo.java | 7 ++-
.../impl/pipe/task/CreatePipeProcedureV2.java | 2 +-
.../iotdb/db/pipe/agent/task/PipeTaskAgent.java | 70 +++++++++++-----------
.../pipe/connector/v1/IoTDBThriftConnectorV1.java | 8 +--
.../pipe/extractor/IoTDBDataRegionExtractor.java | 10 ++--
.../PipeHistoricalDataRegionTsFileExtractor.java | 12 ++--
.../realtime/PipeRealtimeDataRegionExtractor.java | 8 ++-
.../apache/iotdb/db/pipe/task/PipeTaskBuilder.java | 6 +-
.../db/pipe/task/stage/PipeTaskConnectorStage.java | 2 +-
.../db/pipe/task/stage/PipeTaskExtractorStage.java | 10 ++--
.../iotdb/db/pipe/task/stage/PipeTaskStage.java | 4 +-
.../iotdb/db/pipe/task/subtask/PipeSubtask.java | 27 +++++----
.../subtask/connector/PipeConnectorSubtask.java | 24 ++++----
.../connector/PipeConnectorSubtaskManager.java | 4 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 23 +++----
.../exception/pipe/PipeRuntimeException.java | 2 +-
.../builtin/connector/IoTDBThriftConnectorV1.java | 6 ++
.../builtin/connector/IoTDBThriftConnectorV2.java | 6 ++
.../commons/pipe/task/meta/PipeStaticMeta.java | 4 +-
22 files changed, 132 insertions(+), 115 deletions(-)
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeAttributeNotProvidedException.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeAttributeNotProvidedException.java
index 2a529bfdeed..e229cb2c929 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeAttributeNotProvidedException.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeAttributeNotProvidedException.java
@@ -22,6 +22,6 @@ package org.apache.iotdb.pipe.api.exception;
public class PipeAttributeNotProvidedException extends PipeParameterNotValidException {
public PipeAttributeNotProvidedException(String key) {
- super(String.format("attribute \"%s\" is required but was not provided.", key));
+ super(String.format("Attribute \"%s\" is required but was not provided.", key));
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
index 9fa394405f3..eb5d51bbd0e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
@@ -109,7 +109,7 @@ public class PipeHeartbeatParser {
.pipeHandleMetaChange(
needWriteConsensusOnConfigNodes.get(), needPushPipeMetaToDataNodes.get());
- // reset flags after procedure is submitted
+ // Reset flags after procedure is submitted
needWriteConsensusOnConfigNodes.set(false);
needPushPipeMetaToDataNodes.set(false);
}
@@ -160,7 +160,7 @@ public class PipeHeartbeatParser {
continue;
}
- // update progress index
+ // Update progress index
if (!runtimeMetaOnConfigNode
.getValue()
.getProgressIndex()
@@ -183,7 +183,7 @@ public class PipeHeartbeatParser {
needWriteConsensusOnConfigNodes.set(true);
}
- // update runtime exception
+ // Update runtime exception
final PipeTaskMeta pipeTaskMetaOnConfigNode = runtimeMetaOnConfigNode.getValue();
pipeTaskMetaOnConfigNode.clearExceptionMessages();
for (final PipeRuntimeException exception :
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
index 715dabe2a46..98471505947 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
@@ -57,11 +57,11 @@ public class PipeMetaSyncer {
public synchronized void start() {
while (configManager.getConsensusManager() == null) {
try {
- LOGGER.info("consensus layer is not ready, sleep 1s...");
+ LOGGER.info("Consensus layer is not ready, sleep 1s...");
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- LOGGER.warn("unexpected interruption during waiting for consensus layer ready.");
+ LOGGER.warn("Unexpected interruption during waiting for consensus layer ready.");
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 728e3f30792..e4c8c26a0ef 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.persistence.pipe;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
@@ -192,7 +193,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
/////////////////////////////// Pipe Runtime Management ///////////////////////////////
- /** handle the data region leader change event and update the pipe task meta accordingly */
+ /** Handle the data region leader change event and update the pipe task meta accordingly. */
public TSStatus handleLeaderChange(PipeHandleLeaderChangePlan plan) {
plan.getConsensusGroupId2NewDataRegionLeaderIdMap()
.forEach(
@@ -205,7 +206,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
pipeMeta.getRuntimeMeta().getConsensusGroupIdToTaskMetaMap();
if (consensusGroupIdToTaskMetaMap.containsKey(dataRegionGroupId)) {
- // if the data region leader is -1, it means the data region is
+ // If the data region leader is -1, it means the data region is
// removed
if (newDataRegionLeader != -1) {
consensusGroupIdToTaskMetaMap
@@ -215,7 +216,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
consensusGroupIdToTaskMetaMap.remove(dataRegionGroupId);
}
} else {
- // if CN does not contain the data region group, it means the data
+ // If CN does not contain the data region group, it means the data
// region group is newly added.
if (newDataRegionLeader != -1) {
consensusGroupIdToTaskMetaMap.put(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index 072024a577d..43ef5a344d3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -112,7 +112,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getPartitionManager()
.getRegionStorageGroup(regionGroupId);
if (databaseName != null && !databaseName.equals(IoTDBConfig.SYSTEM_DATABASE)) {
- // pipe only collect user's data, filter metric database here.
+ // Pipe only collect user's data, filter metric database here.
consensusGroupIdToTaskMetaMap.put(
regionGroupId,
new PipeTaskMeta(new MinimumProgressIndex(), regionLeaderNodeId));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
index 91cfc43b031..b18b294100e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
@@ -87,14 +87,14 @@ public class PipeTaskAgent {
////////////////////////// Pipe Task Management Entry //////////////////////////
public synchronized void handlePipeMetaChanges(List<PipeMeta> pipeMetaListFromConfigNode) {
- // do nothing if data node is removing or removed
+ // Do nothing if data node is removing or removed
if (PipeAgent.runtime().isShutdown()) {
return;
}
final List<Exception> exceptions = new ArrayList<>();
- // iterate through pipe meta list from config node, check if pipe meta exists on data node
+ // Iterate through pipe meta list from config node, check if pipe meta exists on data node
// or has changed
for (final PipeMeta metaFromConfigNode : pipeMetaListFromConfigNode) {
final String pipeName = metaFromConfigNode.getStaticMeta().getPipeName();
@@ -102,31 +102,31 @@ public class PipeTaskAgent {
try {
final PipeMeta metaOnDataNode = pipeMetaKeeper.getPipeMeta(pipeName);
- // if pipe meta does not exist on data node, create a new pipe
+ // If pipe meta does not exist on data node, create a new pipe
if (metaOnDataNode == null) {
if (createPipe(metaFromConfigNode)) {
- // if the status recorded in config node is RUNNING, start the pipe
+ // If the status recorded in config node is RUNNING, start the pipe
startPipe(pipeName, metaFromConfigNode.getStaticMeta().getCreationTime());
}
- // if the status recorded in config node is STOPPED or DROPPED, do nothing
+ // If the status recorded in config node is STOPPED or DROPPED, do nothing
continue;
}
- // if pipe meta exists on data node, check if it has changed
+ // If pipe meta exists on data node, check if it has changed
final PipeStaticMeta staticMetaOnDataNode = metaOnDataNode.getStaticMeta();
final PipeStaticMeta staticMetaFromConfigNode = metaFromConfigNode.getStaticMeta();
- // first check if pipe static meta has changed, if so, drop the pipe and create a new one
+ // First check if pipe static meta has changed, if so, drop the pipe and create a new one
if (!staticMetaOnDataNode.equals(staticMetaFromConfigNode)) {
dropPipe(pipeName);
if (createPipe(metaFromConfigNode)) {
startPipe(pipeName, metaFromConfigNode.getStaticMeta().getCreationTime());
}
- // if the status is STOPPED or DROPPED, do nothing
+ // If the status is STOPPED or DROPPED, do nothing
continue;
}
- // then check if pipe runtime meta has changed, if so, update the pipe
+ // Then check if pipe runtime meta has changed, if so, update the pipe
final PipeRuntimeMeta runtimeMetaOnDataNode = metaOnDataNode.getRuntimeMeta();
final PipeRuntimeMeta runtimeMetaFromConfigNode = metaFromConfigNode.getRuntimeMeta();
handlePipeRuntimeMetaChanges(
@@ -140,7 +140,7 @@ public class PipeTaskAgent {
}
}
- // check if there are pipes on data node that do not exist on config node, if so, drop them
+ // Check if there are pipes on data node that do not exist on config node, if so, drop them
final Set<String> pipeNamesFromConfigNode =
pipeMetaListFromConfigNode.stream()
.map(meta -> meta.getStaticMeta().getPipeName())
@@ -172,13 +172,13 @@ public class PipeTaskAgent {
@NotNull PipeStaticMeta pipeStaticMeta,
@NotNull PipeRuntimeMeta runtimeMetaFromConfigNode,
@NotNull PipeRuntimeMeta runtimeMetaOnDataNode) {
- // 1. handle data region group leader changed first
+ // 1. Handle data region group leader changed first
final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupIdToTaskMetaMapFromConfigNode =
runtimeMetaFromConfigNode.getConsensusGroupIdToTaskMetaMap();
final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupIdToTaskMetaMapOnDataNode =
runtimeMetaOnDataNode.getConsensusGroupIdToTaskMetaMap();
- // 1.1 iterate over all consensus group ids in config node's pipe runtime meta, decide if we
+ // 1.1 Iterate over all consensus group ids in config node's pipe runtime meta, decide if we
// need to drop and create a new task for each consensus group id
for (final Map.Entry<TConsensusGroupId, PipeTaskMeta> entryFromConfigNode :
consensusGroupIdToTaskMetaMapFromConfigNode.entrySet()) {
@@ -188,10 +188,10 @@ public class PipeTaskAgent {
final PipeTaskMeta taskMetaOnDataNode =
consensusGroupIdToTaskMetaMapOnDataNode.get(consensusGroupIdFromConfigNode);
- // if task meta does not exist on data node, create a new task
+ // If task meta does not exist on data node, create a new task
if (taskMetaOnDataNode == null) {
createPipeTask(consensusGroupIdFromConfigNode, pipeStaticMeta, taskMetaFromConfigNode);
- // we keep the new created task's status consistent with the status recorded in data node's
+ // We keep the new created task's status consistent with the status recorded in data node's
// pipe runtime meta. please note that the status recorded in data node's pipe runtime meta
// is not reliable, but we will have a check later to make sure the status is correct.
if (runtimeMetaOnDataNode.getStatus().get() == PipeStatus.RUNNING) {
@@ -200,14 +200,14 @@ public class PipeTaskAgent {
continue;
}
- // if task meta exists on data node, check if it has changed
+ // If task meta exists on data node, check if it has changed
final int dataNodeIdFromConfigNode = taskMetaFromConfigNode.getLeaderDataNodeId();
final int dataNodeIdOnDataNode = taskMetaOnDataNode.getLeaderDataNodeId();
if (dataNodeIdFromConfigNode != dataNodeIdOnDataNode) {
dropPipeTask(consensusGroupIdFromConfigNode, pipeStaticMeta);
createPipeTask(consensusGroupIdFromConfigNode, pipeStaticMeta, taskMetaFromConfigNode);
- // we keep the new created task's status consistent with the status recorded in data node's
+ // We keep the new created task's status consistent with the status recorded in data node's
// pipe runtime meta. please note that the status recorded in data node's pipe runtime meta
// is not reliable, but we will have a check later to make sure the status is correct.
if (runtimeMetaOnDataNode.getStatus().get() == PipeStatus.RUNNING) {
@@ -216,7 +216,7 @@ public class PipeTaskAgent {
}
}
- // 1.2 iterate over all consensus group ids on data node's pipe runtime meta, decide if we need
+ // 1.2 Iterate over all consensus group ids on data node's pipe runtime meta, decide if we need
// to drop any task. we do not need to create any new task here because we have already done
// that in 1.1.
for (final Map.Entry<TConsensusGroupId, PipeTaskMeta> entryOnDataNode :
@@ -229,7 +229,7 @@ public class PipeTaskAgent {
}
}
- // 2. handle pipe runtime meta status changes
+ // 2. Handle pipe runtime meta status changes
final PipeStatus statusFromConfigNode = runtimeMetaFromConfigNode.getStatus().get();
final PipeStatus statusOnDataNode = runtimeMetaOnDataNode.getStatus().get();
if (statusFromConfigNode == statusOnDataNode) {
@@ -256,7 +256,7 @@ public class PipeTaskAgent {
}
break;
case DROPPED:
- // this should not happen, but we still handle it here
+ // This should not happen, but we still handle it here
dropPipe(pipeStaticMeta.getPipeName(), pipeStaticMeta.getCreationTime());
break;
default:
@@ -322,7 +322,7 @@ public class PipeTaskAgent {
creationTime,
status.name());
}
- // break to drop the pipe and recreate it
+ // Break to drop the pipe and recreate it
break;
default:
throw new IllegalStateException(
@@ -331,14 +331,14 @@ public class PipeTaskAgent {
}
}
- // drop the pipe if
+ // Drop the pipe if
// 1. the pipe with the same name but with different creation time has been created before
// 2. the pipe with the same name and the same creation time has been dropped before, but the
// pipe task meta has not been cleaned up
dropPipe(pipeName, existedPipeMeta.getStaticMeta().getCreationTime());
}
- // create pipe tasks and trigger create() method for each pipe task
+ // Create pipe tasks and trigger create() method for each pipe task
final Map<TConsensusGroupId, PipeTask> pipeTasks =
new PipeBuilder(pipeMetaFromConfigNode).build();
for (PipeTask pipeTask : pipeTasks.values()) {
@@ -381,12 +381,12 @@ public class PipeTaskAgent {
return;
}
- // mark pipe meta as dropped first. this will help us detect if the pipe meta has been dropped
+ // Mark pipe meta as dropped first. this will help us detect if the pipe meta has been dropped
// but the pipe task meta has not been cleaned up (in case of failure when executing
// dropPipeTaskByConsensusGroup).
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
- // drop pipe tasks and trigger drop() method for each pipe task
+ // Drop pipe tasks and trigger drop() method for each pipe task
final Map<TConsensusGroupId, PipeTask> pipeTasks =
pipeTaskManager.removePipeTasks(existedPipeMeta.getStaticMeta());
if (pipeTasks == null) {
@@ -401,7 +401,7 @@ public class PipeTaskAgent {
pipeTask.drop();
}
- // remove pipe meta from pipe meta keeper
+ // Remove pipe meta from pipe meta keeper
pipeMetaKeeper.removePipeMeta(pipeName);
}
@@ -414,12 +414,12 @@ public class PipeTaskAgent {
return;
}
- // mark pipe meta as dropped first. this will help us detect if the pipe meta has been dropped
+ // Mark pipe meta as dropped first. this will help us detect if the pipe meta has been dropped
// but the pipe task meta has not been cleaned up (in case of failure when executing
// dropPipeTaskByConsensusGroup).
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
- // drop pipe tasks and trigger drop() method for each pipe task
+ // Drop pipe tasks and trigger drop() method for each pipe task
final Map<TConsensusGroupId, PipeTask> pipeTasks =
pipeTaskManager.removePipeTasks(existedPipeMeta.getStaticMeta());
if (pipeTasks == null) {
@@ -431,7 +431,7 @@ public class PipeTaskAgent {
pipeTask.drop();
}
- // remove pipe meta from pipe meta keeper
+ // Remove pipe meta from pipe meta keeper
pipeMetaKeeper.removePipeMeta(pipeName);
}
@@ -493,7 +493,7 @@ public class PipeTaskAgent {
+ existedPipeMeta.getRuntimeMeta().getStatus().get().name());
}
- // trigger start() method for each pipe task
+ // Trigger start() method for each pipe task
final Map<TConsensusGroupId, PipeTask> pipeTasks =
pipeTaskManager.getPipeTasks(existedPipeMeta.getStaticMeta());
if (pipeTasks == null) {
@@ -508,9 +508,9 @@ public class PipeTaskAgent {
pipeTask.start();
}
- // set pipe meta status to RUNNING
+ // Set pipe meta status to RUNNING
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
- // clear exception messages if started successfully
+ // Clear exception messages if started successfully
existedPipeMeta
.getRuntimeMeta()
.getConsensusGroupIdToTaskMetaMap()
@@ -574,7 +574,7 @@ public class PipeTaskAgent {
throw new IllegalStateException(MESSAGE_UNEXPECTED_PIPE_STATUS + status.name());
}
- // trigger stop() method for each pipe task
+ // Trigger stop() method for each pipe task
final Map<TConsensusGroupId, PipeTask> pipeTasks =
pipeTaskManager.getPipeTasks(existedPipeMeta.getStaticMeta());
if (pipeTasks == null) {
@@ -589,7 +589,7 @@ public class PipeTaskAgent {
pipeTask.stop();
}
- // set pipe meta status to STOPPED
+ // Set pipe meta status to STOPPED
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
}
@@ -635,7 +635,7 @@ public class PipeTaskAgent {
public synchronized void collectPipeMetaList(THeartbeatReq req, THeartbeatResp resp)
throws TException {
- // do nothing if data node is removing or removed, or request does not need pipe meta list
+ // Do nothing if data node is removing or removed, or request does not need pipe meta list
if (PipeAgent.runtime().isShutdown() || !req.isNeedPipeMetaList()) {
return;
}
@@ -654,7 +654,7 @@ public class PipeTaskAgent {
public synchronized void collectPipeMetaList(TPipeHeartbeatReq req, TPipeHeartbeatResp resp)
throws TException {
- // do nothing if data node is removing or removed, or request does not need pipe meta list
+ // Do nothing if data node is removing or removed, or request does not need pipe meta list
if (PipeAgent.runtime().isShutdown()) {
return;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index b329d9b2014..cc13c4f1447 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -116,7 +116,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
@Override
public void heartbeat() throws Exception {
- // do nothing
+ // Do nothing
}
@Override
@@ -202,7 +202,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
final File tsFile = pipeTsFileInsertionEvent.getTsFile();
- // 1. transfer file piece by piece
+ // 1. Transfer file piece by piece
final int readFileBufferSize = PipeConfig.getInstance().getPipeConnectorReadFileBufferSize();
final byte[] readBuffer = new byte[readFileBufferSize];
long position = 0;
@@ -224,7 +224,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
: Arrays.copyOfRange(readBuffer, 0, readLength))));
position += readLength;
- // this case only happens when the connection is broken, and the connector is reconnected
+ // This case only happens when the connection is broken, and the connector is reconnected
// to the receiver, then the receiver will redirect the file position to the last position
if (resp.getStatus().getCode()
== TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET.getStatusCode()) {
@@ -241,7 +241,7 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
}
}
- // 2. transfer file seal signal, which means the file is transferred completely
+ // 2. Transfer file seal signal, which means the file is transferred completely
final TPipeTransferResp resp =
client.pipeTransfer(
PipeTransferFileSealReq.toTPipeTransferReq(tsFile.getName(), tsFile.length()));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
index 4b3999c06bb..580b31cde47 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
@@ -66,7 +66,7 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
@Override
public void validate(PipeParameterValidator validator) throws Exception {
- // validate extractor.history.enable and extractor.realtime.enable
+ // Validate extractor.history.enable and extractor.realtime.enable
validator
.validateAttributeValueRange(
EXTRACTOR_HISTORY_ENABLE_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString())
@@ -80,7 +80,7 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
validator.getParameters().getBooleanOrDefault(EXTRACTOR_HISTORY_ENABLE_KEY, true),
validator.getParameters().getBooleanOrDefault(EXTRACTOR_REALTIME_ENABLE, true));
- // validate extractor.realtime.mode
+ // Validate extractor.realtime.mode
if (validator.getParameters().getBooleanOrDefault(EXTRACTOR_REALTIME_ENABLE, true)) {
validator.validateAttributeValueRange(
EXTRACTOR_REALTIME_MODE,
@@ -98,18 +98,18 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
}
private void constructHistoricalExtractor() {
- // enable historical extractor by default
+ // Enable historical extractor by default
historicalExtractor = new PipeHistoricalDataRegionTsFileExtractor();
}
private void constructRealtimeExtractor(PipeParameters parameters) {
- // enable realtime extractor by default
+ // Enable realtime extractor by default
if (!parameters.getBooleanOrDefault(EXTRACTOR_REALTIME_ENABLE, true)) {
realtimeExtractor = new PipeRealtimeDataRegionFakeExtractor();
return;
}
- // use hybrid mode by default
+ // Use hybrid mode by default
if (!parameters.hasAttribute(EXTRACTOR_REALTIME_MODE)) {
realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
index ce60ba5ebd4..a6c6b4d60ee 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -62,16 +62,16 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
private String pattern;
- private long historicalDataExtractionStartTime; // event time
- private long historicalDataExtractionEndTime; // event time
+ private long historicalDataExtractionStartTime; // Event time
+ private long historicalDataExtractionEndTime; // Event time
- private long historicalDataExtractionTimeLowerBound; // arrival time
+ private long historicalDataExtractionTimeLowerBound; // Arrival time
private Queue<PipeTsFileInsertionEvent> pendingQueue;
@Override
public void validate(PipeParameterValidator validator) {
- // do nothing
+ // Do nothing
}
@Override
@@ -87,7 +87,7 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
pattern = parameters.getStringOrDefault(EXTRACTOR_PATTERN_KEY, EXTRACTOR_PATTERN_DEFAULT_VALUE);
- // user may set the EXTRACTOR_HISTORY_START_TIME and EXTRACTOR_HISTORY_END_TIME without
+ // User may set the EXTRACTOR_HISTORY_START_TIME and EXTRACTOR_HISTORY_END_TIME without
// enabling the historical data extraction, which may affect the realtime data extraction.
final boolean isHistoricalExtractorEnabledByUser =
parameters.getBooleanOrDefault(EXTRACTOR_HISTORY_ENABLE_KEY, true);
@@ -102,7 +102,7 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
parameters.getString(EXTRACTOR_HISTORY_END_TIME), ZoneId.systemDefault())
: Long.MAX_VALUE;
- // enable historical extractor by default
+ // Enable historical extractor by default
historicalDataExtractionTimeLowerBound =
parameters.getBooleanOrDefault(EXTRACTOR_HISTORY_ENABLE_KEY, true)
? Long.MIN_VALUE
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionExtractor.java
index aa0346aace9..9fd72fb7b67 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionExtractor.java
@@ -35,10 +35,14 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
protected String dataRegionId;
protected PipeTaskMeta pipeTaskMeta;
- protected PipeRealtimeDataRegionExtractor() {}
+ protected PipeRealtimeDataRegionExtractor() {
+ // Do nothing
+ }
@Override
- public void validate(PipeParameterValidator validator) throws Exception {}
+ public void validate(PipeParameterValidator validator) throws Exception {
+ // Do nothing
+ }
@Override
public void customize(PipeParameters parameters, PipeExtractorRuntimeConfiguration configuration)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeTaskBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeTaskBuilder.java
index 47353d89ecf..57ed8834a72 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeTaskBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeTaskBuilder.java
@@ -40,9 +40,9 @@ public class PipeTaskBuilder {
}
public PipeTask build() {
- // event flow: extractor -> processor -> connector
+ // Event flow: extractor -> processor -> connector
- // we first build the extractor and connector, then build the processor.
+ // We first build the extractor and connector, then build the processor.
final PipeTaskExtractorStage extractorStage =
new PipeTaskExtractorStage(
pipeStaticMeta.getPipeName(),
@@ -57,7 +57,7 @@ public class PipeTaskBuilder {
pipeStaticMeta.getCreationTime(),
pipeStaticMeta.getConnectorParameters());
- // the processor connects the extractor and connector.
+ // The processor connects the extractor and connector.
final PipeTaskProcessorStage processorStage =
new PipeTaskProcessorStage(
pipeStaticMeta.getPipeName(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
index 0bd41142395..131b2e7c32e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
@@ -47,7 +47,7 @@ public class PipeTaskConnectorStage extends PipeTaskStage {
@Override
public void createSubtask() throws PipeException {
- // do nothing
+ // Do nothing
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
index 0b86fbccacc..75c5955c491 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskExtractorStage.java
@@ -52,13 +52,13 @@ public class PipeTaskExtractorStage extends PipeTaskStage {
? new IoTDBDataRegionExtractor()
: PipeAgent.plugin().reflectExtractor(extractorParameters);
- // validate and customize should be called before createSubtask. this allows extractor exposing
+ // Validate and customize should be called before createSubtask. this allows extractor exposing
// exceptions in advance.
try {
- // 1. validate extractor parameters
+ // 1. Validate extractor parameters
pipeExtractor.validate(new PipeParameterValidator(extractorParameters));
- // 2. customize extractor
+ // 2. Customize extractor
final PipeTaskRuntimeConfiguration runtimeConfiguration =
new PipeTaskRuntimeConfiguration(
new PipeTaskExtractorRuntimeEnvironment(
@@ -71,7 +71,7 @@ public class PipeTaskExtractorStage extends PipeTaskStage {
@Override
public void createSubtask() throws PipeException {
- // do nothing
+ // Do nothing
}
@Override
@@ -85,7 +85,7 @@ public class PipeTaskExtractorStage extends PipeTaskStage {
@Override
public void stopSubtask() throws PipeException {
- // extractor continuously extracts data, so do nothing in stop
+ // Extractor continuously extracts data, so do nothing in stop
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskStage.java
index c2b4426435d..10b35b6b380 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskStage.java
@@ -53,11 +53,11 @@ public abstract class PipeTaskStage {
if (hasBeenExternallyStopped) {
throw new PipeException(MESSAGE_PIPE_TASK_STAGE_HAS_BEEN_STOPPED);
}
- // otherwise, do nothing to allow retry strategy
+ // Otherwise, do nothing to allow retry strategy
return;
}
- // status == null, register the subtask
+ // Status == null, register the subtask
createSubtask();
status = PipeStatus.STOPPED;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
index fe4172df5e9..4531ed54e98 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
@@ -42,17 +42,17 @@ public abstract class PipeSubtask
private static final Logger LOGGER = LoggerFactory.getLogger(PipeSubtask.class);
- // used for identifying the subtask
+ // Used for identifying the subtask
protected final String taskID;
- // for thread pool to execute subtasks
+ // For thread pool to execute subtasks
protected ListeningExecutorService subtaskWorkerThreadPoolExecutor;
- // for controlling the subtask execution
+ // For controlling the subtask execution
protected final AtomicBoolean shouldStopSubmittingSelf = new AtomicBoolean(true);
protected PipeSubtaskScheduler subtaskScheduler;
- // for fail-over
+ // For fail-over
public static final int MAX_RETRY_TIMES = 5;
protected final AtomicInteger retryCount = new AtomicInteger(0);
protected Event lastEvent;
@@ -71,28 +71,28 @@ public abstract class PipeSubtask
public Boolean call() throws Exception {
boolean hasAtLeastOneEventProcessed = false;
- // if the scheduler allows to schedule, then try to consume an event
+ // If the scheduler allows to schedule, then try to consume an event
while (subtaskScheduler.schedule()) {
- // if the event is consumed successfully, then continue to consume the next event
+ // If the event is consumed successfully, then continue to consume the next event
// otherwise, stop consuming
if (!executeOnce()) {
break;
}
hasAtLeastOneEventProcessed = true;
}
- // reset the scheduler to make sure that the scheduler can schedule again
+ // Reset the scheduler to make sure that the scheduler can schedule again
subtaskScheduler.reset();
return hasAtLeastOneEventProcessed;
}
/**
- * try to consume an event by the pipe plugin.
+ * Try to consume an event by the pipe plugin.
*
* @return true if the event is consumed successfully, false if no more event can be consumed
* @throws Exception if any error occurs when consuming the event
*/
- @SuppressWarnings("squid:S112") // allow to throw Exception
+ @SuppressWarnings("squid:S112") // Allow to throw Exception
protected abstract boolean executeOnce() throws Exception;
@Override
@@ -157,9 +157,9 @@ public abstract class PipeSubtask
throwable);
}
- // although the pipe task will be stopped, we still don't release the last event here
- // because we need to keep it for the next retry. if user wants to restart the task,
- // the last event will be processed again. the last event will be released when the task
+ // Although the pipe task will be stopped, we still don't release the last event here
+ // Because we need to keep it for the next retry. If user wants to restart the task,
+ // the last event will be processed again. The last event will be released when the task
// is dropped or the process is running normally.
}
}
@@ -172,6 +172,9 @@ public abstract class PipeSubtask
}
/**
+ * Set the shouldStopSubmittingSelf state from false to true, in order to stop submitting the
+ * subtask.
+ *
* @return true if the shouldStopSubmittingSelf state is changed from false to true, false
* otherwise
*/
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
index 1b9cd12ff90..5aeb59eeca4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
@@ -47,15 +47,15 @@ public class PipeConnectorSubtask extends PipeSubtask {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeConnectorSubtask.class);
- // for input and output
+ // For input and output
private final BoundedBlockingPendingQueue<Event> inputPendingQueue;
private final PipeConnector outputPipeConnector;
- // for heartbeat scheduling
+ // For heartbeat scheduling
private static final int HEARTBEAT_CHECK_INTERVAL = 1000;
private int executeOnceInvokedTimes;
- // for thread pool to execute callbacks
+ // For thread pool to execute callbacks
protected final DecoratingLock callbackDecoratingLock = new DecoratingLock();
protected ExecutorService subtaskCallbackListeningExecutor;
@@ -83,7 +83,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
public Boolean call() throws Exception {
final boolean hasAtLeastOneEventProcessed = super.call();
- // wait for the callable to be decorated by Futures.addCallback in the executorService
+ // Wait for the callable to be decorated by Futures.addCallback in the executorService
// to make sure that the callback can be submitted again on success or failure.
callbackDecoratingLock.waitForDecorated();
@@ -102,7 +102,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
}
final Event event = lastEvent != null ? lastEvent : inputPendingQueue.waitedPoll();
- // record this event for retrying on connection failure or other exceptions
+ // Record this event for retrying on connection failure or other exceptions
lastEvent = event;
if (event == null) {
return false;
@@ -133,7 +133,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
@Override
public void onFailure(@NotNull Throwable throwable) {
- // retry to connect to the target system if the connection is broken
+ // Retry to connect to the target system if the connection is broken
if (throwable instanceof PipeConnectionException) {
LOGGER.warn(
"PipeConnectionException occurred, retrying to connect to the target system...",
@@ -165,7 +165,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
}
}
- // stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
+ // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
// times
if (retry == MAX_RETRY_TIMES) {
if (lastEvent instanceof EnrichedEvent) {
@@ -194,12 +194,12 @@ public class PipeConnectorSubtask extends PipeSubtask {
// FIXME: non-EnrichedEvent should be reported to the ConfigNode instead of being logged
}
- // although the pipe task will be stopped, we still don't release the last event here
- // because we need to keep it for the next retry. if user wants to restart the task,
- // the last event will be processed again. the last event will be released when the task
+ // Although the pipe task will be stopped, we still don't release the last event here
+ // Because we need to keep it for the next retry. If user wants to restart the task,
+ // the last event will be processed again. The last event will be released when the task
// is dropped or the process is running normally.
- // stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
+ // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
return;
}
} else {
@@ -209,7 +209,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
throwable);
}
- // handle other exceptions as usual
+ // Handle other exceptions as usual
super.onFailure(new PipeRuntimeConnectorCriticalException(throwable.getMessage()));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
index bfaa8f47fff..d23e60e8243 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtaskManager.java
@@ -56,7 +56,7 @@ public class PipeConnectorSubtaskManager {
new TreeMap<>(pipeConnectorParameters.getAttribute()).toString();
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
- // 1. construct, validate and customize PipeConnector, and then handshake (create connection)
+ // 1. Construct, validate and customize PipeConnector, and then handshake (create connection)
// with the target
final String connectorKey =
pipeConnectorParameters.getStringOrDefault(
@@ -86,7 +86,7 @@ public class PipeConnectorSubtaskManager {
"Failed to construct PipeConnector, because of " + e.getMessage(), e);
}
- // 2. construct PipeConnectorSubtaskLifeCycle to manage PipeConnectorSubtask's life cycle
+ // 2. Construct PipeConnectorSubtaskLifeCycle to manage PipeConnectorSubtask's life cycle
final BoundedBlockingPendingQueue<Event> pendingQueue =
new BoundedBlockingPendingQueue<>(
PipeConfig.getInstance().getPipeConnectorPendingQueueSize());
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 a11f92d9372..5c0c71dfde9 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
@@ -503,8 +503,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus invalidateMatchedSchemaCache(TInvalidateMatchedSchemaCacheReq req)
- throws TException {
+ public TSStatus invalidateMatchedSchemaCache(TInvalidateMatchedSchemaCacheReq req) {
DataNodeSchemaCache cache = DataNodeSchemaCache.getInstance();
cache.takeWriteLock();
try {
@@ -517,8 +516,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TFetchSchemaBlackListResp fetchSchemaBlackList(TFetchSchemaBlackListReq req)
- throws TException {
+ public TFetchSchemaBlackListResp fetchSchemaBlackList(TFetchSchemaBlackListReq req) {
PathPatternTree patternTree = PathPatternTree.deserialize(req.pathPatternTree);
TFetchSchemaBlackListResp resp = new TFetchSchemaBlackListResp();
PathPatternTree result = new PathPatternTree();
@@ -548,7 +546,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
try {
result.serialize(dataOutputStream);
} catch (IOException ignored) {
- // won't reach here
+ // Won't reach here
}
resp.setPathPatternTree(outputStream.toByteArray());
return resp;
@@ -636,7 +634,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
try {
result.put(new PartialPath(k), v);
} catch (IllegalPathException ignored) {
- // won't reach here
+ // Won't reach here
}
});
return result;
@@ -668,14 +666,13 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
.getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
.getDatabaseFullPath());
} catch (IllegalPathException ignored) {
- // won't reach here
+ // Won't reach here
}
return storageGroupPath;
}
@Override
- public TSStatus rollbackSchemaBlackListWithTemplate(TRollbackSchemaBlackListWithTemplateReq req)
- throws TException {
+ public TSStatus rollbackSchemaBlackListWithTemplate(TRollbackSchemaBlackListWithTemplateReq req) {
Map<PartialPath, List<Integer>> templateSetInfo =
transformTemplateSetInfo(req.getTemplateSetInfo());
return executeInternalSchemaTask(
@@ -819,8 +816,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus constructViewSchemaBlackList(TConstructViewSchemaBlackListReq req)
- throws TException {
+ public TSStatus constructViewSchemaBlackList(TConstructViewSchemaBlackListReq req) {
PathPatternTree patternTree =
PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
AtomicInteger preDeletedNum = new AtomicInteger(0);
@@ -855,8 +851,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus rollbackViewSchemaBlackList(TRollbackViewSchemaBlackListReq req)
- throws TException {
+ public TSStatus rollbackViewSchemaBlackList(TRollbackViewSchemaBlackListReq req) {
PathPatternTree patternTree =
PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
return executeInternalSchemaTask(
@@ -904,7 +899,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus alterView(TAlterViewReq req) throws TException {
+ public TSStatus alterView(TAlterViewReq req) {
List<TConsensusGroupId> consensusGroupIdList = req.getSchemaRegionIdList();
List<ByteBuffer> viewBinaryList = req.getViewBinaryList();
Map<TConsensusGroupId, Map<PartialPath, ViewExpression>> schemaRegionRequestMap =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
index 9dd4dec5bac..8724b4b20df 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeException.java
@@ -28,7 +28,7 @@ import java.util.Objects;
public abstract class PipeRuntimeException extends PipeException {
- public PipeRuntimeException(String message) {
+ protected PipeRuntimeException(String message) {
super(message);
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java
index e708767aac6..38e09063e45 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java
@@ -19,4 +19,10 @@
package org.apache.iotdb.commons.pipe.plugin.builtin.connector;
+/**
+ * This class is a placeholder and should not be initialized. It represents the IoTDB Thrift
+ * connector V1. There is a real implementation in the server module but cannot be imported here.
+ * The pipe agent in the server module will replace this class with the real implementation when
+ * initializing the IoTDB Thrift connector V1.
+ */
public class IoTDBThriftConnectorV1 extends IoTDBThriftConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java
index c5aa3710ad2..e1019311b2b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java
@@ -19,4 +19,10 @@
package org.apache.iotdb.commons.pipe.plugin.builtin.connector;
+/**
+ * This class is a placeholder and should not be initialized. It represents the IoTDB Thrift
+ * connector V2. There is a real implementation in the server module but cannot be imported here.
+ * The pipe agent in the server module will replace this class with the real implementation when
+ * initializing the IoTDB Thrift connector V2.
+ */
public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
index 64ba11d1827..82e03ff03bc 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
@@ -41,7 +41,9 @@ public class PipeStaticMeta {
private PipeParameters processorParameters;
private PipeParameters connectorParameters;
- private PipeStaticMeta() {}
+ private PipeStaticMeta() {
+ // Do nothing
+ }
public PipeStaticMeta(
String pipeName,