feat(ingestion-pipelines): add agentType filter and stop polling orchestrators for queued status

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/IngestionPipelineAgentTypeIT.java

@@ -0,0 +1,133 @@
+package org.openmetadata.it.tests;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Instant;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.openmetadata.it.factories.DashboardServiceTestFactory;
+import org.openmetadata.it.util.SdkClients;
+import org.openmetadata.it.util.TestNamespace;
+import org.openmetadata.it.util.TestNamespaceExtension;
+import org.openmetadata.schema.api.services.ingestionPipelines.CreateIngestionPipeline;
+import org.openmetadata.schema.entity.services.DashboardService;
+import org.openmetadata.schema.entity.services.ingestionPipelines.AirflowConfig;
+import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
+import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineType;
+import org.openmetadata.schema.metadataIngestion.ApplicationPipeline;
+import org.openmetadata.schema.metadataIngestion.DashboardServiceMetadataPipeline;
+import org.openmetadata.schema.metadataIngestion.SourceConfig;
+import org.openmetadata.schema.utils.ResultList;
+import org.openmetadata.sdk.client.OpenMetadataClient;
+import org.openmetadata.sdk.network.HttpMethod;
+
+/**
+ * Integration tests for the {@code agentType} list filter on {@code
+ * GET /v1/services/ingestionPipelines}, which expands to the set of {@code pipelineType} values
+ * that make up an agent group so clients do not have to enumerate them.
+ */
+@Execution(ExecutionMode.CONCURRENT)
+@ExtendWith(TestNamespaceExtension.class)
+public class IngestionPipelineAgentTypeIT {
+
+  private static final Date START_DATE = Date.from(Instant.parse("2022-06-10T15:06:47Z"));
+  private static final String LIST_PATH = "/v1/services/ingestionPipelines";
+
+  @Test
+  void test_agentTypeMetadata_returnsOnlyMetadataAgents(TestNamespace ns) {
+    OpenMetadataClient adminClient = SdkClients.adminClient();
+    DashboardService service = DashboardServiceTestFactory.createMetabase(ns);
+    String serviceFqn = service.getFullyQualifiedName();
+
+    try {
+      String metadataPipeline = createPipeline(ns, service, "agentMetadata", PipelineType.METADATA);
+      String lineagePipeline = createPipeline(ns, service, "agentLineage", PipelineType.LINEAGE);
+      String reindexPipeline =
+          createPipeline(ns, service, "agentReindex", PipelineType.ELASTIC_SEARCH_REINDEX);
+      String applicationPipeline =
+          createPipeline(ns, service, "agentApplication", PipelineType.APPLICATION);
+
+      Set<String> metadataAgents = listNames(adminClient, serviceFqn, "agentType=metadata");
+      assertTrue(
+          metadataAgents.containsAll(Set.of(metadataPipeline, lineagePipeline)),
+          "agentType=metadata must return every metadata pipelineType");
+      assertTrue(
+          Set.of(reindexPipeline, applicationPipeline).stream().noneMatch(metadataAgents::contains),
+          "agentType=metadata must not fall back to 'everything that is not an application'");
+
+      Set<String> applicationAgents = listNames(adminClient, serviceFqn, "agentType=application");
+      assertEquals(Set.of(applicationPipeline), applicationAgents);
+    } finally {
+      adminClient
+          .dashboardServices()
+          .delete(service.getId().toString(), Map.of("hardDelete", "true", "recursive", "true"));
+    }
+  }
+
+  @Test
+  void test_agentTypeIsIntersectedWithPipelineType(TestNamespace ns) {
+    OpenMetadataClient adminClient = SdkClients.adminClient();
+    DashboardService service = DashboardServiceTestFactory.createMetabase(ns);
+    String serviceFqn = service.getFullyQualifiedName();
+
+    try {
+      createPipeline(ns, service, "narrowMetadata", PipelineType.METADATA);
+      String lineagePipeline = createPipeline(ns, service, "narrowLineage", PipelineType.LINEAGE);
+      createPipeline(ns, service, "narrowApplication", PipelineType.APPLICATION);
+
+      assertEquals(
+          Set.of(lineagePipeline),
+          listNames(adminClient, serviceFqn, "agentType=metadata&pipelineType=lineage"),
+          "Both filters set must intersect, not union");
+      assertTrue(
+          listNames(adminClient, serviceFqn, "agentType=metadata&pipelineType=application")
+              .isEmpty(),
+          "An empty intersection must match no pipeline");
+    } finally {
+      adminClient
+          .dashboardServices()
+          .delete(service.getId().toString(), Map.of("hardDelete", "true", "recursive", "true"));
+    }
+  }
+
+  private String createPipeline(
+      TestNamespace ns, DashboardService service, String name, PipelineType pipelineType) {
+    SourceConfig sourceConfig =
+        PipelineType.APPLICATION.equals(pipelineType)
+            ? new SourceConfig()
+                .withConfig(
+                    new ApplicationPipeline().withAppConfig(Map.of("type", "AgentTypeTestApp")))
+            : new SourceConfig().withConfig(new DashboardServiceMetadataPipeline());
+    IngestionPipeline pipeline =
+        SdkClients.adminClient()
+            .ingestionPipelines()
+            .create(
+                new CreateIngestionPipeline()
+                    .withName(ns.prefix(name))
+                    .withPipelineType(pipelineType)
+                    .withService(service.getEntityReference())
+                    .withSourceConfig(sourceConfig)
+                    .withAirflowConfig(new AirflowConfig().withStartDate(START_DATE)));
+    return pipeline.getName();
+  }
+
+  private Set<String> listNames(OpenMetadataClient client, String serviceFqn, String filter) {
+    String path = String.format("%s?service=%s&limit=100&%s", LIST_PATH, serviceFqn, filter);
+    IngestionPipelineList response =
+        client.getHttpClient().execute(HttpMethod.GET, path, null, IngestionPipelineList.class);
+    List<IngestionPipeline> data = response.getData();
+    return data == null
+        ? Set.of()
+        : data.stream().map(IngestionPipeline::getName).collect(Collectors.toSet());
+  }
+
+  static class IngestionPipelineList extends ResultList<IngestionPipeline> {}
+}

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/IngestionPipelineResourceIT.java

@@ -46,6 +46,7 @@
 import org.openmetadata.schema.entity.services.ingestionPipelines.Progress;
 import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressProperty;
 import org.openmetadata.schema.entity.services.ingestionPipelines.StepSummary;
+import org.openmetadata.schema.metadataIngestion.ApplicationPipeline;
 import org.openmetadata.schema.metadataIngestion.DashboardServiceMetadataPipeline;
 import org.openmetadata.schema.metadataIngestion.DatabaseServiceMetadataPipeline;
 import org.openmetadata.schema.metadataIngestion.DatabaseServiceQueryUsagePipeline;
@@ -59,6 +60,7 @@
 import org.openmetadata.schema.type.EntityHistory;
 import org.openmetadata.schema.type.ProviderType;
 import org.openmetadata.schema.utils.JsonUtils;
+import org.openmetadata.schema.utils.ResultList;
 import org.openmetadata.sdk.client.OpenMetadataClient;
 import org.openmetadata.sdk.exceptions.OpenMetadataException;
 import org.openmetadata.sdk.models.ListParams;
@@ -1877,4 +1879,108 @@ private static int logEndpointStatus(String path) throws Exception {
   private static String encodeSegment(String value) {
     return URLEncoder.encode(value, StandardCharsets.UTF_8).replace("+", "%20");
   }
+
+  /**
+   * A run the orchestrator accepts but never starts leaves a `queued` status behind that no worker
+   * will ever supersede, so it is hidden once older than `queuedStatusTimeoutSeconds`. That cutoff
+   * has to hold for the `pipelineStatuses` entity field too, not just the pipelineStatus endpoint —
+   * the Agents page reads the field, and shows the newest entry as the pipeline's current state.
+   */
+  @Test
+  void test_staleQueuedStatusIsHiddenFromThePipelineStatusesField(TestNamespace ns) {
+    IngestionPipeline pipeline = createEntity(createRequest(ns.prefix("staleQueued"), ns));
+    String fqn = pipeline.getFullyQualifiedName();
+    String serviceFqn = pipeline.getService().getFullyQualifiedName();
+    long twoHoursAgo = System.currentTimeMillis() - Duration.ofHours(2).toMillis();
+    long ninetyMinutesAgo = System.currentTimeMillis() - Duration.ofMinutes(90).toMillis();
+
+    addStatus(fqn, "stale-queued-run", PipelineStatusType.QUEUED, twoHoursAgo);
+    addStatus(fqn, "finished-run", PipelineStatusType.SUCCESS, ninetyMinutesAgo);
+
+    assertEquals(
+        List.of("finished-run"),
+        runIdsOf(
+            get(
+                    "/v1/services/ingestionPipelines/" + encodeSegment(fqn) + "/pipelineStatus",
+                    PipelineStatusList.class)
+                .getData()),
+        "pipelineStatus endpoint must hide the stale queued run");
+
+    // setFields path: single entity read with the field requested
+    assertEquals(
+        List.of("finished-run"),
+        runIdsOf(
+            get(
+                    "/v1/services/ingestionPipelines/name/"
+                        + encodeSegment(fqn)
+                        + "?fields=pipelineStatuses",
+                    IngestionPipeline.class)
+                .getPipelineStatuses()),
+        "pipelineStatuses field must hide it too, or the Agents page shows Queued forever");
+
+    // setFieldsInBulk path: the list call the Agents page actually makes
+    IngestionPipeline fromList =
+        get(
+                "/v1/services/ingestionPipelines?limit=100&fields=pipelineStatuses&service="
+                    + encodeSegment(serviceFqn),
+                IngestionPipelineList.class)
+            .getData()
+            .stream()
+            .filter(p -> fqn.equals(p.getFullyQualifiedName()))
+            .findFirst()
+            .orElseThrow(() -> new AssertionError("pipeline missing from the list response"));
+    assertEquals(
+        List.of("finished-run"),
+        runIdsOf(fromList.getPipelineStatuses()),
+        "the bulk field fetch must apply the same cutoff as the single-entity read");
+  }
+
+  private static List<String> runIdsOf(List<PipelineStatus> statuses) {
+    return statuses == null ? List.of() : statuses.stream().map(PipelineStatus::getRunId).toList();
+  }
+
+  private static <T> T get(String path, Class<T> type) {
+    return SdkClients.adminClient().getHttpClient().execute(HttpMethod.GET, path, null, type);
+  }
+
+  private void addStatus(String fqn, String runId, PipelineStatusType state, long timestamp) {
+    SdkClients.adminClient()
+        .getHttpClient()
+        .execute(
+            HttpMethod.PUT,
+            "/v1/services/ingestionPipelines/" + encodeSegment(fqn) + "/pipelineStatus",
+            new PipelineStatus()
+                .withRunId(runId)
+                .withPipelineState(state)
+                .withStartDate(timestamp)
+                .withTimestamp(timestamp),
+            IngestionPipeline.class);
+  }
+
+  static class PipelineStatusList extends ResultList<PipelineStatus> {}
+
+  static class IngestionPipelineList extends ResultList<IngestionPipeline> {}
+
+  /**
+   * Creating an application pipeline reads the app type off `appConfig` to pick a specific create
+   * permission. With no `appConfig` there is no type to read, and that used to escape as a 500
+   * before authorization even ran instead of falling back to the generic create permission.
+   */
+  @Test
+  void test_createApplicationPipelineWithoutAppConfig(TestNamespace ns) {
+    DatabaseService service = DatabaseServiceTestFactory.createPostgres(ns);
+
+    CreateIngestionPipeline request =
+        new CreateIngestionPipeline()
+            .withName(ns.prefix("appNoConfig"))
+            .withPipelineType(PipelineType.APPLICATION)
+            .withService(service.getEntityReference())
+            .withSourceConfig(new SourceConfig().withConfig(new ApplicationPipeline()))
+            .withAirflowConfig(new AirflowConfig().withStartDate(START_DATE));
+
+    IngestionPipeline pipeline = createEntity(request);
+
+    assertNotNull(pipeline.getId());
+    assertEquals(PipelineType.APPLICATION, pipeline.getPipelineType());
+  }
 }

openmetadata-service/src/main/java/org/openmetadata/service/clients/pipeline/k8s/K8sPipelineClient.java

@@ -67,7 +67,6 @@
 import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineServiceClientResponse;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatus;
-import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatusType;
 import org.openmetadata.schema.utils.JsonUtils;
 import org.openmetadata.sdk.PipelineServiceClientInterface;
 import org.openmetadata.sdk.exception.PipelineServiceClientException;
@@ -603,7 +602,8 @@ public PipelineServiceClientResponse runPipeline(
             correlationId);
       }
       return buildSuccessResponse(
-          "Pipeline triggered successfully", Map.of("runId", runId, "jobName", jobName));
+              "Pipeline triggered successfully", Map.of("runId", runId, "jobName", jobName))
+          .withRunId(runId);
 
     } catch (ApiException e) {
       LOG.error(
@@ -887,72 +887,15 @@ public PipelineServiceClientResponse killIngestion(IngestionPipeline ingestionPi
     }
   }
 
+  /**
+   * Returns nothing on purpose. This client mints the run ID when triggering (see {@link
+   * #runPipeline}) and reports it back on the response, so the server persists the {@code queued}
+   * status itself. Listing Jobs here on every run-history read would cost an API server round trip
+   * for state we already hold.
+   */
   @Override
   public List<PipelineStatus> getQueuedPipelineStatusInternal(IngestionPipeline ingestionPipeline) {
-    // READ-ONLY: Check for queued K8s jobs without storing anything
-    String pipelineName = sanitizeName(ingestionPipeline.getName());
-    List<PipelineStatus> queuedStatuses = new ArrayList<>();
-
-    try {
-      String labelSelector = LABEL_PIPELINE + "=" + pipelineName;
-      V1JobList jobs =
-          batchApi
-              .listNamespacedJob(k8sConfig.getNamespace())
-              .labelSelector(labelSelector)
-              .execute();
-
-      for (V1Job job : jobs.getItems()) {
-        // Only return jobs that are QUEUED (created but not started)
-        if (isJobQueued(job)) {
-          String runId =
-              StringUtils.defaultIfBlank(
-                  job.getMetadata().getLabels() != null
-                      ? job.getMetadata().getLabels().get(LABEL_RUN_ID)
-                      : null,
-                  job.getMetadata().getName());
-
-          Long startTime =
-              job.getMetadata().getCreationTimestamp() != null
-                  ? job.getMetadata().getCreationTimestamp().toInstant().toEpochMilli()
-                  : null;
-
-          // Create READ-ONLY status object (not persisted)
-          PipelineStatus queuedStatus =
-              new PipelineStatus()
-                  .withRunId(runId)
-                  .withPipelineState(PipelineStatusType.QUEUED)
-                  .withStartDate(startTime)
-                  .withTimestamp(startTime);
-
-          queuedStatuses.add(queuedStatus);
-        }
-      }
-
-    } catch (ApiException e) {
-      LOG.error("Failed to check queued pipeline status: {}", e.getResponseBody());
-    }
-
-    return queuedStatuses;
-  }
-
-  private boolean isJobQueued(V1Job job) {
-    if (job.getStatus() == null) {
-      return true; // Job created but status not yet set = queued
-    }
-
-    Integer active = job.getStatus().getActive();
-    Integer succeeded = job.getStatus().getSucceeded();
-    Integer failed = job.getStatus().getFailed();
-
-    // Check if job is being deleted (has deletion timestamp)
-    if (job.getMetadata() != null && job.getMetadata().getDeletionTimestamp() != null) {
-      return false; // Job is being terminated, not queued
-    }
-
-    // Queued = no active pods, no completed pods, no failed pods, and not being deleted
-    return (active == null || active == 0)
-        && (succeeded == null || succeeded == 0)
-        && (failed == null || failed == 0);
+    return List.of();
   }
 
   @Override

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineImpl.java

@@ -10,6 +10,7 @@
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
+import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineServiceClientResponse;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatus;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatusType;
 import org.openmetadata.schema.type.Include;
@@ -96,12 +97,16 @@ private void runIngestionPipeline(IngestionPipeline ingestionPipeline) {
     Retry retry = Retry.of("runIngestionPipeline", retryConfig);
 
     try {
-      retry.executeRunnable(
-          () ->
-              pipelineServiceClient.runPipeline(
-                  ingestionPipeline,
-                  Entity.getEntity(
-                      ingestionPipeline.getService(), "ingestionRunner", Include.NON_DELETED)));
+      PipelineServiceClientResponse response =
+          retry.executeSupplier(
+              () ->
+                  pipelineServiceClient.runPipeline(
+                      ingestionPipeline,
+                      Entity.getEntity(
+                          ingestionPipeline.getService(), "ingestionRunner", Include.NON_DELETED)));
+      ((IngestionPipelineRepository) Entity.getEntityRepository(Entity.INGESTION_PIPELINE))
+          .recordQueuedPipelineStatus(
+              null, ingestionPipeline.getFullyQualifiedName(), response.getRunId());
     } catch (Exception ex) {
       throw new RuntimeException("Failed to run pipeline after retries: " + ex.getMessage(), ex);
     }