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 3adb2e079cc Pipe: Disable batch mode for retry connector in case some
events are never retried & Pipe won't collect TsFile without tail magic & Make
async connector stop properly when retry count exceeds limit (#11425)
3adb2e079cc is described below
commit 3adb2e079ccbadfb344ad77aa07ecc1f649406ba
Author: Zikun Ma <[email protected]>
AuthorDate: Tue Oct 31 10:13:34 2023 +0800
Pipe: Disable batch mode for retry connector in case some events are never
retried & Pipe won't collect TsFile without tail magic & Make async connector
stop properly when retry count exceeds limit (#11425)
This commit includes:
* Disable batch mode for retry connector in case some events are never
retried
* Pipe won't collect TsFile without tail magic
* Make async connector stop properly when retry count exceeds limit
(exception will be properly reported by HeartbeatEvents)
---
.../protocol/airgap/IoTDBAirGapConnector.java | 11 ++++++---
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 11 ++++++---
.../thrift/async/IoTDBThriftAsyncConnector.java | 28 +++++++++++++++-------
.../thrift/sync/IoTDBThriftSyncConnector.java | 11 ++++++---
.../apache/iotdb/db/pipe/event/EnrichedEvent.java | 2 ++
.../event/common/heartbeat/PipeHeartbeatEvent.java | 11 ++++++---
.../common/tsfile/PipeTsFileInsertionEvent.java | 9 ++++++-
.../dataregion/memtable/TsFileProcessor.java | 12 ++++++++++
8 files changed, 74 insertions(+), 21 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
index 8beb6f4d9bc..f7c7d26f94c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
@@ -240,6 +240,13 @@ public class IoTDBAirGapConnector extends IoTDBConnector {
return;
}
+ if (!((PipeTsFileInsertionEvent)
tsFileInsertionEvent).waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent).getTsFile());
+ return;
+ }
+
if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
try {
for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
@@ -307,9 +314,7 @@ public class IoTDBAirGapConnector extends IoTDBConnector {
}
private void doTransfer(Socket socket, PipeTsFileInsertionEvent
pipeTsFileInsertionEvent)
- throws PipeException, InterruptedException, IOException {
- pipeTsFileInsertionEvent.waitForTsFileClose();
-
+ throws PipeException, IOException {
final File tsFile = pipeTsFileInsertionEvent.getTsFile();
// 1. Transfer file piece by piece
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
index f8a05de6f42..af607cdb06d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
@@ -206,6 +206,13 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
"IoTDBLegacyPipeConnector only support PipeTsFileInsertionEvent.");
}
+ if (!((PipeTsFileInsertionEvent)
tsFileInsertionEvent).waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent).getTsFile());
+ return;
+ }
+
try {
doTransfer((PipeTsFileInsertionEvent) tsFileInsertionEvent);
} catch (TException e) {
@@ -244,9 +251,7 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
}
private void doTransfer(PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
- throws PipeException, TException, InterruptedException, IOException {
- pipeTsFileInsertionEvent.waitForTsFileClose();
-
+ throws PipeException, TException, IOException {
final File tsFile = pipeTsFileInsertionEvent.getTsFile();
transportSingleFilePieceByPiece(tsFile);
client.sendPipeData(ByteBuffer.wrap(new TsFilePipeData("",
tsFile.getName(), -1).serialize()));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
index 84b8865604c..84a13d33a74 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
@@ -63,6 +63,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.Optional;
import java.util.PriorityQueue;
import java.util.concurrent.PriorityBlockingQueue;
@@ -70,6 +71,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_BATCH_MODE_ENABLE_KEY;
+
public class IoTDBThriftAsyncConnector extends IoTDBConnector {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBThriftAsyncConnector.class);
@@ -119,7 +122,10 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
throws Exception {
super.customize(parameters, configuration);
- retryConnector.customize(parameters, configuration);
+ // Disable batch mode for retry connector, in case retry events are never
sent again
+ PipeParameters retryParameters = new PipeParameters(new
HashMap<>(parameters.getAttribute()));
+ retryParameters.getAttribute().put(SINK_IOTDB_BATCH_MODE_ENABLE_KEY,
"false");
+ retryConnector.customize(retryParameters, configuration);
if (isTabletBatchModeEnabled) {
tabletBatchBuilder = new
IoTDBThriftAsyncPipeTransferBatchReqBuilder(parameters);
@@ -291,26 +297,32 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
return;
}
- if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
+ final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
+ (PipeTsFileInsertionEvent) tsFileInsertionEvent;
+ if (!pipeTsFileInsertionEvent.waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
+ pipeTsFileInsertionEvent.getTsFile());
+ return;
+ }
+
+ if ((pipeTsFileInsertionEvent).shouldParsePatternOrTime()) {
try {
- for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
+ for (final TabletInsertionEvent event :
+ pipeTsFileInsertionEvent.toTabletInsertionEvents()) {
transfer(event);
}
} finally {
- tsFileInsertionEvent.close();
+ pipeTsFileInsertionEvent.close();
}
return;
}
final long requestCommitId = commitIdGenerator.incrementAndGet();
-
- final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
- (PipeTsFileInsertionEvent) tsFileInsertionEvent;
final PipeTransferTsFileInsertionEventHandler
pipeTransferTsFileInsertionEventHandler =
new PipeTransferTsFileInsertionEventHandler(
requestCommitId, pipeTsFileInsertionEvent, this);
- pipeTsFileInsertionEvent.waitForTsFileClose();
transfer(requestCommitId, pipeTransferTsFileInsertionEventHandler);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
index dfecd2170b3..dd74f2dd7f5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
@@ -231,6 +231,13 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
return;
}
+ if (!((PipeTsFileInsertionEvent)
tsFileInsertionEvent).waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent).getTsFile());
+ return;
+ }
+
if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
transfer(event);
@@ -327,9 +334,7 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
private void doTransfer(
IoTDBThriftSyncConnectorClient client, PipeTsFileInsertionEvent
pipeTsFileInsertionEvent)
- throws PipeException, TException, InterruptedException, IOException {
- pipeTsFileInsertionEvent.waitForTsFileClose();
-
+ throws PipeException, TException, IOException {
final File tsFile = pipeTsFileInsertionEvent.getTsFile();
// 1. Transfer file piece by piece
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
index 9be4867afa5..801b92cc51f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
@@ -181,6 +181,8 @@ public abstract class EnrichedEvent implements Event {
public void reportException(PipeRuntimeException pipeRuntimeException) {
if (pipeTaskMeta != null) {
PipeAgent.runtime().report(pipeTaskMeta, pipeRuntimeException);
+ } else {
+ LOGGER.warn("Attempt to report pipe exception to a null PipeTaskMeta.",
pipeRuntimeException);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 9c1ad160f39..b5e92f69c55 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -73,8 +73,12 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
this.shouldPrintMessage = shouldPrintMessage;
}
- public PipeHeartbeatEvent(String dataRegionId, long timePublished, boolean
shouldPrintMessage) {
- super(null, null);
+ public PipeHeartbeatEvent(
+ PipeTaskMeta pipeTaskMeta,
+ String dataRegionId,
+ long timePublished,
+ boolean shouldPrintMessage) {
+ super(pipeTaskMeta, null);
this.dataRegionId = dataRegionId;
this.timePublished = timePublished;
this.shouldPrintMessage = shouldPrintMessage;
@@ -103,7 +107,8 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
@Override
public EnrichedEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport(
PipeTaskMeta pipeTaskMeta, String pattern) {
- return new PipeHeartbeatEvent(dataRegionId, timePublished,
shouldPrintMessage);
+ // Should record PipeTaskMeta, for sometimes HeartbeatEvents should report
exceptions.
+ return new PipeHeartbeatEvent(pipeTaskMeta, dataRegionId, timePublished,
shouldPrintMessage);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 43f439a45b4..0acefd60f54 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -45,6 +45,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
implements TsFileIns
private final long startTime;
private final long endTime;
private final boolean needParseTime;
+ private boolean isTsFileFormatValid = true;
private final TsFileResource resource;
private File tsFile;
@@ -93,6 +94,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
implements TsFileIns
processor.addCloseFileListener(
o -> {
synchronized (isClosed) {
+ isTsFileFormatValid = o.isTsFileFormatValidForPipe();
isClosed.set(true);
isClosed.notifyAll();
}
@@ -103,7 +105,11 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent implements TsFileIns
isClosed.set(resource.isClosed());
}
- public void waitForTsFileClose() throws InterruptedException {
+ /**
+ * @return {@code false} if this file can't be sent by pipe due to format
violations. {@code true}
+ * otherwise.
+ */
+ public boolean waitForTsFileClose() throws InterruptedException {
if (!isClosed.get()) {
synchronized (isClosed) {
while (!isClosed.get()) {
@@ -111,6 +117,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
implements TsFileIns
}
}
}
+ return isTsFileFormatValid;
}
public File getTsFile() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 1363128e181..1c01ced6695 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -161,6 +161,12 @@ public class TsFileProcessor {
private static final String FLUSH_QUERY_WRITE_RELEASE =
"{}: {} get flushQueryLock write lock released";
+ /**
+ * Whether this file keeps TsFile format. If the file violates TsFile
format, then it shouldn't be
+ * captured by pipe engine.
+ */
+ private boolean isTsFileFormatValidForPipe = true;
+
/** close file listener. */
private final List<CloseFileListener> closeFileListeners = new
CopyOnWriteArrayList<>();
@@ -1384,6 +1390,7 @@ public class TsFileProcessor {
// remove this processor from Closing list in DataRegion,
// mark the TsFileResource closed, no need writer anymore
writer.close();
+ isTsFileFormatValidForPipe = false; // empty file, no need to be captured
by pipe
for (CloseFileListener closeFileListener : closeFileListeners) {
closeFileListener.onClosed(this);
}
@@ -1399,6 +1406,11 @@ public class TsFileProcessor {
writer = null;
}
+ /** Only useful after TsFile is closed. */
+ public boolean isTsFileFormatValidForPipe() {
+ return isTsFileFormatValidForPipe;
+ }
+
public boolean isManagedByFlushManager() {
return managedByFlushManager;
}
