Compress ZooKeeper replication logs

Motivation:

When a file is newly created or has a large diff, the replication log
size can grow significantly. With compression, up to about 7–8 MB of
data can be stored in a single znode, improving replication performance.

Modifcations:

- Add `zstd-jni` dependency to the server module.
- Compress replication logs using zstd before storing them in ZooKeeper.
- Decompress replication logs when compressed prior to replay.

Result:

Central Dogma server now compresses replication logs with zstd.

Author: ikhoon

dependencies.toml

@@ -83,6 +83,7 @@ thrift09 = { strictly = "0.9.3-1" }
 # See: https://github.com/apache/curator/blob/master/pom.xml
 #      (Switch to the right tag to find out the right version.)
 zookeeper = "3.9.3"
+zstd = "1.5.7-4"
 
 [boms]
 armeria = { module = "com.linecorp.armeria:armeria-bom", version.ref = "armeria" }
@@ -465,6 +466,10 @@ exclusions = [
     "org.apache.yetus:audience-annotations",
     "org.slf4j:slf4j-log4j12"]
 
+[libraries.zstd]
+module = "com.github.luben:zstd-jni"
+version.ref = "zstd"
+
 [plugins]
 download = { id = "de.undercouch.download", version.ref = "download" }
 jmh = { id = "me.champeau.jmh", version.ref = "jmh-gradle-plugin" }

server/build.gradle

@@ -47,6 +47,8 @@ dependencies {
     // Snappy
     implementation libs.snappy
 
+    implementation libs.zstd
+
     // Logging
     optionalImplementation libs.logback15
     // For Caffiene. See https://github.com/line/centraldogma/pull/499

server/src/main/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutor.java

@@ -16,6 +16,7 @@
 
 package com.linecorp.centraldogma.server.internal.replication;
 
+import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.util.Objects.requireNonNull;
 
@@ -72,6 +73,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.github.luben.zstd.Zstd;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.base.MoreObjects;
@@ -1029,18 +1031,21 @@ private static class LogMeta {
         private final int replicaId;
         private final long timestamp;
         private final int size;
+        private final boolean compressed;
         private final List<Long> blocks = new ArrayList<>();
 
         @JsonCreator
         LogMeta(@JsonProperty(value = "replicaId", required = true) int replicaId,
                 @JsonProperty(value = "timestamp", defaultValue = "0") Long timestamp,
-                @JsonProperty("size") int size) {
+                @JsonProperty("size") int size, @JsonProperty("compressed") Boolean compressed) {
             this.replicaId = replicaId;
             if (timestamp == null) {
                 timestamp = 0L;
             }
             this.timestamp = timestamp;
             this.size = size;
+            // Defaults to null for backward compatibility.
+            this.compressed = firstNonNull(compressed, false);
         }
 
         @JsonProperty
@@ -1058,6 +1063,11 @@ int size() {
             return size;
         }
 
+        @JsonProperty("compressed")
+        boolean compressed() {
+            return compressed;
+        }
+
         @JsonProperty
         List<Long> blocks() {
             return Collections.unmodifiableList(blocks);
@@ -1073,17 +1083,20 @@ public String toString() {
                               .add("replicaId", replicaId)
                               .add("timestamp", timestamp)
                               .add("size", size)
+                              .add("compressed", compressed)
                               .add("blocks", blocks)
                               .toString();
         }
     }
 
     private long storeLog(ReplicationLog<?> log) {
         try {
-            final byte[] bytes = Jackson.writeValueAsBytes(log);
+            byte[] bytes = Jackson.writeValueAsBytes(log);
             assert bytes.length > 0;
+            bytes = Zstd.compress(bytes);
 
-            final LogMeta logMeta = new LogMeta(log.replicaId(), System.currentTimeMillis(), bytes.length);
+            final LogMeta logMeta = new LogMeta(log.replicaId(), System.currentTimeMillis(), bytes.length,
+                                                true);
 
             final int count = (bytes.length + MAX_BYTES - 1) / MAX_BYTES;
             for (int i = 0; i < count; ++i) {
@@ -1122,7 +1135,7 @@ Optional<ReplicationLog<?>> loadLog(long revision, boolean skipIfSameReplica) {
                 return Optional.empty();
             }
 
-            final byte[] bytes = new byte[logMeta.size()];
+            byte[] bytes = new byte[logMeta.size()];
             int offset = 0;
             for (long blockId : logMeta.blocks()) {
                 final String blockPath = absolutePath(LOG_BLOCK_PATH) + '/' + pathFromRevision(blockId);
@@ -1132,6 +1145,9 @@ Optional<ReplicationLog<?>> loadLog(long revision, boolean skipIfSameReplica) {
             }
             assert logMeta.size() == offset;
 
+            if (logMeta.compressed()) {
+                bytes = Zstd.decompress(bytes);
+            }
             final ReplicationLog<?> log = Jackson.readValue(bytes, ReplicationLog.class);
             return Optional.of(log);
         } catch (Exception e) {