HDDS-9223. Use protobuf for SnapshotDiffJobCodec

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotDiffJob.java

@@ -316,12 +316,17 @@ public Class<SnapshotDiffJob> getTypeClass() {
 
     @Override
     public byte[] toPersistedFormatImpl(SnapshotDiffJob object) throws IOException {
-      return MAPPER.writeValueAsBytes(object);
+      return object.toProtoBuf().toByteArray();
     }
 
     @Override
     public SnapshotDiffJob fromPersistedFormatImpl(byte[] rawData) throws IOException {
-      return MAPPER.readValue(rawData, SnapshotDiffJob.class);
+      try {
+        SnapshotDiffJobProto proto = SnapshotDiffJobProto.parseFrom(rawData);
+        return SnapshotDiffJob.getFromProtoBuf(proto);
+      } catch (IOException e) {
+        return MAPPER.readValue(rawData, SnapshotDiffJob.class);
+      }
     }
 
     @Override

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/OldSnapshotDiffJobCodecForTesting.java

@@ -0,0 +1,37 @@
+package org.apache.hadoop.ozone.om.helpers;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import java.io.IOException;
+
+
+public class OldSnapshotDiffJobCodecForTesting
+    implements Codec<SnapshotDiffJob> {
+
+  private static final ObjectMapper MAPPER = new ObjectMapper()
+      .setSerializationInclusion(JsonInclude.Include.NON_NULL)
+      .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+  @Override
+  public Class<SnapshotDiffJob> getTypeClass() {
+    return SnapshotDiffJob.class;
+  }
+
+  @Override
+  public byte[] toPersistedFormatImpl(SnapshotDiffJob object) throws IOException {
+    return MAPPER.writeValueAsBytes(object);
+  }
+
+  @Override
+  public SnapshotDiffJob fromPersistedFormatImpl(byte[] rawData) throws IOException {
+    return MAPPER.readValue(rawData, SnapshotDiffJob.class);
+  }
+
+  @Override
+  public SnapshotDiffJob copyObject(SnapshotDiffJob object) {
+    // Note: Not really a "copy". from OmDBDiffReportEntryCodec
+    return object;
+  }
+}

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmSnapshotDiffJobCodec.java

@@ -0,0 +1,57 @@
+package org.apache.hadoop.ozone.om.helpers;
+
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.JobStatus;
+import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+/**
+ * Testing serialization of SnapshotDiffJobCodec objects to/from RocksDB.
+ */
+public class TestOmSnapshotDiffJobCodec {
+  private final OldSnapshotDiffJobCodecForTesting oldCodec
+      = new OldSnapshotDiffJobCodecForTesting();
+  private final Codec<SnapshotDiffJob> newCodec = SnapshotDiffJob.getCodec();
+
+  @Test
+  public void testOldJsonSerializedDataCanBeReadByNewCodec() throws Exception {
+    // Step 1: Construct a SnapshotDiffJob instance
+    SnapshotDiffJob original = new SnapshotDiffJob(
+        123456789L,
+        "job-001",
+        JobStatus.IN_PROGRESS,
+        "volA",
+        "buckB",
+        "snap1",
+        "snap2",
+        true,
+        false,
+        100L,
+        SubStatus.SST_FILE_DELTA_DAG_WALK,
+        0.0
+    );
+
+    // Step 2: Serialize using the old Jackson-based codec
+    byte[] oldFormatData = oldCodec.toPersistedFormatImpl(original);
+
+    // Step 3: Deserialize using the new default codec (with Protobuf + JSON fallback)
+    SnapshotDiffJob parsed = newCodec.fromPersistedFormatImpl(oldFormatData);
+
+    // Step 4: Verify critical fields remain consistent after round-trip
+    assertEquals(original.getJobId(), parsed.getJobId());
+    assertEquals(original.getStatus(), parsed.getStatus());
+    assertEquals(original.getVolume(), parsed.getVolume());
+    assertEquals(original.getBucket(), parsed.getBucket());
+    assertEquals(original.getFromSnapshot(), parsed.getFromSnapshot());
+    assertEquals(original.getToSnapshot(), parsed.getToSnapshot());
+    assertEquals(original.isForceFullDiff(), parsed.isForceFullDiff());
+    assertEquals(original.isNativeDiffDisabled(), parsed.isNativeDiffDisabled());
+    assertEquals(original.getSubStatus(), parsed.getSubStatus());
+    assertEquals(original.getTotalDiffEntries(), parsed.getTotalDiffEntries());
+
+    assertEquals(0.0, parsed.getKeysProcessedPct());
+  }
+}