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 2024/09/18 11:39:39 UTC
(iotdb) branch pipe-startup-failure created (now c7c7a8874e6)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-startup-failure
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at c7c7a8874e6 Pipe: Fix startup failure of PipeHistoricalDataRegionTsFileExtractor due to unprepared StorageEngine (#13526)
This branch includes the following new commits:
new c7c7a8874e6 Pipe: Fix startup failure of PipeHistoricalDataRegionTsFileExtractor due to unprepared StorageEngine (#13526)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Pipe: Fix startup failure of PipeHistoricalDataRegionTsFileExtractor due to unprepared StorageEngine (#13526)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-startup-failure
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c7c7a8874e6e52608365265d89d5cafe09e4555b
Author: Zhenyu Luo <lu...@gmail.com>
AuthorDate: Wed Sep 18 19:35:18 2024 +0800
Pipe: Fix startup failure of PipeHistoricalDataRegionTsFileExtractor due to unprepared StorageEngine (#13526)
Co-authored-by: Steve Yurong Su <ro...@apache.org>
(cherry picked from commit 07f14753418def64d38228f0a16499a7d7408c3b)
---
.../PipeHistoricalDataRegionTsFileExtractor.java | 23 +++++++++++++++++-----
1 file changed, 18 insertions(+), 5 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
index 310ed2640d3..98280f8b9ec 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -120,6 +120,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
private boolean shouldTerminatePipeOnAllHistoricalEventsConsumed;
private boolean isTerminateSignalSent = false;
+ private volatile boolean hasBeenStarted = false;
+
private Queue<TsFileResource> pendingQueue;
@Override
@@ -369,12 +371,18 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
@Override
public synchronized void start() {
- if (!StorageEngine.getInstance().isReadyForNonReadWriteFunctions()) {
+ if (!shouldExtractInsertion) {
+ hasBeenStarted = true;
return;
}
- if (!shouldExtractInsertion) {
+ if (!StorageEngine.getInstance().isReadyForNonReadWriteFunctions()) {
+ LOGGER.info(
+ "Pipe {}@{}: failed to start to extract historical TsFile, storage engine is not ready. Will retry later.",
+ pipeName,
+ dataRegionId);
return;
}
+ hasBeenStarted = true;
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId));
@@ -570,6 +578,10 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
@Override
public synchronized Event supply() {
+ if (!hasBeenStarted && StorageEngine.getInstance().isReadyForNonReadWriteFunctions()) {
+ start();
+ }
+
if (Objects.isNull(pendingQueue)) {
return null;
}
@@ -639,9 +651,10 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
public synchronized boolean hasConsumedAll() {
// If the pendingQueue is null when the function is called, it implies that the extractor only
// extracts deletion thus the historical event has nothing to consume.
- return Objects.isNull(pendingQueue)
- || pendingQueue.isEmpty()
- && (!shouldTerminatePipeOnAllHistoricalEventsConsumed || isTerminateSignalSent);
+ return hasBeenStarted
+ && (Objects.isNull(pendingQueue)
+ || pendingQueue.isEmpty()
+ && (!shouldTerminatePipeOnAllHistoricalEventsConsumed || isTerminateSignalSent));
}
@Override