[region migration] Fix tsfile error caused by target DataNode's fast …

…rebooting #14031
apache · Nov 9, 2024 · 4be1020 · 4be1020
1 parent 8a92aaf
commit 4be1020
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 9 deletions.
diff --git a/...b-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/...b-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
@@ -373,7 +373,7 @@ public void transmitSnapshot(Peer targetPeer) throws ConsensusGroupModifyPeerExc
   }
 
   public void receiveSnapshotFragment(
-      String snapshotId, String originalFilePath, ByteBuffer fileChunk)
+      String snapshotId, String originalFilePath, ByteBuffer fileChunk, long fileOffset)
       throws ConsensusGroupModifyPeerException {
     try {
       String targetFilePath = calculateSnapshotPath(snapshotId, originalFilePath);
@@ -384,7 +384,7 @@ public void receiveSnapshotFragment(
       }
       try (FileOutputStream fos = new FileOutputStream(targetFile.getAbsolutePath(), true);
           FileChannel channel = fos.getChannel()) {
-        channel.write(fileChunk.slice());
+        channel.write(fileChunk.slice(), fileOffset);
       }
     } catch (IOException e) {
       throw new ConsensusGroupModifyPeerException(

diff --git a/...src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java b/...src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
@@ -270,7 +270,7 @@ public TSendSnapshotFragmentRes sendSnapshotFragment(TSendSnapshotFragmentReq re
     }
     TSStatus responseStatus;
     try {
-      impl.receiveSnapshotFragment(req.snapshotId, req.filePath, req.fileChunk);
+      impl.receiveSnapshotFragment(req.snapshotId, req.filePath, req.fileChunk, req.offset);
       responseStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
     } catch (ConsensusGroupModifyPeerException e) {
       responseStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());

diff --git a/...ore/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/SnapshotFragment.java b/...ore/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/SnapshotFragment.java
@@ -50,6 +50,7 @@ public TSendSnapshotFragmentReq toTSendSnapshotFragmentReq() {
     TSendSnapshotFragmentReq req = new TSendSnapshotFragmentReq();
     req.setSnapshotId(snapshotId);
     req.setFilePath(filePath);
+    req.setOffset(startOffset);
     req.setChunkLength(fragmentSize);
     req.setFileChunk(fileChunk);
     return req;

diff --git a/...nsensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/SnapshotFragmentReader.java b/...nsensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/SnapshotFragmentReader.java
@@ -46,12 +46,12 @@ public SnapshotFragmentReader(String snapshotId, Path path) throws IOException {
 
   public boolean hasNext() throws IOException {
     buf.clear();
-    int actualReadSize = fileChannel.read(buf);
+    int readSize = fileChannel.read(buf);
     buf.flip();
-    if (actualReadSize > 0) {
+    if (readSize > 0) {
       cachedSnapshotFragment =
-          new SnapshotFragment(snapshotId, filePath, fileSize, totalReadSize, actualReadSize, buf);
-      totalReadSize += actualReadSize;
+          new SnapshotFragment(snapshotId, filePath, fileSize, totalReadSize, readSize, buf);
+      totalReadSize += readSize;
       return true;
     }
     return false;

diff --git a/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift b/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift
@@ -80,8 +80,9 @@ struct TSendSnapshotFragmentReq {
   1: required common.TConsensusGroupId consensusGroupId
   2: required string snapshotId
   3: required string filePath
-  4: required i64 chunkLength
-  5: required binary fileChunk
+  4: required i64 offset
+  5: required i64 chunkLength
+  6: required binary fileChunk
 }
 
 struct TWaitSyncLogCompleteReq {