From 971136a31b58283b50b1381ef4ff0343260e45b0 Mon Sep 17 00:00:00 2001 From: Ayush Shah Date: Mon, 27 Jul 2026 11:13:10 +0530 Subject: [PATCH 1/4] ingestion: reject source configs without type --- .../jdbi3/IngestionPipelineRepository.java | 26 ++++++ .../IngestionPipelineRepositoryTest.java | 84 +++++++++++++++++++ .../support/entity/PipelineClass.ts | 4 +- 3 files changed, 113 insertions(+), 1 deletion(-) diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java index 80d0657c7125..e8a1b89ebb11 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java @@ -510,6 +510,32 @@ 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) { + if (ingestionPipeline.getSourceConfig() == null + || ingestionPipeline.getSourceConfig().getConfig() == null) { + throw new BadRequestException("sourceConfig.config.type is required"); + } + + Object config = ingestionPipeline.getSourceConfig().getConfig(); + Object type; + boolean generatedConfig = !(config instanceof Map); + try { + type = + generatedConfig ? JsonUtils.getMap(config).get("type") : ((Map) config).get("type"); + } catch (IllegalArgumentException e) { + throw new BadRequestException("sourceConfig.config must be an object with type"); + } + + if (type instanceof String typeValue && !typeValue.isBlank()) { + return; + } + if (generatedConfig && type instanceof Enum) { + return; + } + throw new BadRequestException("sourceConfig.config.type is required"); } protected boolean requiresRedeployment(IngestionPipeline original, IngestionPipeline updated) { diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java index a6316e36d27a..6fa837746f57 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java @@ -12,11 +12,18 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.UUID; +import java.util.stream.Stream; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.mockito.Mockito; import org.openmetadata.schema.entity.services.ingestionPipelines.AirflowConfig; import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline; @@ -29,6 +36,7 @@ import org.openmetadata.sdk.PipelineServiceClientInterface; import org.openmetadata.sdk.exception.IngestionRunnerUnavailableException; import org.openmetadata.sdk.exception.PipelineServiceClientException; +import org.openmetadata.service.exception.BadRequestException; import org.openmetadata.service.secrets.SecretsManagerFactory; class IngestionPipelineRepositoryTest { @@ -344,6 +352,78 @@ void testBuildIngestionPipelineDecrypted_ServicePreserved() { assertEquals("OpenMetadata", decrypted.getService().getName()); } + @ParameterizedTest(name = "{0}") + @MethodSource("invalidSourceConfigs") + void validateSourceConfigHasTypeRejectsInvalidConfig( + String testCase, IngestionPipeline pipeline, String expectedMessage) { + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> IngestionPipelineRepository.validateSourceConfigHasType(pipeline)); + + assertEquals(expectedMessage, exception.getMessage()); + } + + @Test + void validateSourceConfigHasTypeAcceptsRawConfigWithType() { + IngestionPipeline pipeline = pipelineWithConfig(Map.of("type", "DatabaseMetadata")); + + assertDoesNotThrow(() -> IngestionPipelineRepository.validateSourceConfigHasType(pipeline)); + } + + @Test + void validateSourceConfigHasTypeAcceptsTypedConfigWithDefaultType() { + IngestionPipeline pipeline = pipelineWithConfig(new DatabaseServiceMetadataPipeline()); + + assertDoesNotThrow(() -> IngestionPipelineRepository.validateSourceConfigHasType(pipeline)); + } + + private static Stream invalidSourceConfigs() { + Map nullType = new HashMap<>(); + nullType.put("type", null); + + return Stream.of( + Arguments.of( + "missing sourceConfig", + new IngestionPipeline(), + "sourceConfig.config.type is required"), + Arguments.of( + "missing config", + new IngestionPipeline().withSourceConfig(new SourceConfig()), + "sourceConfig.config.type is required"), + Arguments.of( + "empty config", pipelineWithConfig(Map.of()), "sourceConfig.config.type is required"), + Arguments.of( + "null type", pipelineWithConfig(nullType), "sourceConfig.config.type is required"), + Arguments.of( + "empty type", + pipelineWithConfig(Map.of("type", "")), + "sourceConfig.config.type is required"), + Arguments.of( + "blank type", + pipelineWithConfig(Map.of("type", " ")), + "sourceConfig.config.type is required"), + Arguments.of( + "non-string type", + pipelineWithConfig(Map.of("type", 42)), + "sourceConfig.config.type is required"), + Arguments.of( + "raw-map enum type", + pipelineWithConfig( + Map.of( + "type", + DatabaseServiceMetadataPipeline.DatabaseMetadataConfigType.DATABASE_METADATA)), + "sourceConfig.config.type is required"), + Arguments.of( + "scalar config", + pipelineWithConfig("DatabaseMetadata"), + "sourceConfig.config must be an object with type"), + Arguments.of( + "list config", + pipelineWithConfig(List.of()), + "sourceConfig.config must be an object with type")); + } + private static IngestionPipeline createPipelineWithSchedule(String schedule) { IngestionPipeline pipeline = createBasicPipeline(); AirflowConfig airflowConfig = new AirflowConfig(); @@ -388,4 +468,8 @@ private static IngestionPipeline createBasicPipeline() { return pipeline; } + + private static IngestionPipeline pipelineWithConfig(Object config) { + return new IngestionPipeline().withSourceConfig(new SourceConfig().withConfig(config)); + } } diff --git a/openmetadata-ui/src/main/resources/ui/playwright/support/entity/PipelineClass.ts b/openmetadata-ui/src/main/resources/ui/playwright/support/entity/PipelineClass.ts index de4e8f96756e..400c47510400 100644 --- a/openmetadata-ui/src/main/resources/ui/playwright/support/entity/PipelineClass.ts +++ b/openmetadata-ui/src/main/resources/ui/playwright/support/entity/PipelineClass.ts @@ -180,7 +180,9 @@ export class PipelineClass extends EntityClass { type: 'pipelineService', }, sourceConfig: { - config: {}, + config: { + type: 'PipelineMetadata', + }, }, }, } From 438b56fc389db4e96f3f3d3abb09dd7aa53e6038 Mon Sep 17 00:00:00 2001 From: Ayush Shah Date: Fri, 21 Aug 2026 10:11:43 +0530 Subject: [PATCH 2/4] ingestion: repair legacy source config types --- ingestion/src/metadata/workflow/base.py | 16 +- .../tests/unit/workflow/test_base_workflow.py | 51 +++ .../jdbi3/IngestionPipelineRepository.java | 120 ++++++- .../IngestionPipelineRepositoryTest.java | 293 ++++++++++++++++++ 4 files changed, 466 insertions(+), 14 deletions(-) diff --git a/ingestion/src/metadata/workflow/base.py b/ingestion/src/metadata/workflow/base.py index a2ccd03de673..dac38f814501 100644 --- a/ingestion/src/metadata/workflow/base.py +++ b/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, ) diff --git a/ingestion/tests/unit/workflow/test_base_workflow.py b/ingestion/tests/unit/workflow/test_base_workflow.py index 804c844a7503..e1defab51fa1 100644 --- a/ingestion/tests/unit/workflow/test_base_workflow.py +++ b/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: diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java index e8a1b89ebb11..a58ab3b70c65 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java +++ b/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> 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 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"; @@ -514,28 +564,70 @@ public void prepare(IngestionPipeline ingestionPipeline, boolean update) { } 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 || configMap.get(SOURCE_CONFIG_TYPE) != null) { + return; + } + + String sourceConfigType = getLegacySourceConfigType(ingestionPipeline, configMap); + if (sourceConfigType == null) { + return; + } + + Map 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 Object getRequiredSourceConfig(IngestionPipeline ingestionPipeline) { if (ingestionPipeline.getSourceConfig() == null || ingestionPipeline.getSourceConfig().getConfig() == null) { - throw new BadRequestException("sourceConfig.config.type is required"); + throw new BadRequestException(SOURCE_CONFIG_TYPE_REQUIRED); } + return ingestionPipeline.getSourceConfig().getConfig(); + } - Object config = ingestionPipeline.getSourceConfig().getConfig(); - Object type; - boolean generatedConfig = !(config instanceof Map); + private static Map getSourceConfigMap(Object config) { try { - type = - generatedConfig ? JsonUtils.getMap(config).get("type") : ((Map) config).get("type"); + return config instanceof Map map ? map : JsonUtils.getMap(config); } catch (IllegalArgumentException e) { - throw new BadRequestException("sourceConfig.config must be an object with type"); + throw new BadRequestException(SOURCE_CONFIG_OBJECT_REQUIRED); } + } - if (type instanceof String typeValue && !typeValue.isBlank()) { - return; - } - if (generatedConfig && type instanceof Enum) { - return; + 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; } - throw new BadRequestException("sourceConfig.config.type is required"); + + Map 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) { @@ -1590,6 +1682,8 @@ public RestUtil.PutResponse addOperationMetrics( public PipelineServiceClientResponse deployIngestionPipeline( IngestionPipeline ingestionPipeline, ServiceEntityInterface service) { + repairLegacySourceConfig(ingestionPipeline); + validateSourceConfigHasType(ingestionPipeline); applyStreamableLogsConfig(ingestionPipeline); return pipelineServiceClient.deployPipeline(ingestionPipeline, service); } diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java index 6fa837746f57..9b019160159a 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java @@ -7,9 +7,12 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doCallRealMethod; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import java.util.HashMap; @@ -25,8 +28,11 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.mockito.Mockito; +import org.openmetadata.schema.ServiceEntityInterface; import org.openmetadata.schema.entity.services.ingestionPipelines.AirflowConfig; import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline; +import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineServiceClientResponse; +import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineType; import org.openmetadata.schema.metadataIngestion.DatabaseServiceMetadataPipeline; import org.openmetadata.schema.metadataIngestion.LogLevels; import org.openmetadata.schema.metadataIngestion.SourceConfig; @@ -36,6 +42,7 @@ import org.openmetadata.sdk.PipelineServiceClientInterface; import org.openmetadata.sdk.exception.IngestionRunnerUnavailableException; import org.openmetadata.sdk.exception.PipelineServiceClientException; +import org.openmetadata.service.Entity; import org.openmetadata.service.exception.BadRequestException; import org.openmetadata.service.secrets.SecretsManagerFactory; @@ -130,6 +137,273 @@ private IngestionPipelineRepository repositoryWithClient( return cleanupRepository; } + @ParameterizedTest(name = "{0}") + @MethodSource("legacySourceConfigTypes") + void deployLegacyPipelineAddsDefaultSourceConfigTypeBeforeCallingRunner( + String testCase, PipelineType pipelineType, String serviceType, String expectedConfigType) { + IngestionPipeline pipeline = legacyPipeline(pipelineType, serviceType); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + PipelineServiceClientResponse response = new PipelineServiceClientResponse().withCode(200); + when(pipelineServiceClient.deployPipeline( + any(IngestionPipeline.class), any(ServiceEntityInterface.class))) + .thenReturn(response); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + ServiceEntityInterface service = mock(ServiceEntityInterface.class); + + PipelineServiceClientResponse actual = + deploymentRepository.deployIngestionPipeline(pipeline, service); + + assertEquals(response, actual); + assertEquals(expectedConfigType, sourceConfigMap(pipeline).get("type")); + assertEquals("preserved", sourceConfigMap(pipeline).get("existingSetting")); + verify(pipelineServiceClient).deployPipeline(pipeline, service); + } + + @Test + void deployLegacyPipelineRejectsUnknownSourceConfigTypeBeforeCallingRunner() { + IngestionPipeline pipeline = legacyPipeline(PipelineType.USAGE, Entity.DASHBOARD_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> + deploymentRepository.deployIngestionPipeline( + pipeline, mock(ServiceEntityInterface.class))); + + assertEquals("sourceConfig.config.type is required", exception.getMessage()); + verifyNoInteractions(pipelineServiceClient); + } + + @Test + void deployPipelineRejectsMalformedSourceConfigBeforeCallingRunner() { + IngestionPipeline pipeline = legacyPipeline(PipelineType.METADATA, Entity.DATABASE_SERVICE); + pipeline.getSourceConfig().setConfig("not-an-object"); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> + deploymentRepository.deployIngestionPipeline( + pipeline, mock(ServiceEntityInterface.class))); + + assertEquals("sourceConfig.config must be an object with type", exception.getMessage()); + verifyNoInteractions(pipelineServiceClient); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("invalidLegacySourceConfigs") + void deployLegacyPipelineRejectsInvalidExplicitSourceConfigTypeBeforeCallingRunner( + String testCase, Object config) { + IngestionPipeline pipeline = + legacyPipelineWithConfig(config, PipelineType.METADATA, Entity.DATABASE_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> + deploymentRepository.deployIngestionPipeline( + pipeline, mock(ServiceEntityInterface.class))); + + assertEquals("sourceConfig.config.type is required", exception.getMessage()); + verifyNoInteractions(pipelineServiceClient); + } + + @Test + void deployNewPipelineWithoutSourceConfigTypeRejectsBeforeCallingRunner() { + IngestionPipeline pipeline = legacyPipeline(PipelineType.METADATA, Entity.DATABASE_SERVICE); + pipeline.setId(null); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> + deploymentRepository.deployIngestionPipeline( + pipeline, mock(ServiceEntityInterface.class))); + + assertEquals("sourceConfig.config.type is required", exception.getMessage()); + verifyNoInteractions(pipelineServiceClient); + } + + @Test + void deployLegacyPipelineWithNullSourceConfigTypeAddsDefaultBeforeCallingRunner() { + Map config = new HashMap<>(); + config.put("type", null); + IngestionPipeline pipeline = + legacyPipelineWithConfig(config, PipelineType.METADATA, Entity.DATABASE_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + PipelineServiceClientResponse response = new PipelineServiceClientResponse().withCode(200); + when(pipelineServiceClient.deployPipeline( + any(IngestionPipeline.class), any(ServiceEntityInterface.class))) + .thenReturn(response); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + ServiceEntityInterface service = mock(ServiceEntityInterface.class); + + PipelineServiceClientResponse actual = + deploymentRepository.deployIngestionPipeline(pipeline, service); + + assertEquals(response, actual); + assertEquals("DatabaseMetadata", sourceConfigMap(pipeline).get("type")); + verify(pipelineServiceClient).deployPipeline(pipeline, service); + } + + @Test + void deployLegacyReverseIngestionAddsDefaultSourceConfigTypeBeforeCallingRunner() { + Map config = new HashMap<>(); + config.put("operations", List.of()); + IngestionPipeline pipeline = + legacyPipelineWithConfig(config, PipelineType.METADATA, Entity.DATABASE_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + PipelineServiceClientResponse response = new PipelineServiceClientResponse().withCode(200); + when(pipelineServiceClient.deployPipeline( + any(IngestionPipeline.class), any(ServiceEntityInterface.class))) + .thenReturn(response); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + ServiceEntityInterface service = mock(ServiceEntityInterface.class); + + PipelineServiceClientResponse actual = + deploymentRepository.deployIngestionPipeline(pipeline, service); + + assertEquals(response, actual); + assertEquals("ReverseIngestion", sourceConfigMap(pipeline).get("type")); + verify(pipelineServiceClient).deployPipeline(pipeline, service); + } + + @Test + void deployPipelinePreservesExplicitSourceConfigType() { + IngestionPipeline pipeline = + legacyPipelineWithConfig( + new HashMap<>(Map.of("type", "DatabaseMetadata")), + PipelineType.METADATA, + Entity.DATABASE_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + PipelineServiceClientResponse response = new PipelineServiceClientResponse().withCode(200); + when(pipelineServiceClient.deployPipeline( + any(IngestionPipeline.class), any(ServiceEntityInterface.class))) + .thenReturn(response); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + ServiceEntityInterface service = mock(ServiceEntityInterface.class); + + PipelineServiceClientResponse actual = + deploymentRepository.deployIngestionPipeline(pipeline, service); + + assertEquals(response, actual); + assertEquals("DatabaseMetadata", sourceConfigMap(pipeline).get("type")); + verify(pipelineServiceClient).deployPipeline(pipeline, service); + } + + private static Stream legacySourceConfigTypes() { + return Stream.of( + Arguments.of( + "database metadata", + PipelineType.METADATA, + Entity.DATABASE_SERVICE, + "DatabaseMetadata"), + Arguments.of( + "dashboard metadata", + PipelineType.METADATA, + Entity.DASHBOARD_SERVICE, + "DashboardMetadata"), + Arguments.of( + "messaging metadata", + PipelineType.METADATA, + Entity.MESSAGING_SERVICE, + "MessagingMetadata"), + Arguments.of( + "pipeline metadata", + PipelineType.METADATA, + Entity.PIPELINE_SERVICE, + "PipelineMetadata"), + Arguments.of( + "machine-learning metadata", + PipelineType.METADATA, + Entity.MLMODEL_SERVICE, + "MlModelMetadata"), + Arguments.of( + "storage metadata", PipelineType.METADATA, Entity.STORAGE_SERVICE, "StorageMetadata"), + Arguments.of( + "drive metadata", PipelineType.METADATA, Entity.DRIVE_SERVICE, "DriveMetadata"), + Arguments.of( + "search metadata", PipelineType.METADATA, Entity.SEARCH_SERVICE, "SearchMetadata"), + Arguments.of("api metadata", PipelineType.METADATA, Entity.API_SERVICE, "ApiMetadata"), + Arguments.of("mcp metadata", PipelineType.METADATA, Entity.MCP_SERVICE, "McpMetadata"), + Arguments.of( + "security metadata", + PipelineType.METADATA, + Entity.SECURITY_SERVICE, + "SecurityMetadata"), + Arguments.of( + "OpenMetadata service metadata", + PipelineType.METADATA, + Entity.METADATA_SERVICE, + "DatabaseMetadata"), + Arguments.of( + "database usage", PipelineType.USAGE, Entity.DATABASE_SERVICE, "DatabaseUsage"), + Arguments.of( + "database lineage", PipelineType.LINEAGE, Entity.DATABASE_SERVICE, "DatabaseLineage"), + Arguments.of( + "dashboard lineage", + PipelineType.LINEAGE, + Entity.DASHBOARD_SERVICE, + "DashboardMetadata"), + Arguments.of( + "database profiler", PipelineType.PROFILER, Entity.DATABASE_SERVICE, "Profiler"), + Arguments.of( + "database auto classification", + PipelineType.AUTO_CLASSIFICATION, + Entity.DATABASE_SERVICE, + "AutoClassification"), + Arguments.of( + "messaging auto classification", + PipelineType.AUTO_CLASSIFICATION, + Entity.MESSAGING_SERVICE, + "AutoClassification"), + Arguments.of( + "storage auto classification", + PipelineType.AUTO_CLASSIFICATION, + Entity.STORAGE_SERVICE, + "AutoClassification"), + Arguments.of("dbt", PipelineType.DBT, Entity.DATABASE_SERVICE, "DBT"), + Arguments.of("test suite", PipelineType.TEST_SUITE, Entity.TEST_SUITE, "TestSuite"), + Arguments.of( + "data insight", PipelineType.DATA_INSIGHT, Entity.METADATA_SERVICE, "dataInsight"), + Arguments.of( + "search reindex", + PipelineType.ELASTIC_SEARCH_REINDEX, + Entity.METADATA_SERVICE, + "MetadataToElasticSearch"), + Arguments.of( + "application", PipelineType.APPLICATION, Entity.METADATA_SERVICE, "Application"), + Arguments.of( + "policy agent", PipelineType.POLICY_AGENT, Entity.DATABASE_SERVICE, "PolicyAgent")); + } + + private static Stream invalidLegacySourceConfigs() { + return Stream.of( + Arguments.of("blank type", Map.of("type", " ")), + Arguments.of("non-string type", Map.of("type", 42)), + Arguments.of( + "raw-map enum type", + Map.of( + "type", + DatabaseServiceMetadataPipeline.DatabaseMetadataConfigType.DATABASE_METADATA))); + } + @Test @DisplayName("requiresRedeployment should detect schedule changes from Scheduled to On-Demand") void testRequiresRedeployment_ScheduleToOnDemand_ShouldReturnTrue() { @@ -469,6 +743,25 @@ private static IngestionPipeline createBasicPipeline() { return pipeline; } + private static IngestionPipeline legacyPipeline(PipelineType pipelineType, String serviceType) { + return legacyPipelineWithConfig( + new HashMap<>(Map.of("existingSetting", "preserved")), pipelineType, serviceType); + } + + private static IngestionPipeline legacyPipelineWithConfig( + Object config, PipelineType pipelineType, String serviceType) { + IngestionPipeline pipeline = pipelineWithConfig(config); + pipeline.setId(UUID.randomUUID()); + pipeline.setName("legacy-pipeline"); + pipeline.setPipelineType(pipelineType); + pipeline.setService(new EntityReference().withName("legacy-service").withType(serviceType)); + return pipeline; + } + + private static Map sourceConfigMap(IngestionPipeline pipeline) { + return (Map) pipeline.getSourceConfig().getConfig(); + } + private static IngestionPipeline pipelineWithConfig(Object config) { return new IngestionPipeline().withSourceConfig(new SourceConfig().withConfig(config)); } From 488cc28b0b09c82a3799103717bf3e982a7d3f11 Mon Sep 17 00:00:00 2001 From: Ayush Shah Date: Fri, 21 Aug 2026 12:07:24 +0530 Subject: [PATCH 3/4] fix: deduplicate initial ingestion progress event Preserve listener-first registration while preventing a raced snapshot from replaying a live update. --- .../jdbi3/IngestionPipelineRepository.java | 39 ++- .../IngestionPipelineProgressStreamTest.java | 242 ++++++++++++++++++ 2 files changed, 279 insertions(+), 2 deletions(-) create mode 100644 openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java index a58ab3b70c65..ff541cf6f909 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java @@ -1593,20 +1593,55 @@ public boolean isProgressTrackingEnabled() { } public void streamProgress(String pipelineFQN, UUID runId, SseEventSink eventSink, Sse sse) { - Consumer listener = update -> emitProgressUpdate(eventSink, sse, update); + ProgressStreamEventEmitter emitter = new ProgressStreamEventEmitter(eventSink, sse); + Consumer 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)) { diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java new file mode 100644 index 000000000000..3e5eb9daca6c --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java @@ -0,0 +1,242 @@ +/* + * Copyright 2026 Collate. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.jdbi3; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import jakarta.ws.rs.sse.OutboundSseEvent; +import jakarta.ws.rs.sse.Sse; +import jakarta.ws.rs.sse.SseEventSink; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.function.Consumer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressUpdate; +import org.openmetadata.schema.entity.services.ingestionPipelines.ProgressUpdateType; +import org.openmetadata.service.monitoring.IngestionProgressTracker; +import org.openmetadata.service.monitoring.IngestionProgressTracker.ProgressState; +import org.openmetadata.service.resources.services.ingestionpipelines.ProgressSseManager; + +class IngestionPipelineProgressStreamTest { + + private final List registeredSinks = new ArrayList<>(); + + @AfterEach + void closeProgressStreams() { + registeredSinks.forEach(ProgressSseManager.getInstance()::close); + } + + @Test + void streamProgressEmitsExistingSnapshotThenSubsequentLiveUpdate() { + String pipelineFqn = "service.pipeline"; + UUID runId = UUID.randomUUID(); + IngestionProgressTracker tracker = new IngestionProgressTracker(new SimpleMeterRegistry()); + ProgressUpdate snapshot = update(runId, ProgressUpdateType.DISCOVERY, "snapshot"); + ProgressUpdate liveUpdate = update(runId, ProgressUpdateType.PROCESSING, "live"); + tracker.updateProgress(pipelineFqn, runId, snapshot); + CapturingSink sink = capturingSink(); + + repositoryWith(tracker).streamProgress(pipelineFqn, runId, sink, sse()); + tracker.updateProgress(pipelineFqn, runId, liveUpdate); + + assertMessages(sink, "snapshot", "live"); + } + + @Test + void streamProgressDoesNotReplayLiveUpdateDeliveredDuringListenerRegistration() { + String pipelineFqn = "service.pipeline"; + UUID runId = UUID.randomUUID(); + ProgressUpdate liveUpdate = update(runId, ProgressUpdateType.PROCESSING, "live"); + IngestionProgressTracker tracker = + new UpdateDuringListenerRegistrationTracker(runId, liveUpdate); + CapturingSink sink = capturingSink(); + + repositoryWith(tracker).streamProgress(pipelineFqn, runId, sink, sse()); + + assertMessages(sink, "live"); + } + + @Test + void streamProgressDoesNotEmitStaleSnapshotAfterLiveTerminalUpdate() { + String pipelineFqn = "service.pipeline"; + UUID runId = UUID.randomUUID(); + ProgressUpdate snapshot = update(runId, ProgressUpdateType.PROCESSING, "snapshot"); + ProgressUpdate terminalUpdate = update(runId, ProgressUpdateType.PIPELINE_COMPLETE, "terminal"); + IngestionProgressTracker tracker = + new UpdateDuringSnapshotReadTracker(runId, snapshot, terminalUpdate); + DelayedFirstSendSink sink = delayedFirstSendSink(); + + repositoryWith(tracker).streamProgress(pipelineFqn, runId, sink, sse()); + + assertMessages(sink, "terminal"); + assertFalse(sink.isClosed()); + + sink.completeFirstSend(); + + assertTrue(sink.isClosed()); + } + + private IngestionPipelineRepository repositoryWith(IngestionProgressTracker tracker) { + IngestionPipelineRepository repository = + mock(IngestionPipelineRepository.class, Mockito.CALLS_REAL_METHODS); + repository.setProgressTracker(tracker); + return repository; + } + + private CapturingSink capturingSink() { + CapturingSink sink = new CapturingSink(); + registeredSinks.add(sink); + return sink; + } + + private DelayedFirstSendSink delayedFirstSendSink() { + DelayedFirstSendSink sink = new DelayedFirstSendSink(); + registeredSinks.add(sink); + return sink; + } + + private Sse sse() { + Sse sse = mock(Sse.class); + when(sse.newEvent(anyString())) + .thenAnswer( + invocation -> { + OutboundSseEvent event = mock(OutboundSseEvent.class); + when(event.getData()).thenReturn(invocation.getArgument(0)); + return event; + }); + return sse; + } + + private static ProgressUpdate update(UUID runId, ProgressUpdateType updateType, String message) { + return new ProgressUpdate() + .withRunId(runId.toString()) + .withTimestamp(System.currentTimeMillis()) + .withUpdateType(updateType) + .withMessage(message); + } + + private static void assertMessages(CapturingSink sink, String... messages) { + assertEquals(messages.length, sink.payloads.size()); + for (int i = 0; i < messages.length; i++) { + assertTrue(sink.payloads.get(i).contains(messages[i])); + } + } + + private static class CapturingSink implements SseEventSink { + protected final List payloads = new ArrayList<>(); + private boolean closed; + + @Override + public boolean isClosed() { + return closed; + } + + @Override + public CompletionStage send(OutboundSseEvent event) { + capture(event); + return CompletableFuture.completedFuture(null); + } + + @Override + public void close() { + closed = true; + } + + protected void capture(OutboundSseEvent event) { + if (event.getData() != null) { + payloads.add((String) event.getData()); + } + } + } + + private static final class DelayedFirstSendSink extends CapturingSink { + private final CompletableFuture firstSend = new CompletableFuture<>(); + private boolean awaitingFirstSend = true; + + @Override + public CompletionStage send(OutboundSseEvent event) { + capture(event); + if (awaitingFirstSend) { + awaitingFirstSend = false; + return firstSend; + } + return CompletableFuture.completedFuture(null); + } + + void completeFirstSend() { + firstSend.complete(null); + } + } + + private static final class UpdateDuringListenerRegistrationTracker + extends IngestionProgressTracker { + private final UUID expectedRunId; + private final ProgressUpdate liveUpdate; + private boolean updateSent; + + private UpdateDuringListenerRegistrationTracker(UUID expectedRunId, ProgressUpdate liveUpdate) { + super(new SimpleMeterRegistry()); + this.expectedRunId = expectedRunId; + this.liveUpdate = liveUpdate; + } + + @Override + public void registerProgressListener( + String pipelineFqn, UUID runId, Consumer listener) { + super.registerProgressListener(pipelineFqn, runId, listener); + if (!updateSent && expectedRunId.equals(runId)) { + updateSent = true; + updateProgress(pipelineFqn, runId, liveUpdate); + } + } + } + + private static final class UpdateDuringSnapshotReadTracker extends IngestionProgressTracker { + private final UUID expectedRunId; + private final ProgressUpdate snapshot; + private final ProgressUpdate liveUpdate; + private boolean snapshotRead; + + private UpdateDuringSnapshotReadTracker( + UUID expectedRunId, ProgressUpdate snapshot, ProgressUpdate liveUpdate) { + super(new SimpleMeterRegistry()); + this.expectedRunId = expectedRunId; + this.snapshot = snapshot; + this.liveUpdate = liveUpdate; + } + + @Override + public ProgressState getProgressState(String pipelineFqn, UUID runId) { + if (!snapshotRead && expectedRunId.equals(runId)) { + snapshotRead = true; + updateProgress(pipelineFqn, runId, liveUpdate); + ProgressState frozenSnapshot = new ProgressState(); + frozenSnapshot.applyUpdate(snapshot); + return frozenSnapshot; + } + return super.getProgressState(pipelineFqn, runId); + } + } +} From 1d0f34da5fd89cd6fd8dcd03920717578f8f7ee7 Mon Sep 17 00:00:00 2001 From: Ayush Shah Date: Fri, 21 Aug 2026 13:00:41 +0530 Subject: [PATCH 4/4] ingestion: cover legacy repair and progress deduplication --- .../jdbi3/IngestionPipelineRepository.java | 7 +++- .../IngestionPipelineProgressStreamTest.java | 34 +++++++++++++++++++ .../IngestionPipelineRepositoryTest.java | 24 ++++++++++++- 3 files changed, 63 insertions(+), 2 deletions(-) diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java index ff541cf6f909..f2fb213ad99a 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java @@ -574,7 +574,7 @@ static void validateSourceConfigHasType(IngestionPipeline ingestionPipeline) { private static void repairLegacySourceConfig(IngestionPipeline ingestionPipeline) { Object config = getRequiredSourceConfig(ingestionPipeline); Map configMap = getSourceConfigMap(config); - if (ingestionPipeline.getId() == null || configMap.get(SOURCE_CONFIG_TYPE) != null) { + if (ingestionPipeline.getId() == null || !isMissingOrBlankSourceConfigType(configMap)) { return; } @@ -589,6 +589,11 @@ private static void repairLegacySourceConfig(IngestionPipeline ingestionPipeline 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) { diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java index 3e5eb9daca6c..aaff97969138 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineProgressStreamTest.java @@ -78,6 +78,21 @@ void streamProgressDoesNotReplayLiveUpdateDeliveredDuringListenerRegistration() assertMessages(sink, "live"); } + @Test + void streamProgressDoesNotReplaySnapshotWhenDelayedListenerReceivesSameInstance() { + String pipelineFqn = "service.pipeline"; + UUID runId = UUID.randomUUID(); + ProgressUpdate snapshot = update(runId, ProgressUpdateType.DISCOVERY, "snapshot"); + DelayedSnapshotListenerTracker tracker = new DelayedSnapshotListenerTracker(); + tracker.updateProgress(pipelineFqn, runId, snapshot); + CapturingSink sink = capturingSink(); + + repositoryWith(tracker).streamProgress(pipelineFqn, runId, sink, sse()); + tracker.deliverDelayedCallback(snapshot); + + assertMessages(sink, "snapshot"); + } + @Test void streamProgressDoesNotEmitStaleSnapshotAfterLiveTerminalUpdate() { String pipelineFqn = "service.pipeline"; @@ -239,4 +254,23 @@ public ProgressState getProgressState(String pipelineFqn, UUID runId) { return super.getProgressState(pipelineFqn, runId); } } + + private static final class DelayedSnapshotListenerTracker extends IngestionProgressTracker { + private Consumer listener; + + private DelayedSnapshotListenerTracker() { + super(new SimpleMeterRegistry()); + } + + @Override + public void registerProgressListener( + String pipelineFqn, UUID runId, Consumer registeredListener) { + super.registerProgressListener(pipelineFqn, runId, registeredListener); + listener = registeredListener; + } + + void deliverDelayedCallback(ProgressUpdate snapshot) { + listener.accept(snapshot); + } + } } diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java index 9b019160159a..c8e1475ebe5c 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java @@ -260,6 +260,29 @@ void deployLegacyPipelineWithNullSourceConfigTypeAddsDefaultBeforeCallingRunner( verify(pipelineServiceClient).deployPipeline(pipeline, service); } + @Test + void deployLegacyPipelineWithBlankSourceConfigTypeAddsDefaultBeforeCallingRunner() { + Map config = new HashMap<>(); + config.put("type", " "); + IngestionPipeline pipeline = + legacyPipelineWithConfig(config, PipelineType.METADATA, Entity.DATABASE_SERVICE); + PipelineServiceClientInterface pipelineServiceClient = + mock(PipelineServiceClientInterface.class); + PipelineServiceClientResponse response = new PipelineServiceClientResponse().withCode(200); + when(pipelineServiceClient.deployPipeline( + any(IngestionPipeline.class), any(ServiceEntityInterface.class))) + .thenReturn(response); + IngestionPipelineRepository deploymentRepository = repositoryWithClient(pipelineServiceClient); + ServiceEntityInterface service = mock(ServiceEntityInterface.class); + + PipelineServiceClientResponse actual = + deploymentRepository.deployIngestionPipeline(pipeline, service); + + assertEquals(response, actual); + assertEquals("DatabaseMetadata", sourceConfigMap(pipeline).get("type")); + verify(pipelineServiceClient).deployPipeline(pipeline, service); + } + @Test void deployLegacyReverseIngestionAddsDefaultSourceConfigTypeBeforeCallingRunner() { Map config = new HashMap<>(); @@ -395,7 +418,6 @@ private static Stream legacySourceConfigTypes() { private static Stream invalidLegacySourceConfigs() { return Stream.of( - Arguments.of("blank type", Map.of("type", " ")), Arguments.of("non-string type", Map.of("type", 42)), Arguments.of( "raw-map enum type",