linkedin · beijxu · Jul 14, 2026
diff --git a/ambry-api/src/main/java/com/github/ambry/account/MigrationConfig.java b/ambry-api/src/main/java/com/github/ambry/account/MigrationConfig.java
@@ -40,6 +40,10 @@ public class MigrationConfig {
   @JsonProperty(LIST_RAMP_KEY)
   private ListRamp listRamp;
 
+  public static final String DELETE_RAMP_KEY = "deleteRamp";
+  @JsonProperty(DELETE_RAMP_KEY)
+  private DeleteRamp deleteRamp;
+
   // Write ramp config.
   @JsonIgnoreProperties(ignoreUnknown = true)
   public static class WriteRamp {
@@ -289,21 +293,109 @@ public int hashCode() {
     }
   }
 
+  // Delete ramp config.
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public static class DeleteRamp {
+    @JsonProperty("forceDisableDualDelete")
+    private boolean forceDisableDualDelete;
+
+    @JsonProperty("dualDeleteAsyncPct")
+    private double dualDeleteAsyncPct;
+
+    @JsonProperty("dualDeleteSyncPctNonStrict")
+    private double dualDeleteSyncPctNonStrict;
+
+    @JsonProperty("dualDeleteSyncPctStrict")
+    private double dualDeleteSyncPctStrict;
+
+    @JsonProperty("deleteOnlyToSecondary")
+    private boolean deleteOnlyToSecondary;
+
+    @JsonCreator
+    public DeleteRamp(
+        @JsonProperty("forceDisableDualDelete") boolean forceDisableDualDelete,
+        @JsonProperty("dualDeleteAsyncPct") double dualDeleteAsyncPct,
+        @JsonProperty("dualDeleteSyncPctNonStrict") double dualDeleteSyncPctNonStrict,
+        @JsonProperty("dualDeleteSyncPctStrict") double dualDeleteSyncPctStrict,
+        @JsonProperty("deleteOnlyToSecondary") boolean deleteOnlyToSecondary) {
+      this.forceDisableDualDelete = forceDisableDualDelete;
+      this.dualDeleteAsyncPct = dualDeleteAsyncPct;
+      this.dualDeleteSyncPctNonStrict = dualDeleteSyncPctNonStrict;
+      this.dualDeleteSyncPctStrict = dualDeleteSyncPctStrict;
+      this.deleteOnlyToSecondary = deleteOnlyToSecondary;
+    }
+
+    // Default constructor for DeleteRamp
+    public DeleteRamp() {
+      this.forceDisableDualDelete = false;
+      this.dualDeleteAsyncPct = 0.0;
+      this.dualDeleteSyncPctNonStrict = 0.0;
+      this.dualDeleteSyncPctStrict = 0.0;
+      this.deleteOnlyToSecondary = false;
+    }
+
+    // Getters
+    public boolean isForceDisableDualDelete() {
+      return forceDisableDualDelete;
+    }
+
+    public double getDualDeleteAsyncPct() {
+      return dualDeleteAsyncPct;
+    }
+
+    public double getDualDeleteSyncPctNonStrict() {
+      return dualDeleteSyncPctNonStrict;
+    }
+
+    public double getDualDeleteSyncPctStrict() {
+      return dualDeleteSyncPctStrict;
+    }
+
+    public boolean isDeleteOnlyToSecondary() {
+      return deleteOnlyToSecondary;
+    }
+
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      DeleteRamp that = (DeleteRamp) o;
+      return forceDisableDualDelete == that.forceDisableDualDelete &&
+          Double.compare(that.dualDeleteAsyncPct, dualDeleteAsyncPct) == 0 &&
+          Double.compare(that.dualDeleteSyncPctNonStrict, dualDeleteSyncPctNonStrict) == 0 &&
+          Double.compare(that.dualDeleteSyncPctStrict, dualDeleteSyncPctStrict) == 0 &&
+          deleteOnlyToSecondary == that.deleteOnlyToSecondary;
+    }
+
+    @Override
+    public int hashCode() {
+      return java.util.Objects.hash(forceDisableDualDelete, dualDeleteAsyncPct,
+          dualDeleteSyncPctNonStrict, dualDeleteSyncPctStrict, deleteOnlyToSecondary);
+    }
+  }
+
   // Default migration config.
   public MigrationConfig() {
-    this(false, new WriteRamp(), new ReadRamp(), new ListRamp());
+    this(false, new WriteRamp(), new ReadRamp(), new ListRamp(), null);
+  }
+
+  // Backward-compatible constructor without deleteRamp.
+  public MigrationConfig(boolean overrideAccountMigrationConfig, WriteRamp writeRamp, ReadRamp readRamp,
+      ListRamp listRamp) {
+    this(overrideAccountMigrationConfig, writeRamp, readRamp, listRamp, null);
   }
 
   @JsonCreator
   public MigrationConfig(
       @JsonProperty(OVERRIDE_ACCOUNT_MIGRATION_CONFIG) boolean overrideAccountMigrationConfig,
       @JsonProperty(WRITE_RAMP_KEY) WriteRamp writeRamp,
       @JsonProperty(READ_RAMP_KEY) ReadRamp readRamp,
-      @JsonProperty(LIST_RAMP_KEY) ListRamp listRamp) {
+      @JsonProperty(LIST_RAMP_KEY) ListRamp listRamp,
+      @JsonProperty(DELETE_RAMP_KEY) DeleteRamp deleteRamp) {
     this.overrideAccountMigrationConfig = overrideAccountMigrationConfig;
     this.writeRamp = writeRamp;
     this.readRamp = readRamp;
     this.listRamp = listRamp;
+    this.deleteRamp = deleteRamp;
   }
 
   public boolean isOverrideAccountMigrationConfig() {
@@ -322,6 +414,10 @@ public ListRamp getListRamp() {
     return listRamp;
   }
 
+  public DeleteRamp getDeleteRamp() {
+    return deleteRamp;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -330,11 +426,12 @@ public boolean equals(Object o) {
     return overrideAccountMigrationConfig == that.overrideAccountMigrationConfig &&
         java.util.Objects.equals(writeRamp, that.writeRamp) &&
         java.util.Objects.equals(readRamp, that.readRamp) &&
-        java.util.Objects.equals(listRamp, that.listRamp);
+        java.util.Objects.equals(listRamp, that.listRamp) &&
+        java.util.Objects.equals(deleteRamp, that.deleteRamp);
   }
 
   @Override
   public int hashCode() {
-    return java.util.Objects.hash(overrideAccountMigrationConfig, writeRamp, readRamp, listRamp);
+    return java.util.Objects.hash(overrideAccountMigrationConfig, writeRamp, readRamp, listRamp, deleteRamp);
   }
 }
diff --git a/ambry-api/src/test/java/com/github/ambry/account/BackwardsCompatibilityTest.java b/ambry-api/src/test/java/com/github/ambry/account/BackwardsCompatibilityTest.java
@@ -159,7 +159,7 @@ public class BackwardsCompatibilityTest {
   // When you add a new field to a class, add it here too. This is the change-detector.
 
   private static final Set<String> EXPECTED_MIGRATION_CONFIG_FIELDS = Set.of(
-      "overrideAccountMigrationConfig", "writeRamp", "readRamp", "listRamp"
+      "overrideAccountMigrationConfig", "writeRamp", "readRamp", "listRamp", "deleteRamp"
   );
 
   private static final Set<String> EXPECTED_WRITE_RAMP_FIELDS = Set.of(
@@ -179,6 +179,12 @@ public class BackwardsCompatibilityTest {
       "disableFallbackToPrimary"
   );
 
+  private static final Set<String> EXPECTED_DELETE_RAMP_FIELDS = Set.of(
+      "forceDisableDualDelete", "dualDeleteAsyncPct",
+      "dualDeleteSyncPctNonStrict", "dualDeleteSyncPctStrict",
+      "deleteOnlyToSecondary"
+  );
+
   // Note: "secondaryEnabled" is a computed getter (Account#isSecondaryEnabled()) that Jackson serializes
   // as a top-level property, even though it is derived from rampControl and not independently stored.
   // It must be in this set because it appears in serialized output, but it is NOT part of the stored
@@ -245,6 +251,19 @@ public void testMigrationConfigV1SnapshotDeserialization() throws Exception {
     assertEquals(0.0, config.getReadRamp().getDualHeadSyncPct(), 0.001);
   }
 
+  /**
+   * Verify that MigrationConfig snapshots without the deleteRamp field (all existing production data)
+   * still deserialize correctly. The missing deleteRamp field must default to null.
+   */
+  @Test
+  public void testMigrationConfigSnapshotWithoutDeleteRamp() throws Exception {
+    MigrationConfig config = objectMapper.readValue(MIGRATION_CONFIG_SNAPSHOT_JSON, MigrationConfig.class);
+    assertNull("deleteRamp should be null when absent from JSON", config.getDeleteRamp());
+
+    config = objectMapper.readValue(MIGRATION_CONFIG_V1_SNAPSHOT_JSON, MigrationConfig.class);
+    assertNull("deleteRamp should be null when absent from V1 JSON", config.getDeleteRamp());
+  }
+
   /**
    * FIELD SET REGRESSION: Verify that serialized MigrationConfig contains exactly the expected fields.
    * If a new field is added to any ramp class, this test will fail, alerting the developer to:
@@ -257,7 +276,8 @@ public void testMigrationConfigFieldSetRegression() throws Exception {
     MigrationConfig config = new MigrationConfig(true,
         new MigrationConfig.WriteRamp(false, 50.0, 30.0, 20.0, true),
         new MigrationConfig.ReadRamp(false, 40.0, 60.0, 10.0, 5.0, true, 25.0),
-        new MigrationConfig.ListRamp(false, 70.0, 80.0, true));
+        new MigrationConfig.ListRamp(false, 70.0, 80.0, true),
+        new MigrationConfig.DeleteRamp(false, 40.0, 20.0, 10.0, true));
 
     String json = objectMapper.writeValueAsString(config);
     JsonNode root = objectMapper.readTree(json);
@@ -278,6 +298,10 @@ public void testMigrationConfigFieldSetRegression() throws Exception {
     Set<String> actualListRampFields = fieldNames(root.get("listRamp"));
     assertEquals("ListRamp fields changed. Update EXPECTED_LIST_RAMP_FIELDS and add a backwards compat test.",
         EXPECTED_LIST_RAMP_FIELDS, actualListRampFields);
+
+    Set<String> actualDeleteRampFields = fieldNames(root.get("deleteRamp"));
+    assertEquals("DeleteRamp fields changed. Update EXPECTED_DELETE_RAMP_FIELDS and add a backwards compat test.",
+        EXPECTED_DELETE_RAMP_FIELDS, actualDeleteRampFields);
   }
 
   // ==================== Account Tests ====================

diff --git a/ambry-api/src/test/java/com/github/ambry/account/MigrationConfigTest.java b/ambry-api/src/test/java/com/github/ambry/account/MigrationConfigTest.java
@@ -27,23 +27,33 @@ public void testConstructorAndGetters() {
     MigrationConfig.WriteRamp writeRamp = new MigrationConfig.WriteRamp(false, 50.0, 30.0, 20.0, true);
     MigrationConfig.ReadRamp readRamp = new MigrationConfig.ReadRamp(false, 40.0, 60.0, 10.0, 5.0, true, 25.0);
     MigrationConfig.ListRamp listRamp = new MigrationConfig.ListRamp(false, 70.0, 80.0, true);
+    MigrationConfig.DeleteRamp deleteRamp = new MigrationConfig.DeleteRamp(false, 40.0, 20.0, 10.0, true);
 
-    MigrationConfig migrationConfig = new MigrationConfig(true, writeRamp, readRamp, listRamp);
+    MigrationConfig migrationConfig = new MigrationConfig(true, writeRamp, readRamp, listRamp, deleteRamp);
 
     assertTrue(migrationConfig.isOverrideAccountMigrationConfig());
     assertEquals(writeRamp, migrationConfig.getWriteRamp());
     assertEquals(readRamp, migrationConfig.getReadRamp());
     assertEquals(listRamp, migrationConfig.getListRamp());
+    assertEquals(deleteRamp, migrationConfig.getDeleteRamp());
     assertEquals(25.0, readRamp.getDualHeadSyncPct(), 0.001);
+
+    // Verify DeleteRamp getters
+    assertFalse(deleteRamp.isForceDisableDualDelete());
+    assertEquals(40.0, deleteRamp.getDualDeleteAsyncPct(), 0.001);
+    assertEquals(20.0, deleteRamp.getDualDeleteSyncPctNonStrict(), 0.001);
+    assertEquals(10.0, deleteRamp.getDualDeleteSyncPctStrict(), 0.001);
+    assertTrue(deleteRamp.isDeleteOnlyToSecondary());
   }
 
   @Test
   public void testJsonSerializationAndDeserialization() throws Exception {
     MigrationConfig.WriteRamp writeRamp = new MigrationConfig.WriteRamp(false, 50.0, 30.0, 20.0, true);
     MigrationConfig.ReadRamp readRamp = new MigrationConfig.ReadRamp(false, 40.0, 60.0, 10.0, 5.0, true, 15.0);
     MigrationConfig.ListRamp listRamp = new MigrationConfig.ListRamp(false, 70.0, 80.0, true);
+    MigrationConfig.DeleteRamp deleteRamp = new MigrationConfig.DeleteRamp(false, 40.0, 20.0, 10.0, true);
 
-    MigrationConfig originalConfig = new MigrationConfig(true, writeRamp, readRamp, listRamp);
+    MigrationConfig originalConfig = new MigrationConfig(true, writeRamp, readRamp, listRamp, deleteRamp);
 
     // Serialize to JSON
     String json = objectMapper.writeValueAsString(originalConfig);
@@ -55,16 +65,18 @@ public void testJsonSerializationAndDeserialization() throws Exception {
     assertEquals(originalConfig.getWriteRamp(), deserializedConfig.getWriteRamp());
     assertEquals(originalConfig.getReadRamp(), deserializedConfig.getReadRamp());
     assertEquals(originalConfig.getListRamp(), deserializedConfig.getListRamp());
+    assertEquals(originalConfig.getDeleteRamp(), deserializedConfig.getDeleteRamp());
   }
 
   @Test
   public void testDefaultValues() {
-    MigrationConfig migrationConfig = new MigrationConfig(false, null, null, null);
+    MigrationConfig migrationConfig = new MigrationConfig(false, null, null, null, null);
 
     assertFalse(migrationConfig.isOverrideAccountMigrationConfig());
     assertNull(migrationConfig.getWriteRamp());
     assertNull(migrationConfig.getReadRamp());
     assertNull(migrationConfig.getListRamp());
+    assertNull(migrationConfig.getDeleteRamp());
   }
 
   @Test
@@ -104,6 +116,13 @@ public void testDeserializationIgnoresUnknownFields() throws Exception {
         + "\"serveListFromSecondaryPct\":0.0,\"disableFallbackToPrimary\":false,\"someNewListField\":\"hello\"}";
     MigrationConfig.ListRamp listRamp = objectMapper.readValue(listRampJson, MigrationConfig.ListRamp.class);
     assertEquals(30.0, listRamp.getShadowListPct(), 0.001);
+
+    // DeleteRamp JSON with an unknown field "someNewDeleteField"
+    String deleteRampJson = "{\"forceDisableDualDelete\":false,\"dualDeleteAsyncPct\":45.0,"
+        + "\"dualDeleteSyncPctNonStrict\":0.0,\"dualDeleteSyncPctStrict\":0.0,"
+        + "\"deleteOnlyToSecondary\":false,\"someNewDeleteField\":true}";
+    MigrationConfig.DeleteRamp deleteRamp = objectMapper.readValue(deleteRampJson, MigrationConfig.DeleteRamp.class);
+    assertEquals(45.0, deleteRamp.getDualDeleteAsyncPct(), 0.001);
   }
 
   /**
@@ -156,4 +175,59 @@ public void testDeserializationWithoutDualHeadSyncPct() throws Exception {
     assertEquals(0.0, deserialized.getReadRamp().getDualHeadSyncPct(), 0.001);
     assertEquals(10.0, deserialized.getReadRamp().getShadowReadMetadataPct(), 0.001);
   }
+
+  @Test
+  public void testDeserializationWithoutDeleteRamp() throws Exception {
+    String json = "{\"overrideAccountMigrationConfig\":false,"
+        + "\"writeRamp\":{\"forceDisableDualWriteAndDelete\":false,\"dualWriteAndDeleteAsyncPct\":50.0,"
+        + "\"dualWriteAndDeleteSyncPctNonStrict\":0.0,\"dualWriteAndDeleteSyncPctStrict\":0.0,"
+        + "\"writeAndDeleteOnlyToSecondary\":false},"
+        + "\"readRamp\":{\"forceDisableReadFromSecondary\":false,\"shadowReadMetadataPct\":10.0,"
+        + "\"shadowReadMd5Pct\":0.0,\"shadowReadContentPct\":0.0,\"serveReadFromSecondaryPct\":0.0,"
+        + "\"disableFallbackToPrimary\":false,\"dualHeadSyncPct\":0.0},"
+        + "\"listRamp\":{\"forceDisableListFromSecondary\":false,\"shadowListPct\":0.0,"
+        + "\"serveListFromSecondaryPct\":0.0,\"disableFallbackToPrimary\":false}}";
+
+    MigrationConfig deserialized = objectMapper.readValue(json, MigrationConfig.class);
+    assertNull("deleteRamp should be null when absent from JSON", deserialized.getDeleteRamp());
+    assertNotNull(deserialized.getWriteRamp());
+    assertNotNull(deserialized.getReadRamp());
+    assertNotNull(deserialized.getListRamp());
+  }
+
+  @Test
+  public void testFourArgConstructorBackwardCompat() {
+    MigrationConfig.WriteRamp writeRamp = new MigrationConfig.WriteRamp();
+    MigrationConfig.ReadRamp readRamp = new MigrationConfig.ReadRamp();
+    MigrationConfig.ListRamp listRamp = new MigrationConfig.ListRamp();
+
+    MigrationConfig config = new MigrationConfig(false, writeRamp, readRamp, listRamp);
+
+    assertNull("deleteRamp should be null when using 4-arg constructor", config.getDeleteRamp());
+    assertEquals(writeRamp, config.getWriteRamp());
+    assertEquals(readRamp, config.getReadRamp());
+    assertEquals(listRamp, config.getListRamp());
+  }
+
+  @Test
+  public void testDeleteRampDefaultConstructor() {
+    MigrationConfig.DeleteRamp deleteRamp = new MigrationConfig.DeleteRamp();
+
+    assertFalse(deleteRamp.isForceDisableDualDelete());
+    assertEquals(0.0, deleteRamp.getDualDeleteAsyncPct(), 0.001);
+    assertEquals(0.0, deleteRamp.getDualDeleteSyncPctNonStrict(), 0.001);
+    assertEquals(0.0, deleteRamp.getDualDeleteSyncPctStrict(), 0.001);
+    assertFalse(deleteRamp.isDeleteOnlyToSecondary());
+  }
+
+  @Test
+  public void testDeleteRampEqualsAndHashCode() {
+    MigrationConfig.DeleteRamp ramp1 = new MigrationConfig.DeleteRamp(true, 50.0, 30.0, 20.0, false);
+    MigrationConfig.DeleteRamp ramp2 = new MigrationConfig.DeleteRamp(true, 50.0, 30.0, 20.0, false);
+    MigrationConfig.DeleteRamp ramp3 = new MigrationConfig.DeleteRamp(false, 50.0, 30.0, 20.0, false);
+
+    assertEquals(ramp1, ramp2);
+    assertEquals(ramp1.hashCode(), ramp2.hashCode());
+    assertNotEquals(ramp1, ramp3);
+  }
 }