[ratis] 02/05: RATIS-1765. [GrpcLogAppender] Calculate streaming md5 file-wise when installSnapshot (#803)

[wi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

williamsong pushed a commit to branch snapshot-branch2
in repository https://gitbox.apache.org/repos/asf/ratis.git

commit 765cb40feaccbac89eae2d3cb9c3b5c333f139ed
Author: William Song <48...@users.noreply.github.com>
AuthorDate: Fri Jan 6 21:29:46 2023 +0800

    RATIS-1765. [GrpcLogAppender] Calculate streaming md5 file-wise when installSnapshot (#803)
---
 .../org/apache/ratis/server/storage/FileChunkReader.java    | 13 ++++++++++---
 .../org/apache/ratis/InstallSnapshotFromLeaderTests.java    | 13 ++++++++++++-
 2 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/storage/FileChunkReader.java b/ratis-server/src/main/java/org/apache/ratis/server/storage/FileChunkReader.java
index a5ee66258..47d70a771 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/storage/FileChunkReader.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/storage/FileChunkReader.java
@@ -73,14 +73,21 @@ public class FileChunkReader implements Closeable {
     final long remaining = info.getFileSize() - offset;
     final int chunkLength = remaining < chunkMaxSize ? (int) remaining : chunkMaxSize;
     final ByteString data = ByteString.readFrom(in, chunkLength);
-    final ByteString fileDigest = ByteString.copyFrom(
-            digester != null? digester.digest(): info.getFileDigest().getDigest());
+    // whether this chunk is the last chunk of current file
+    final boolean isDone = offset + chunkLength == info.getFileSize();
+    final ByteString fileDigest;
+    if (digester != null) {
+      // file digest is calculated once in the end and shipped with last FileChunkProto
+      fileDigest = isDone ? ByteString.copyFrom(digester.digest()) : ByteString.EMPTY;
+    } else {
+      fileDigest = ByteString.copyFrom(info.getFileDigest().getDigest());
+    }
 
     final FileChunkProto proto = FileChunkProto.newBuilder()
         .setFilename(relativePath.toString())
         .setOffset(offset)
         .setChunkIndex(chunkIndex)
-        .setDone(offset + chunkLength == info.getFileSize())
+        .setDone(isDone)
         .setData(data)
         .setFileDigest(fileDigest)
         .build();
diff --git a/ratis-server/src/test/java/org/apache/ratis/InstallSnapshotFromLeaderTests.java b/ratis-server/src/test/java/org/apache/ratis/InstallSnapshotFromLeaderTests.java
index c8e04bb00..3d4b49b81 100644
--- a/ratis-server/src/test/java/org/apache/ratis/InstallSnapshotFromLeaderTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/InstallSnapshotFromLeaderTests.java
@@ -37,12 +37,14 @@ import org.apache.ratis.statemachine.impl.FileListSnapshotInfo;
 import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
 import org.apache.ratis.util.FileUtils;
 import org.apache.ratis.util.LifeCycle;
+import org.apache.ratis.util.SizeInBytes;
 import org.junit.Assert;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
@@ -64,6 +66,7 @@ public abstract class InstallSnapshotFromLeaderTests<CLUSTER extends MiniRaftClu
     RaftServerConfigKeys.Snapshot.setAutoTriggerThreshold(
         prop, SNAPSHOT_TRIGGER_THRESHOLD);
     RaftServerConfigKeys.Snapshot.setAutoTriggerEnabled(prop, true);
+    RaftServerConfigKeys.Log.Appender.setSnapshotChunkSizeMax(prop, SizeInBytes.ONE_KB);
   }
 
   private static final int SNAPSHOT_TRIGGER_THRESHOLD = 64;
@@ -114,7 +117,9 @@ public abstract class InstallSnapshotFromLeaderTests<CLUSTER extends MiniRaftClu
       // Check the installed snapshot file number on each Follower matches with the
       // leader snapshot.
       for (RaftServer.Division follower : cluster.getFollowers()) {
-        Assert.assertEquals(3, follower.getStateMachine().getLatestSnapshot().getFiles().size());
+        final SnapshotInfo info = follower.getStateMachine().getLatestSnapshot();
+        Assert.assertNotNull(info);
+        Assert.assertEquals(3, info.getFiles().size());
       }
     } finally {
       cluster.shutdown();
@@ -163,6 +168,12 @@ public abstract class InstallSnapshotFromLeaderTests<CLUSTER extends MiniRaftClu
           FileUtils.createDirectories(file2.getParentFile());
           FileUtils.createNewFile(file1.toPath());
           FileUtils.createNewFile(file2.toPath());
+          // write 4MB data to simulate multiple chunk scene
+          final byte[] data = new byte[4096];
+          Arrays.fill(data, (byte)0x01);
+          try (FileOutputStream fout = new FileOutputStream(file2)) {
+              fout.write(data);
+          }
         }
 
       } catch (IOException ioException) {