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/11/04 04:51:30 UTC
(iotdb) branch pipe-receiver-fsync-file updated (0889c612d91 -> 491448c90ae)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-receiver-fsync-file
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 0889c612d91 Update PipeConsensusReceiver.java
new 207258190ca Update IoTDBFileReceiver.java
new 96383eacfb6 Update PipeConsensusReceiver.java
new 491448c90ae refactor
The 3 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.
Summary of changes:
.../pipeconsensus/PipeConsensusReceiver.java | 16 +++++++++----
.../apache/iotdb/commons/conf/CommonConfig.java | 10 +++++++++
.../iotdb/commons/conf/CommonDescriptor.java | 5 +++++
.../iotdb/commons/pipe/config/PipeConfig.java | 5 +++++
.../commons/pipe/receiver/IoTDBFileReceiver.java | 26 ++++++++++++++++++----
5 files changed, 54 insertions(+), 8 deletions(-)
(iotdb) 01/03: Update IoTDBFileReceiver.java
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-receiver-fsync-file
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 207258190cade2548139923bb933ab09652f52b6
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Nov 4 11:57:41 2024 +0800
Update IoTDBFileReceiver.java
---
.../commons/pipe/receiver/IoTDBFileReceiver.java | 19 +++++++++++++++----
1 file changed, 15 insertions(+), 4 deletions(-)
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index 59bcc06c45c..5b7a8c22105 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -337,7 +337,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
fileName,
writingFile == null ? "null" : writingFile.getPath());
- closeCurrentWritingFileWriter();
+ closeCurrentWritingFileWriter(!isSingleFile);
// If there are multiple files we can not delete the current file
// instead they will be deleted after seal request
if (writingFile != null && isSingleFile) {
@@ -372,10 +372,13 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
return writingFile != null && writingFile.getName().equals(fileName);
}
- private void closeCurrentWritingFileWriter() {
+ private void closeCurrentWritingFileWriter(final boolean fsyncAfterClose) {
if (writingFileWriter != null) {
try {
writingFileWriter.close();
+ if (fsyncAfterClose) {
+ writingFileWriter.getFD().sync();
+ }
LOGGER.info(
"Receiver id = {}: Current writing file writer {} was closed.",
receiverId.get(),
@@ -476,6 +479,10 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// loaded file. Since the writing file writer has already been closed, it will throw a Stream
// Close exception.
writingFileWriter.close();
+ // Sync here is necessary to ensure that the data is written to the disk. Or data region may
+ // load the file before the data is written to the disk and cause unexpected behavior after
+ // system restart. (e.g., empty file in data region's data directory)
+ writingFileWriter.getFD().sync();
writingFileWriter = null;
// writingFile will be deleted after load if no exception occurs
@@ -508,7 +515,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// If the writing file is not sealed successfully, the writing file will be deleted.
// All pieces of the writing file and its mod (if exists) should be retransmitted by the
// sender.
- closeCurrentWritingFileWriter();
+ closeCurrentWritingFileWriter(false);
deleteCurrentWritingFile();
}
}
@@ -552,6 +559,10 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// loaded file. Since the writing file writer has already been closed, it will throw a Stream
// Close exception.
writingFileWriter.close();
+ // Sync here is necessary to ensure that the data is written to the disk. Or data region may
+ // load the file before the data is written to the disk and cause unexpected behavior after
+ // system restart. (e.g., empty file in data region's data directory)
+ writingFileWriter.getFD().sync();
writingFileWriter = null;
// WritingFile will be deleted after load if no exception occurs
@@ -583,7 +594,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// If the writing file is not sealed successfully, the writing file will be deleted.
// All pieces of the writing file and its mod(if exists) should be retransmitted by the
// sender.
- closeCurrentWritingFileWriter();
+ closeCurrentWritingFileWriter(false);
// Clear the directory instead of only deleting the referenced files in seal request
// to avoid previously undeleted file being redundant when transferring multi files
IoTDBReceiverAgent.cleanPipeReceiverDir(receiverFileDirWithIdSuffix.get());
(iotdb) 02/03: Update PipeConsensusReceiver.java
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-receiver-fsync-file
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 96383eacfb663ae6ce858dd04dc70e068e09f308
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Nov 4 12:39:59 2024 +0800
Update PipeConsensusReceiver.java
---
.../protocol/pipeconsensus/PipeConsensusReceiver.java | 16 ++++++++++++----
1 file changed, 12 insertions(+), 4 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
index 0a8b04dbef7..3721c0277a9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
@@ -487,7 +487,7 @@ public class PipeConsensusReceiver {
// If the writing file is not sealed successfully, the writing file will be deleted.
// All pieces of the writing file and its mod (if exists) should be retransmitted by the
// sender.
- closeCurrentWritingFileWriter(tsFileWriter);
+ closeCurrentWritingFileWriter(tsFileWriter, false);
deleteCurrentWritingFile(tsFileWriter);
}
}
@@ -550,6 +550,10 @@ public class PipeConsensusReceiver {
// loaded file. Since the writing file writer has already been closed, it will throw a Stream
// Close exception.
writingFileWriter.close();
+ // Sync here is necessary to ensure that the data is written to the disk. Or data region may
+ // load the file before the data is written to the disk and cause unexpected behavior after
+ // system restart. (e.g., empty file in data region's data directory)
+ writingFileWriter.getFD().sync();
tsFileWriter.setWritingFileWriter(null);
// WritingFile will be deleted after load if no exception occurs
@@ -599,7 +603,7 @@ public class PipeConsensusReceiver {
// If the writing file is not sealed successfully, the writing file will be deleted.
// All pieces of the writing file and its mod(if exists) should be retransmitted by the
// sender.
- closeCurrentWritingFileWriter(tsFileWriter);
+ closeCurrentWritingFileWriter(tsFileWriter, false);
// Clear the directory instead of only deleting the referenced files in seal request
// to avoid previously undeleted file being redundant when transferring multi files
IoTDBReceiverAgent.cleanPipeReceiverDir(receiverFileDirWithIdSuffix.get());
@@ -794,10 +798,14 @@ public class PipeConsensusReceiver {
return !offsetCorrect;
}
- private void closeCurrentWritingFileWriter(PipeConsensusTsFileWriter tsFileWriter) {
+ private void closeCurrentWritingFileWriter(
+ PipeConsensusTsFileWriter tsFileWriter, boolean fsyncAfterClose) {
if (tsFileWriter.getWritingFileWriter() != null) {
try {
tsFileWriter.getWritingFileWriter().close();
+ if (fsyncAfterClose) {
+ tsFileWriter.getWritingFileWriter().getFD().sync();
+ }
LOGGER.info(
"PipeConsensus-PipeName-{}: Current writing file writer {} was closed.",
consensusPipeName,
@@ -878,7 +886,7 @@ public class PipeConsensusReceiver {
fileName,
tsFileWriter.getWritingFile() == null ? "null" : tsFileWriter.getWritingFile().getPath());
- closeCurrentWritingFileWriter(tsFileWriter);
+ closeCurrentWritingFileWriter(tsFileWriter, !isSingleFile);
// If there are multiple files we can not delete the current file
// instead they will be deleted after seal request
if (tsFileWriter.getWritingFile() != null && isSingleFile) {
(iotdb) 03/03: refactor
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-receiver-fsync-file
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 491448c90ae6302421ca1d2c6281bca1d87b50cd
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Nov 4 12:51:22 2024 +0800
refactor
---
.../java/org/apache/iotdb/commons/conf/CommonConfig.java | 10 ++++++++++
.../org/apache/iotdb/commons/conf/CommonDescriptor.java | 5 +++++
.../org/apache/iotdb/commons/pipe/config/PipeConfig.java | 5 +++++
.../iotdb/commons/pipe/receiver/IoTDBFileReceiver.java | 13 ++++++++++---
4 files changed, 30 insertions(+), 3 deletions(-)
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 9f5721d18c3..cf33d9e98f1 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -193,6 +193,8 @@ public class CommonConfig {
private boolean pipeHardLinkWALEnabled = false;
+ private boolean pipeFileReceiverFsyncEnabled = true;
+
private int pipeRealTimeQueuePollHistoryThreshold = 100;
/** The maximum number of threads that can be used to execute subtasks in PipeSubtaskExecutor. */
@@ -674,6 +676,14 @@ public class CommonConfig {
this.pipeHardLinkWALEnabled = pipeHardLinkWALEnabled;
}
+ public boolean getPipeFileReceiverFsyncEnabled() {
+ return pipeFileReceiverFsyncEnabled;
+ }
+
+ public void setPipeFileReceiverFsyncEnabled(boolean pipeFileReceiverFsyncEnabled) {
+ this.pipeFileReceiverFsyncEnabled = pipeFileReceiverFsyncEnabled;
+ }
+
public int getPipeDataStructureTabletRowSize() {
return pipeDataStructureTabletRowSize;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index 872fad33550..c1294b91e02 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -266,6 +266,11 @@ public class CommonDescriptor {
properties.getProperty(
"pipe_hardlink_wal_enabled",
Boolean.toString(config.getPipeHardLinkWALEnabled()))));
+ config.setPipeFileReceiverFsyncEnabled(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "pipe_file_receiver_fsync_enabled",
+ Boolean.toString(config.getPipeFileReceiverFsyncEnabled()))));
config.setPipeDataStructureTabletRowSize(
Integer.parseInt(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 373b21a5b1c..4f1d6da436d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -54,6 +54,10 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeHardLinkWALEnabled();
}
+ public boolean getPipeFileReceiverFsyncEnabled() {
+ return COMMON_CONFIG.getPipeFileReceiverFsyncEnabled();
+ }
+
/////////////////////////////// Tablet ///////////////////////////////
public int getPipeDataStructureTabletRowSize() {
@@ -335,6 +339,7 @@ public class PipeConfig {
LOGGER.info("PipeHardlinkTsFileDirName: {}", getPipeHardlinkTsFileDirName());
LOGGER.info("PipeHardlinkWALDirName: {}", getPipeHardlinkWALDirName());
LOGGER.info("PipeHardLinkWALEnabled: {}", getPipeHardLinkWALEnabled());
+ LOGGER.info("PipeFileReceiverFsyncEnabled: {}", getPipeFileReceiverFsyncEnabled());
LOGGER.info("PipeDataStructureTabletRowSize: {}", getPipeDataStructureTabletRowSize());
LOGGER.info("PipeDataStructureTabletSizeInBytes: {}", getPipeDataStructureTabletSizeInBytes());
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index 5b7a8c22105..7ceac6c9c6e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.commons.pipe.receiver;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.common.PipeTransferHandshakeConstant;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.IoTDBConnectorRequestVersion;
@@ -72,6 +73,8 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
protected String username = CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
protected String password = CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;
+ private static final boolean IS_FSYNC_ENABLED =
+ PipeConfig.getInstance().getPipeFileReceiverFsyncEnabled();
private File writingFile;
private RandomAccessFile writingFileWriter;
@@ -376,7 +379,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
if (writingFileWriter != null) {
try {
writingFileWriter.close();
- if (fsyncAfterClose) {
+ if (IS_FSYNC_ENABLED && fsyncAfterClose) {
writingFileWriter.getFD().sync();
}
LOGGER.info(
@@ -482,7 +485,9 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// Sync here is necessary to ensure that the data is written to the disk. Or data region may
// load the file before the data is written to the disk and cause unexpected behavior after
// system restart. (e.g., empty file in data region's data directory)
- writingFileWriter.getFD().sync();
+ if (IS_FSYNC_ENABLED) {
+ writingFileWriter.getFD().sync();
+ }
writingFileWriter = null;
// writingFile will be deleted after load if no exception occurs
@@ -562,7 +567,9 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
// Sync here is necessary to ensure that the data is written to the disk. Or data region may
// load the file before the data is written to the disk and cause unexpected behavior after
// system restart. (e.g., empty file in data region's data directory)
- writingFileWriter.getFD().sync();
+ if (IS_FSYNC_ENABLED) {
+ writingFileWriter.getFD().sync();
+ }
writingFileWriter = null;
// WritingFile will be deleted after load if no exception occurs