Fixes #25666: Add retry-with-backoff to getServiceStatus for transient failure recovery

Author: RajdeepKushwaha5

openmetadata-service/src/test/java/org/openmetadata/service/clients/pipeline/airflow/AirflowRESTClientTest.java

@@ -500,6 +500,31 @@ void runPipelineThrowsDeploymentExceptionWhenAirflowReturnsAnError() throws Exce
     }
   }
 
+  @Test
+  void getServiceStatusRecoversFromTransientFailure() throws Exception {
+    try (AirflowTestServer server = new AirflowTestServer()) {
+      String basePath = "/airflow";
+      String healthPath = basePath + "/pluginsv2/api/v2/openmetadata/health-auth";
+      String version = PipelineServiceClient.getServerVersion();
+
+      // First call: detection probe succeeds
+      server.enqueue("GET", healthPath, 200, "{\"version\":\"" + version + "\"}");
+      // Second call (first getServiceStatus attempt): transient 500
+      server.enqueue("GET", healthPath, 500, "{\"error\":\"selector manager closed\"}");
+      // Third call (retry): healthy response
+      server.enqueue("GET", healthPath, 200, "{\"version\":\"" + version + "\"}");
+
+      AirflowRESTClient client = newClient(server, basePath);
+
+      PipelineServiceClientResponse status = client.getServiceStatus();
+
+      assertEquals(200, status.getCode());
+      assertEquals(version, status.getVersion());
+      List<RequestRecord> healthRequests = server.requests("GET", healthPath);
+      assertEquals(3, healthRequests.size());
+    }
+  }
+
   private static AirflowRESTClient newClient(AirflowTestServer server, String basePath)
       throws KeyStoreException {
     Parameters parameters = new Parameters();

openmetadata-spec/src/main/java/org/openmetadata/service/clients/pipeline/PipelineServiceClient.java

@@ -27,7 +27,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import lombok.Getter;
 import lombok.Setter;
@@ -196,38 +195,36 @@ private Response getHostIpInternal() {
 
   /**
    * Check the pipeline service status with an exception backoff to make sure we don't raise any
-   * false positives.
+   * false positives. Delegates to {@link #getServiceStatus()}, which already retries transient
+   * failures.
    */
   public String getServiceStatusBackoff() {
+    PipelineServiceClientResponse status = getServiceStatus();
+    return status.getCode() != 200 ? UNHEALTHY_STATUS : HEALTHY_STATUS;
+  }
+
+  /**
+   * Check the status of pipeline service to ensure it is healthy. Retries up to {@link
+   * #MAX_ATTEMPTS} times with {@link #BACKOFF_TIME_SECONDS}-second backoff so that a single
+   * transient failure (e.g. Airflow restart, network blip) does not permanently mark the agent as
+   * unavailable.
+   */
+  public PipelineServiceClientResponse getServiceStatus() {
+    if (!pipelineServiceClientEnabled) {
+      return buildHealthyStatus(DISABLED_STATUS).withPlatform(DISABLED_STATUS);
+    }
+
     RetryConfig retryConfig =
-        RetryConfig.<String>custom()
+        RetryConfig.<PipelineServiceClientResponse>custom()
             .maxAttempts(MAX_ATTEMPTS)
             .waitDuration(Duration.ofMillis(BACKOFF_TIME_SECONDS * 1_000L))
-            .retryOnResult(response -> !HEALTHY_STATUS.equals(response))
+            .retryOnResult(response -> response.getCode() != 200)
             .failAfterMaxAttempts(false)
             .build();
 
     Retry retry = Retry.of("getServiceStatus", retryConfig);
 
-    Supplier<String> responseSupplier =
-        () -> {
-          try {
-            PipelineServiceClientResponse status = getServiceStatus();
-            return status.getCode() != 200 ? UNHEALTHY_STATUS : HEALTHY_STATUS;
-          } catch (Exception e) {
-            throw new RuntimeException(e);
-          }
-        };
-
-    return retry.executeSupplier(responseSupplier);
-  }
-
-  /* Check the status of pipeline service to ensure it is healthy */
-  public PipelineServiceClientResponse getServiceStatus() {
-    if (pipelineServiceClientEnabled) {
-      return getServiceStatusInternal();
-    }
-    return buildHealthyStatus(DISABLED_STATUS).withPlatform(DISABLED_STATUS);
+    return retry.executeSupplier(this::getServiceStatusInternal);
   }
 
   public List<PipelineStatus> getQueuedPipelineStatus(IngestionPipeline ingestionPipeline) {