This is an automated email from the ASF dual-hosted git repository.
tanxinyu 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 db614fb8eda IoTConsensusV2: Fix concurrency bug when client traffic is
extremely high (#15129)
db614fb8eda is described below
commit db614fb8edae04d7bf37e5538b388011f14e2a23
Author: Peng Junzhi <[email protected]>
AuthorDate: Wed Mar 19 13:27:47 2025 +0800
IoTConsensusV2: Fix concurrency bug when client traffic is extremely high
(#15129)
* fix npe
* fix comment
* fix review
---
.../protocol/pipeconsensus/PipeConsensusReceiver.java | 17 +++++++++++------
1 file changed, 11 insertions(+), 6 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 699c0147fd7..9f683e11a1d 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
@@ -1145,13 +1145,13 @@ public class PipeConsensusReceiver {
private final ConsensusPipeName consensusPipeName;
private final int index;
private File localWritingDir;
+ private File writingFile;
+ private RandomAccessFile writingFileWriter;
// whether this buffer is used. this will be updated when first transfer
tsFile piece or
// when transfer seal.
- private boolean isUsed = false;
+ private volatile boolean isUsed = false;
// If isUsed is true, this variable will be set to the TCommitId of
holderEvent
- private TCommitId commitIdOfCorrespondingHolderEvent;
- private File writingFile;
- private RandomAccessFile writingFileWriter;
+ private volatile TCommitId commitIdOfCorrespondingHolderEvent;
public PipeConsensusTsFileWriter(int index, ConsensusPipeName
consensusPipeName) {
this.index = index;
@@ -1247,12 +1247,17 @@ public class PipeConsensusReceiver {
public void returnSelf(ConsensusPipeName consensusPipeName)
throws DiskSpaceInsufficientException, IOException {
- this.isUsed = false;
- this.commitIdOfCorrespondingHolderEvent = null;
// if config multi-disks, tsFileWriter will roll to new writing path.
+ // must roll before set used to false, because the writing file may be
deleted if other event
+ // uses this tsfileWriter.
if (receiveDirs.size() > 1) {
rollToNextWritingPath();
}
+ // must set used to false after set commitIdOfCorrespondingHolderEvent
to null to avoid the
+ // situation that tsfileWriter is used by other event before set
+ // commitIdOfCorrespondingHolderEvent to null
+ this.commitIdOfCorrespondingHolderEvent = null;
+ this.isUsed = false;
LOGGER.info(
"PipeConsensus-PipeName-{}: tsFileWriter-{} returned self",
consensusPipeName.toString(),
