Add K8s native scheduler (#24940)

Author: harshach

openmetadata-service/pom.xml

@@ -641,6 +641,12 @@
       <version>${org.testcontainers.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>k3s</artifactId>
+      <version>${org.testcontainers.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.opensearch</groupId>
       <artifactId>opensearch-testcontainers</artifactId>
@@ -926,6 +932,11 @@
       <artifactId>resilience4j-ratelimiter</artifactId>
       <version>${resilience4j-ratelimiter.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.github.resilience4j</groupId>
+      <artifactId>resilience4j-retry</artifactId>
+      <version>${resilience4j-ratelimiter.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.junit.platform</groupId>
       <artifactId>junit-platform-commons</artifactId>

openmetadata-service/src/main/java/org/openmetadata/service/clients/pipeline/MeteredPipelineServiceClient.java

@@ -34,6 +34,12 @@ public MeteredPipelineServiceClient(PipelineServiceClientInterface decoratedClie
     this.decoratedClient = decoratedClient;
   }
 
+  /** Get the underlying decorated client. Used for testing to check the actual client type. */
+  @com.google.common.annotations.VisibleForTesting
+  public PipelineServiceClientInterface getDecoratedClient() {
+    return decoratedClient;
+  }
+
   private <T> T executeWithMetering(String name, Supplier<T> operation) {
     try {
       T result = operation.get();

openmetadata-service/src/main/java/org/openmetadata/service/clients/pipeline/PipelineServiceClientFactory.java

@@ -27,39 +27,97 @@ private PipelineServiceClientFactory() {
     // Final class
   }
 
-  @Getter private static PipelineServiceClientInterface pipelineServiceClient;
+  /** Lock for thread-safe client creation. */
+  private static final Object LOCK = new Object();
 
+  /** Volatile to ensure visibility across threads. */
+  @Getter private static volatile PipelineServiceClientInterface pipelineServiceClient;
+
+  /**
+   * Reset the cached pipeline service client. This is used by tests that need to switch between
+   * different pipeline client implementations (e.g., MockPipelineServiceClient vs
+   * K8sPipelineClient).
+   */
+  @com.google.common.annotations.VisibleForTesting
+  public static void reset() {
+    synchronized (LOCK) {
+      pipelineServiceClient = null;
+    }
+  }
+
+  /**
+   * Creates or returns the cached pipeline service client. Thread-safe with double-checked locking.
+   *
+   * @param config The pipeline service client configuration
+   * @return The pipeline service client instance
+   */
   public static PipelineServiceClientInterface createPipelineServiceClient(
       PipelineServiceClientConfiguration config) {
-    if (pipelineServiceClient != null || CommonUtil.nullOrEmpty(config)) {
+    if (CommonUtil.nullOrEmpty(config)) {
       return pipelineServiceClient;
     }
 
-    if (Boolean.FALSE.equals(config.getEnabled())) {
-      LOG.debug("Pipeline Service Client is disabled. Skipping initialization.");
-      return null;
+    // Fast path: check if we can reuse cached client without locking
+    PipelineServiceClientInterface cached = pipelineServiceClient;
+    if (cached != null && isSameClientType(cached, config.getClassName())) {
+      return cached;
     }
 
-    String pipelineServiceClientClass = config.getClassName();
-    LOG.debug("Registering PipelineServiceClient: {}", pipelineServiceClientClass);
+    // Slow path: synchronized creation
+    synchronized (LOCK) {
+      // Double-check after acquiring lock
+      cached = pipelineServiceClient;
+      if (cached != null && isSameClientType(cached, config.getClassName())) {
+        return cached;
+      }
 
-    try {
-      PipelineServiceClientInterface client =
-          Class.forName(pipelineServiceClientClass)
-              .asSubclass(PipelineServiceClient.class)
-              .getConstructor(PipelineServiceClientConfiguration.class)
-              .newInstance(config);
-      pipelineServiceClient = new MeteredPipelineServiceClient(client);
-      return pipelineServiceClient;
-    } catch (ClassNotFoundException
-        | NoSuchMethodException
-        | InvocationTargetException
-        | InstantiationException
-        | IllegalAccessException e) {
-      throw new PipelineServiceClientException(
-          String.format(
-              "Error trying to load PipelineServiceClient %s: %s",
-              pipelineServiceClientClass, e.getCause()));
+      if (cached != null) {
+        LOG.info(
+            "Requested PipelineServiceClient class {} differs from cached {}, creating new client",
+            config.getClassName(),
+            getActualClientClassName(cached));
+      }
+
+      if (Boolean.FALSE.equals(config.getEnabled())) {
+        LOG.debug("Pipeline Service Client is disabled. Skipping initialization.");
+        return null;
+      }
+
+      String pipelineServiceClientClass = config.getClassName();
+      LOG.info("Creating PipelineServiceClient: {}", pipelineServiceClientClass);
+
+      try {
+        PipelineServiceClientInterface client =
+            Class.forName(pipelineServiceClientClass)
+                .asSubclass(PipelineServiceClient.class)
+                .getConstructor(PipelineServiceClientConfiguration.class)
+                .newInstance(config);
+        pipelineServiceClient = new MeteredPipelineServiceClient(client);
+        return pipelineServiceClient;
+      } catch (ClassNotFoundException
+          | NoSuchMethodException
+          | InvocationTargetException
+          | InstantiationException
+          | IllegalAccessException e) {
+        throw new PipelineServiceClientException(
+            String.format(
+                "Error trying to load PipelineServiceClient %s: %s",
+                pipelineServiceClientClass, e.getCause()));
+      }
+    }
+  }
+
+  /** Check if the cached client is of the same type as requested. */
+  private static boolean isSameClientType(
+      PipelineServiceClientInterface cached, String requestedClassName) {
+    return getActualClientClassName(cached).equals(requestedClassName);
+  }
+
+  /** Get the actual client class name, unwrapping MeteredPipelineServiceClient if needed. */
+  private static String getActualClientClassName(PipelineServiceClientInterface client) {
+    if (client instanceof MeteredPipelineServiceClient) {
+      return ((MeteredPipelineServiceClient) client).getDecoratedClient().getClass().getName();
     }
+    return client.getClass().getName();
   }
 }