Make compressed and encrypted nullable (#1206)

Motivation:

The old replica could fail to replay logs if `encrypted` or `compressed`
fields are included in JSON data since
`@JsonIgnoreProperties(ignoreUnknown = true)` isn't applied to `LogMeta`

Modifications:

- Make `encrypted` and `compressed` in `LogMeta` nullable and exclude
from serialization if them are null.

Result:

`LogMeta` is now backward compatible.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Refactor**
* Log metadata now uses nullable flags for compression/encryption and
omits null fields from JSON; timestamp handling tolerates missing values
and null flags no longer imply true/false implicitly.

* **Tests**
* Added tests ensuring null fields are excluded from log metadata JSON
serialization and that serialization/deserialization remain compatible.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: ikhoon

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

@@ -16,8 +16,6 @@
 
 package com.linecorp.centraldogma.server.internal.replication;
 
-import static com.google.common.base.MoreObjects.firstNonNull;
-
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -27,22 +25,26 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
 
+@JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonIgnoreProperties(ignoreUnknown = true)
 class LogMeta {
 
     private final int replicaId;
     private final long timestamp;
     private final int size;
-    private final boolean compressed;
-    private final boolean encrypted;
+    @Nullable
+    private final Boolean compressed;
+    @Nullable
+    private final Boolean encrypted;
     private final List<Long> blocks;
 
     @JsonCreator
     LogMeta(@JsonProperty(value = "replicaId", required = true) int replicaId,
-            @JsonProperty(value = "timestamp", defaultValue = "0") Long timestamp,
+            @JsonProperty(value = "timestamp", defaultValue = "0") @Nullable Long timestamp,
             @JsonProperty("size") int size,
             @JsonProperty("blocks") List<Long> blocks,
             @Nullable @JsonProperty("compressed") Boolean compressed,
@@ -54,13 +56,13 @@ class LogMeta {
         }
         this.timestamp = timestamp;
         this.size = size;
-        // Defaults to false for backward compatibility.
-        this.compressed = firstNonNull(compressed, false);
-        this.encrypted = firstNonNull(encrypted, false);
+        this.compressed = compressed;
+        this.encrypted = encrypted;
         this.blocks = blocks;
     }
 
-    LogMeta(int replicaId, Long timestamp, int size, boolean compressed, boolean encrypted) {
+    LogMeta(int replicaId, Long timestamp, int size,
+            @Nullable Boolean compressed, @Nullable Boolean encrypted) {
         this(replicaId, timestamp, size, new ArrayList<>(4), compressed, encrypted);
     }
 
@@ -79,13 +81,15 @@ int size() {
         return size;
     }
 
+    @Nullable
     @JsonProperty("compressed")
-    boolean compressed() {
+    Boolean compressed() {
         return compressed;
     }
 
+    @Nullable
     @JsonProperty("encrypted")
-    boolean encrypted() {
+    Boolean encrypted() {
         return encrypted;
     }
 
@@ -107,8 +111,8 @@ public boolean equals(Object o) {
         return replicaId == logMeta.replicaId &&
                timestamp == logMeta.timestamp &&
                size == logMeta.size &&
-               compressed == logMeta.compressed &&
-               encrypted == logMeta.encrypted &&
+               Objects.equals(compressed, logMeta.compressed) &&
+               Objects.equals(encrypted, logMeta.encrypted) &&
                Objects.equals(blocks, logMeta.blocks);
     }
 
@@ -120,6 +124,7 @@ public int hashCode() {
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)
+                          .omitNullValues()
                           .add("replicaId", replicaId)
                           .add("timestamp", timestamp)
                           .add("size", size)

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

@@ -1026,7 +1026,7 @@ private long storeLog(ReplicationLog<?> log) {
 
             // TODO(ikhoon): Enable compression once releasing decompression support for forward compatibility.
             final LogMeta logMeta = new LogMeta(log.replicaId(), System.currentTimeMillis(), bytes.length,
-                                                false, false);
+                                                null, null);
 
             final int count = (bytes.length + MAX_BYTES - 1) / MAX_BYTES;
             for (int i = 0; i < count; ++i) {
@@ -1075,7 +1075,8 @@ Optional<ReplicationLog<?>> loadLog(long revision, boolean skipIfSameReplica) {
             }
             assert logMeta.size() == offset;
 
-            if (logMeta.compressed()) {
+            final Boolean compressed = logMeta.compressed();
+            if (compressed != null && compressed) {
                 bytes = Zstd.decompress(bytes);
             }
             final ReplicationLog<?> log = Jackson.readValue(bytes, ReplicationLog.class);

server/src/test/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutorTest.java

@@ -16,6 +16,7 @@
 package com.linecorp.centraldogma.server.internal.replication;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
+import static net.javacrumbs.jsonunit.fluent.JsonFluentAssert.assertThatJson;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.catchThrowable;
@@ -602,6 +603,19 @@ void testLogMetaSerde() throws JsonProcessingException {
         assertThat(deserialized).isEqualTo(logMeta);
     }
 
+    @Test
+    void testLogMetaSerde_excludeNull() throws JsonProcessingException {
+        final LogMeta logMeta = new LogMeta(1, 1L, 10, null, null);
+        logMeta.appendBlock(1);
+        logMeta.appendBlock(2);
+        logMeta.appendBlock(3);
+        assertThat(logMeta.blocks()).containsExactly(1L, 2L, 3L);
+        final String jsonStr = Jackson.writeValueAsString(logMeta);
+        assertThatJson(jsonStr).isEqualTo("{\"replicaId\":1,\"timestamp\":1,\"size\":10,\"blocks\":[1,2,3]}");
+        final LogMeta deserialized = Jackson.readValue(jsonStr, LogMeta.class);
+        assertThat(deserialized).isEqualTo(logMeta);
+    }
+
     private static <T> void awaitUntilReplicated(Cluster cluster, Command<T> command) {
         for (int i = 0; i < cluster.size(); i++) {
             final Replica replica = cluster.get(i);