[controller] Add DegradedModeRecoveryService, metrics, and E2E test (#2760)

DegradedModeRecoveryService
2-phase recovery orchestrator: Phase 1 initiates data recovery for all PARTIALLY_ONLINE stores in parallel (prepare → poll readiness → initiate). Phase 2 monitors child DC completion and transitions version status from PARTIALLY_ONLINE → ONLINE.
Orphan detection: Periodic monitor detects PARTIALLY_ONLINE versions with no active recovery (e.g., after controller leader failover) and re-triggers recovery. The recovery flow is idempotent.
Configurable concurrency: Bounded thread pools for recovery and monitoring, configurable via degraded.mode.recovery.thread.pool.size.
Retry with exponential backoff: Failed store recoveries are retried up to 3 times.
Version supersession handling: If a newer version becomes current during recovery polling, the old version is treated as successfully healed.
DegradedModeStats (9 OTel metrics + latency histogram)
recovery.store_success_count / recovery.store_failure_count
recovery.version_transitioned_count
recovery.progress (gauge, 0.0-1.0)
push.auto_converted_count / push.blocked_incremental_count
dc.active_count / dc.duration_minutes
recovery.store_duration_ms (avg/max)

Author: mynameborat

build.gradle

@@ -328,15 +328,7 @@ subprojects {
       // Protocol changes should pin the current version via this override and remove the override in a follow-up PR
       // when actually using the new protocol. Example to pin KME to v12 when introducing v13:
       // project(':internal:venice-common').file('src/main/resources/avro/KafkaMessageEnvelope/v12', PathValidation.DIRECTORY)
-      def versionOverrides = [
-          // AdminOperation v100 stages degradedDatacenters on AddVersion.
-          // Pinned to the current active version until the Java wiring lands in a follow-up PR.
-          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v99', PathValidation.DIRECTORY),
-          // StoreMetaValue v45 stages isDegradedPush on StoreVersion (set by parent when a push is
-          // auto-converted under degraded mode; recovery service uses it as the discriminator).
-          // Pinned to the current active version until the Java wiring lands in a follow-up PR.
-          project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v44', PathValidation.DIRECTORY)
-      ]
+      def versionOverrides = []
 
       def schemaDirs = [sourceDir]
       sourceDir.eachDir { typeDir ->

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -591,6 +591,21 @@ private ConfigKeys() {
    */
   public static final String DEGRADED_MODE_ENABLED = "degraded.mode.enabled";
 
+  /**
+   * Whether auto-recovery is enabled when a degraded DC is unmarked.
+   * When true, the controller will automatically trigger data recovery for stores
+   * with PARTIALLY_ONLINE versions after a DC is unmarked as degraded.
+   */
+  public static final String DEGRADED_MODE_AUTO_RECOVERY_ENABLED = "degraded.mode.auto.recovery.enabled";
+
+  /**
+   * Thread pool size for the degraded mode recovery service.
+   * Controls how many store recoveries can run concurrently. Default is 5, which
+   * supports typical clusters with tens of stores. Increase for clusters with hundreds
+   * of stores to reduce total recovery wall-clock time.
+   */
+  public static final String DEGRADED_MODE_RECOVERY_THREAD_POOL_SIZE = "degraded.mode.recovery.thread.pool.size";
+
   /**
    * Whether stores are allowed to be migrated from/to a specific cluster.
    * The value for this config is read from cluster configs in Zk.

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerRoute.java

@@ -345,7 +345,10 @@ public enum ControllerRoute implements VeniceDimensionInterface {
   ),
   UNMARK_DC_DEGRADED(
       "/unmark_dc_degraded", HttpMethod.POST, Arrays.asList(CLUSTER, ControllerApiConstants.DATACENTER_NAME)
-  ), GET_DEGRADED_DCS("/get_degraded_dcs", HttpMethod.GET, Collections.singletonList(CLUSTER));
+  ), GET_DEGRADED_DCS("/get_degraded_dcs", HttpMethod.GET, Collections.singletonList(CLUSTER)),
+  GET_RECOVERY_PROGRESS(
+      "/get_recovery_progress", HttpMethod.GET, Arrays.asList(CLUSTER, ControllerApiConstants.DATACENTER_NAME)
+  );
 
   private final String path;
   private final HttpMethod httpMethod;

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/RecoveryProgressResponse.java

@@ -0,0 +1,88 @@
+package com.linkedin.venice.controllerapi;
+
+public class RecoveryProgressResponse extends ControllerResponse {
+  /**
+   * Indicates the state of the recovery operation.
+   */
+  public enum RecoveryStatus {
+    /** No recovery has been triggered for this datacenter. */
+    NOT_FOUND,
+    /** Recovery is currently in progress. */
+    IN_PROGRESS,
+    /** Recovery has completed (check failedStores for partial success). */
+    COMPLETED
+  }
+
+  private String datacenterName;
+  private RecoveryStatus status;
+  private int totalStores;
+  private int recoveredStores;
+  private int failedStores;
+  private int versionsTransitioned;
+  private boolean complete;
+  private double progressFraction;
+
+  public String getDatacenterName() {
+    return datacenterName;
+  }
+
+  public void setDatacenterName(String datacenterName) {
+    this.datacenterName = datacenterName;
+  }
+
+  public RecoveryStatus getStatus() {
+    return status;
+  }
+
+  public void setStatus(RecoveryStatus status) {
+    this.status = status;
+  }
+
+  public int getTotalStores() {
+    return totalStores;
+  }
+
+  public void setTotalStores(int totalStores) {
+    this.totalStores = totalStores;
+  }
+
+  public int getRecoveredStores() {
+    return recoveredStores;
+  }
+
+  public void setRecoveredStores(int recoveredStores) {
+    this.recoveredStores = recoveredStores;
+  }
+
+  public int getFailedStores() {
+    return failedStores;
+  }
+
+  public void setFailedStores(int failedStores) {
+    this.failedStores = failedStores;
+  }
+
+  public int getVersionsTransitioned() {
+    return versionsTransitioned;
+  }
+
+  public void setVersionsTransitioned(int versionsTransitioned) {
+    this.versionsTransitioned = versionsTransitioned;
+  }
+
+  public boolean isComplete() {
+    return complete;
+  }
+
+  public void setComplete(boolean complete) {
+    this.complete = complete;
+  }
+
+  public double getProgressFraction() {
+    return progressFraction;
+  }
+
+  public void setProgressFraction(double progressFraction) {
+    this.progressFraction = progressFraction;
+  }
+}

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/UpdateClusterConfigQueryParams.java

@@ -1,5 +1,6 @@
 package com.linkedin.venice.controllerapi;
 
+import static com.linkedin.venice.ConfigKeys.DEGRADED_MODE_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.ALLOW_STORE_MIGRATION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.CHILD_CONTROLLER_ADMIN_TOPIC_CONSUMPTION_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.SERVER_KAFKA_FETCH_QUOTA_RECORDS_PER_SECOND;
@@ -56,6 +57,14 @@ public Optional<Boolean> getChildControllerAdminTopicConsumptionEnabled() {
     return getBoolean(CHILD_CONTROLLER_ADMIN_TOPIC_CONSUMPTION_ENABLED);
   }
 
+  public UpdateClusterConfigQueryParams setDegradedModeEnabled(boolean degradedModeEnabled) {
+    return putBoolean(DEGRADED_MODE_ENABLED, degradedModeEnabled);
+  }
+
+  public Optional<Boolean> getDegradedModeEnabled() {
+    return getBoolean(DEGRADED_MODE_ENABLED);
+  }
+
   // ***************** above this line are getters and setters *****************
 
   private UpdateClusterConfigQueryParams putBoolean(String name, boolean value) {

internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixReadOnlyDegradedDcStatesRepository.java

@@ -6,7 +6,7 @@
 
 /**
  * Represents metadata about a datacenter that has been marked as degraded.
- * Stored in a separate ZK node per cluster, not in LiveClusterConfig.
+ * Stored as a value in {@link LiveClusterConfig#getDegradedDatacenters()} (one entry per DC).
  */
 public class DegradedDcInfo {
   @JsonProperty("timestamp")