Skip to content

Commit fe0b45a

Browse files
authored
fixup! Pipe: split full-sync pipe into history and realtime pipes (#16250) (#16316) (#16317)
1 parent 4945fbc commit fe0b45a

1 file changed

Lines changed: 20 additions & 20 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1808,58 +1808,58 @@ public SettableFuture<ConfigTaskResult> createPipe(CreatePipeStatement createPip
18081808
&& PipeDataNodeAgent.task().isFullSync(extractorPipeParameters)) {
18091809
try (final ConfigNodeClient configNodeClient =
18101810
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
1811-
// 1. Send request to create the historical data synchronization pipeline
1812-
final TCreatePipeReq historyReq =
1811+
// 1. Send request to create the real-time data synchronization pipeline
1812+
final TCreatePipeReq realtimeReq =
18131813
new TCreatePipeReq()
1814-
// Append suffix to the pipeline name for historical data
1815-
.setPipeName(createPipeStatement.getPipeName() + "_history")
1814+
// Append suffix to the pipeline name for real-time data
1815+
.setPipeName(createPipeStatement.getPipeName() + "_realtime")
18161816
// NOTE: set if not exists always to true to handle partial failure
18171817
.setIfNotExistsCondition(true)
1818-
// Use extractor parameters for historical data
1818+
// Use extractor parameters for real-time data
18191819
.setExtractorAttributes(
18201820
extractorPipeParameters
18211821
.addOrReplaceEquivalentAttributesWithClone(
18221822
new PipeParameters(
18231823
ImmutableMap.of(
1824-
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
1825-
Boolean.toString(true),
18261824
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
1825+
Boolean.toString(true),
1826+
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
18271827
Boolean.toString(false))))
18281828
.getAttribute())
18291829
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
18301830
.setConnectorAttributes(createPipeStatement.getConnectorAttributes());
18311831

1832-
final TSStatus historyTsStatus = configNodeClient.createPipe(historyReq);
1832+
final TSStatus realtimeTsStatus = configNodeClient.createPipe(realtimeReq);
18331833
// If creation fails, immediately return with exception
1834-
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != historyTsStatus.getCode()) {
1835-
future.setException(new IoTDBException(historyTsStatus));
1834+
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != realtimeTsStatus.getCode()) {
1835+
future.setException(new IoTDBException(realtimeTsStatus));
18361836
return future;
18371837
}
18381838

1839-
// 2. Send request to create the real-time data synchronization pipeline
1840-
final TCreatePipeReq realtimeReq =
1839+
// 2. Send request to create the historical data synchronization pipeline
1840+
final TCreatePipeReq historyReq =
18411841
new TCreatePipeReq()
1842-
// Append suffix to the pipeline name for real-time data
1843-
.setPipeName(createPipeStatement.getPipeName() + "_realtime")
1842+
// Append suffix to the pipeline name for historical data
1843+
.setPipeName(createPipeStatement.getPipeName() + "_history")
18441844
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
1845-
// Use extractor parameters for real-time data
1845+
// Use extractor parameters for historical data
18461846
.setExtractorAttributes(
18471847
extractorPipeParameters
18481848
.addOrReplaceEquivalentAttributesWithClone(
18491849
new PipeParameters(
18501850
ImmutableMap.of(
1851-
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
1852-
Boolean.toString(false),
18531851
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
1852+
Boolean.toString(false),
1853+
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
18541854
Boolean.toString(true))))
18551855
.getAttribute())
18561856
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
18571857
.setConnectorAttributes(createPipeStatement.getConnectorAttributes());
18581858

1859-
final TSStatus realtimeTsStatus = configNodeClient.createPipe(realtimeReq);
1859+
final TSStatus historyTsStatus = configNodeClient.createPipe(historyReq);
18601860
// If creation fails, immediately return with exception
1861-
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != realtimeTsStatus.getCode()) {
1862-
future.setException(new IoTDBException(realtimeTsStatus));
1861+
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != historyTsStatus.getCode()) {
1862+
future.setException(new IoTDBException(historyTsStatus));
18631863
return future;
18641864
}
18651865

0 commit comments

Comments
 (0)