Reject ingestion pipeline sourceConfig without type

ingestion/src/metadata/workflow/base.py

@@ -40,12 +40,14 @@
 )
 from metadata.generated.schema.metadataIngestion.workflow import (
     LogLevels,
+    SourceConfig,
     WorkflowConfig,
 )
 from metadata.generated.schema.tests.testSuite import ServiceType
 from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.ingestion import diagnostics
 from metadata.ingestion.api.step import Step, Summary
+from metadata.ingestion.models.custom_pydantic import BaseModel as OpenMetadataBaseModel
 from metadata.ingestion.ometa.client_utils import create_ometa_client
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.timer.repeated_timer import RepeatedTimer
@@ -367,6 +369,18 @@ def run_id(self) -> str:
 
         return self._run_id
 
+    def _source_config_with_explicit_type(self) -> SourceConfig:
+        source_config = self.config.source.sourceConfig
+        config = source_config.config
+        if not isinstance(config, OpenMetadataBaseModel):
+            return source_config
+
+        config_type = getattr(config, "type", None)
+        if config_type is None:
+            return source_config
+
+        return source_config.model_copy(update={"config": config.model_copy(update={"type": config_type})})
+
     def get_or_create_ingestion_pipeline(self) -> Optional[IngestionPipeline]:  # noqa: UP045
         """
         If we get the `ingestionPipelineFqn` from the `workflowConfig`, it means we want to
@@ -404,7 +418,7 @@ def get_or_create_ingestion_pipeline(self) -> Optional[IngestionPipeline]:  # no
                             type=get_reference_type_from_service_type(self.service_type),
                         ),
                         pipelineType=get_pipeline_type_from_source_config(self.config.source.sourceConfig),
-                        sourceConfig=self.config.source.sourceConfig,
+                        sourceConfig=self._source_config_with_explicit_type(),
                         airflowConfig=AirflowConfig(),
                         enableStreamableLogs=self.config.enableStreamableLogs,
                     )

ingestion/tests/unit/workflow/test_base_workflow.py

@@ -12,6 +12,8 @@
 Validate the logic and status handling of the base workflow
 """
 
+import uuid
+from types import SimpleNamespace
 from typing import Iterable, Tuple  # noqa: UP035
 from unittest import TestCase
 from unittest.mock import MagicMock, patch
@@ -201,6 +203,55 @@ def test_workflow_config_supports_ingestion_runner_name(self):
         self.assertEqual(workflow_config.ingestionRunnerName, "test-runner")
 
 
+def _self_registration_request(source_config):
+    workflow_config = config.model_copy(
+        update={
+            "source": config.source.model_copy(
+                update={"type": "mysql", "sourceConfig": SourceConfig(config=source_config)}
+            )
+        },
+        deep=True,
+    )
+
+    metadata = MagicMock()
+    metadata.config.forceEntityOverwriting = True
+    with patch("metadata.workflow.base.create_ometa_client", return_value=metadata):
+        workflow = SimpleWorkflow(config=workflow_config)
+
+    workflow.config = workflow_config.model_copy(
+        update={"ingestionPipelineFQN": "test-service.self-registered-pipeline"}
+    )
+    workflow.metadata.get_by_name.return_value = None
+    with patch.object(
+        workflow,
+        "_get_ingestion_pipeline_service",
+        return_value=SimpleNamespace(id=uuid.uuid4()),
+    ):
+        workflow.get_or_create_ingestion_pipeline()
+
+    return workflow.metadata.create_or_update.call_args.args[0]
+
+
+def test_self_registration_serializes_default_source_config_type():
+    source_config = DatabaseServiceMetadataPipeline()
+    assert "type" not in source_config.model_dump(exclude_unset=True)
+
+    request = _self_registration_request(source_config)
+    payload = request.model_dump(mode="json", exclude_unset=True, exclude_none=True)
+    assert payload["sourceConfig"]["config"]["type"] == "DatabaseMetadata"
+    assert "type" not in source_config.model_dump(exclude_unset=True)
+
+
+def test_self_registration_preserves_explicit_source_config_type():
+    source_config = DatabaseServiceMetadataPipeline(type="DatabaseMetadata")
+    assert source_config.model_dump(mode="json", exclude_unset=True)["type"] == "DatabaseMetadata"
+
+    request = _self_registration_request(source_config)
+    payload = request.model_dump(mode="json", exclude_unset=True, exclude_none=True)
+
+    assert payload["sourceConfig"]["config"]["type"] == "DatabaseMetadata"
+
+
 class TestWorkflowExecuteTeardown:
     """
     Validates the execute() teardown contract:

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -99,6 +100,55 @@ public class IngestionPipelineRepository extends EntityRepository<IngestionPipel
       "sourceConfig,airflowConfig,loggerLevel,enabled,deployed,processingEngine";
   private static final String PATCH_FIELDS =
       "sourceConfig,airflowConfig,loggerLevel,enabled,deployed,processingEngine";
+  private static final String SOURCE_CONFIG_TYPE = "type";
+  private static final String SOURCE_CONFIG_TYPE_REQUIRED = "sourceConfig.config.type is required";
+  private static final String SOURCE_CONFIG_OBJECT_REQUIRED =
+      "sourceConfig.config must be an object with type";
+  private static final String REVERSE_INGESTION_OPERATIONS = "operations";
+  private static final String REVERSE_INGESTION_CONFIG_TYPE = "ReverseIngestion";
+
+  private static final Map<String, Map<PipelineType, String>> LEGACY_SOURCE_CONFIG_TYPES =
+      Map.ofEntries(
+          Map.entry(
+              Entity.DATABASE_SERVICE,
+              Map.of(
+                  PipelineType.METADATA, "DatabaseMetadata",
+                  PipelineType.USAGE, "DatabaseUsage",
+                  PipelineType.LINEAGE, "DatabaseLineage",
+                  PipelineType.PROFILER, "Profiler",
+                  PipelineType.AUTO_CLASSIFICATION, "AutoClassification")),
+          Map.entry(
+              Entity.DASHBOARD_SERVICE,
+              Map.of(
+                  PipelineType.METADATA, "DashboardMetadata",
+                  PipelineType.LINEAGE, "DashboardMetadata")),
+          Map.entry(
+              Entity.MESSAGING_SERVICE,
+              Map.of(
+                  PipelineType.METADATA, "MessagingMetadata",
+                  PipelineType.AUTO_CLASSIFICATION, "AutoClassification")),
+          Map.entry(Entity.PIPELINE_SERVICE, Map.of(PipelineType.METADATA, "PipelineMetadata")),
+          Map.entry(Entity.MLMODEL_SERVICE, Map.of(PipelineType.METADATA, "MlModelMetadata")),
+          Map.entry(
+              Entity.STORAGE_SERVICE,
+              Map.of(
+                  PipelineType.METADATA, "StorageMetadata",
+                  PipelineType.AUTO_CLASSIFICATION, "AutoClassification")),
+          Map.entry(Entity.DRIVE_SERVICE, Map.of(PipelineType.METADATA, "DriveMetadata")),
+          Map.entry(Entity.SEARCH_SERVICE, Map.of(PipelineType.METADATA, "SearchMetadata")),
+          Map.entry(Entity.API_SERVICE, Map.of(PipelineType.METADATA, "ApiMetadata")),
+          Map.entry(Entity.MCP_SERVICE, Map.of(PipelineType.METADATA, "McpMetadata")),
+          Map.entry(Entity.SECURITY_SERVICE, Map.of(PipelineType.METADATA, "SecurityMetadata")),
+          Map.entry(Entity.METADATA_SERVICE, Map.of(PipelineType.METADATA, "DatabaseMetadata")));
+
+  private static final Map<PipelineType, String> LEGACY_SOURCE_CONFIG_TYPES_BY_PIPELINE =
+      Map.of(
+          PipelineType.DBT, "DBT",
+          PipelineType.TEST_SUITE, "TestSuite",
+          PipelineType.DATA_INSIGHT, "dataInsight",
+          PipelineType.ELASTIC_SEARCH_REINDEX, "MetadataToElasticSearch",
+          PipelineType.APPLICATION, "Application",
+          PipelineType.POLICY_AGENT, "PolicyAgent");
 
   private static final String PIPELINE_STATUS_JSON_SCHEMA = "ingestionPipelineStatus";
   public static final String PIPELINE_STATUS_EXTENSION = "ingestionPipeline.pipelineStatus";
@@ -510,6 +560,79 @@ public void clearFields(IngestionPipeline ingestionPipeline, Fields fields) {
   public void prepare(IngestionPipeline ingestionPipeline, boolean update) {
     var service = getCachedParentOrLoad(ingestionPipeline.getService(), "", Include.NON_DELETED);
     ingestionPipeline.setService(service.getEntityReference());
+    validateSourceConfigHasType(ingestionPipeline);
+  }
+
+  static void validateSourceConfigHasType(IngestionPipeline ingestionPipeline) {
+    Object config = getRequiredSourceConfig(ingestionPipeline);
+    Map<?, ?> configMap = getSourceConfigMap(config);
+    if (!hasSourceConfigType(config, configMap)) {
+      throw new BadRequestException(SOURCE_CONFIG_TYPE_REQUIRED);
+    }
+  }
+
+  private static void repairLegacySourceConfig(IngestionPipeline ingestionPipeline) {
+    Object config = getRequiredSourceConfig(ingestionPipeline);
+    Map<?, ?> configMap = getSourceConfigMap(config);
+    if (ingestionPipeline.getId() == null || !isMissingOrBlankSourceConfigType(configMap)) {
+      return;
+    }
+
+    String sourceConfigType = getLegacySourceConfigType(ingestionPipeline, configMap);
+    if (sourceConfigType == null) {
+      return;
+    }
+
+    Map<String, Object> repairedConfig = new LinkedHashMap<>();
+    configMap.forEach((key, value) -> repairedConfig.put(String.valueOf(key), value));
+    repairedConfig.put(SOURCE_CONFIG_TYPE, sourceConfigType);
+    ingestionPipeline.getSourceConfig().setConfig(repairedConfig);
+  }
+
+  private static boolean isMissingOrBlankSourceConfigType(Map<?, ?> configMap) {
+    Object type = configMap.get(SOURCE_CONFIG_TYPE);
+    return type == null || type instanceof String typeValue && typeValue.isBlank();
+  }
+
+  private static Object getRequiredSourceConfig(IngestionPipeline ingestionPipeline) {
+    if (ingestionPipeline.getSourceConfig() == null
+        || ingestionPipeline.getSourceConfig().getConfig() == null) {
+      throw new BadRequestException(SOURCE_CONFIG_TYPE_REQUIRED);
+    }
+    return ingestionPipeline.getSourceConfig().getConfig();
+  }
+
+  private static Map<?, ?> getSourceConfigMap(Object config) {
+    try {
+      return config instanceof Map<?, ?> map ? map : JsonUtils.getMap(config);
+    } catch (IllegalArgumentException e) {
+      throw new BadRequestException(SOURCE_CONFIG_OBJECT_REQUIRED);
+    }
+  }
+
+  private static boolean hasSourceConfigType(Object config, Map<?, ?> configMap) {
+    Object type = configMap.get(SOURCE_CONFIG_TYPE);
+    return type instanceof String typeValue && !typeValue.isBlank()
+        || !(config instanceof Map<?, ?>) && type instanceof Enum<?>;
+  }
+
+  private static String getLegacySourceConfigType(
+      IngestionPipeline ingestionPipeline, Map<?, ?> configMap) {
+    if (configMap.containsKey(REVERSE_INGESTION_OPERATIONS)) {
+      return REVERSE_INGESTION_CONFIG_TYPE;
+    }
+
+    Map<PipelineType, String> serviceConfigTypes =
+        ingestionPipeline.getService() == null
+            ? null
+            : LEGACY_SOURCE_CONFIG_TYPES.get(ingestionPipeline.getService().getType());
+    String sourceConfigType =
+        serviceConfigTypes == null
+            ? null
+            : serviceConfigTypes.get(ingestionPipeline.getPipelineType());
+    return sourceConfigType == null
+        ? LEGACY_SOURCE_CONFIG_TYPES_BY_PIPELINE.get(ingestionPipeline.getPipelineType())
+        : sourceConfigType;
   }
 
   protected boolean requiresRedeployment(IngestionPipeline original, IngestionPipeline updated) {
@@ -1475,20 +1598,55 @@ public boolean isProgressTrackingEnabled() {
   }
 
   public void streamProgress(String pipelineFQN, UUID runId, SseEventSink eventSink, Sse sse) {
-    Consumer<ProgressUpdate> listener = update -> emitProgressUpdate(eventSink, sse, update);
+    ProgressStreamEventEmitter emitter = new ProgressStreamEventEmitter(eventSink, sse);
+    Consumer<ProgressUpdate> listener = emitter::emitLiveUpdate;
     Runnable onClose =
         () -> progressTracker.unregisterProgressListener(pipelineFQN, runId, listener);
     if (ProgressSseManager.getInstance().register(eventSink, sse, onClose)) {
       progressTracker.registerProgressListener(pipelineFQN, runId, listener);
       ProgressUpdate snapshot = getLatestProgressUpdate(pipelineFQN, runId);
       if (snapshot != null) {
-        emitProgressUpdate(eventSink, sse, snapshot);
+        emitter.emitSnapshot(snapshot);
       }
     } else {
       eventSink.close();
     }
   }
 
+  /**
+   * Delivers the initial progress snapshot and live updates in one ordered stream. A listener must
+   * be registered before reading the snapshot so updates cannot be lost. When a live update races
+   * with that read, the tracker stores and dispatches the same {@link ProgressUpdate} instance;
+   * emitting the snapshot first marks that instance so its delayed listener callback is not replayed.
+   */
+  private final class ProgressStreamEventEmitter {
+    private final SseEventSink eventSink;
+    private final Sse sse;
+    private ProgressUpdate emittedSnapshot;
+    private boolean liveUpdateDelivered;
+
+    private ProgressStreamEventEmitter(SseEventSink eventSink, Sse sse) {
+      this.eventSink = eventSink;
+      this.sse = sse;
+    }
+
+    private synchronized void emitLiveUpdate(ProgressUpdate update) {
+      liveUpdateDelivered = true;
+      if (emittedSnapshot == update) {
+        return;
+      }
+      emitProgressUpdate(eventSink, sse, update);
+    }
+
+    private synchronized void emitSnapshot(ProgressUpdate snapshot) {
+      if (liveUpdateDelivered) {
+        return;
+      }
+      emittedSnapshot = snapshot;
+      emitProgressUpdate(eventSink, sse, snapshot);
+    }
+  }
+
   private void emitProgressUpdate(SseEventSink eventSink, Sse sse, ProgressUpdate update) {
     CompletionStage<?> event = sendProgressEvent(eventSink, sse, update);
     if (isTerminalProgressUpdate(update)) {
@@ -1564,6 +1722,8 @@ public RestUtil.PutResponse<?> addOperationMetrics(
 
   public PipelineServiceClientResponse deployIngestionPipeline(
       IngestionPipeline ingestionPipeline, ServiceEntityInterface service) {
+    repairLegacySourceConfig(ingestionPipeline);
+    validateSourceConfigHasType(ingestionPipeline);
     applyStreamableLogsConfig(ingestionPipeline);
     return pipelineServiceClient.deployPipeline(ingestionPipeline, service);
   }