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 0b71b58a56ea..f8c4a16762c4 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 @@ -74,6 +74,7 @@ import org.openmetadata.service.Entity; import org.openmetadata.service.OpenMetadataApplicationConfig; import org.openmetadata.service.events.lifecycle.EntityLifecycleEventDispatcher; +import org.openmetadata.service.exception.BadRequestException; import org.openmetadata.service.exception.EntityNotFoundException; import org.openmetadata.service.logstorage.LogStorageInterface; import org.openmetadata.service.logstorage.S3LogStorage.LogStreamListener; @@ -298,6 +299,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 a3bc02f8d3e4..cc88da2fd9d0 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 @@ -5,16 +5,24 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; 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.Mockito.doCallRealMethod; 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.openmetadata.schema.entity.services.ingestionPipelines.AirflowConfig; import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline; import org.openmetadata.schema.metadataIngestion.DatabaseServiceMetadataPipeline; @@ -23,6 +31,7 @@ import org.openmetadata.schema.security.secrets.SecretsManagerConfiguration; import org.openmetadata.schema.security.secrets.SecretsManagerProvider; import org.openmetadata.schema.type.EntityReference; +import org.openmetadata.service.exception.BadRequestException; import org.openmetadata.service.secrets.SecretsManagerFactory; class IngestionPipelineRepositoryTest { @@ -268,6 +277,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(); @@ -312,4 +393,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 211063e3a930..e47b216ae129 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 @@ -171,7 +171,9 @@ export class PipelineClass extends EntityClass { type: 'pipelineService', }, sourceConfig: { - config: {}, + config: { + type: 'PipelineMetadata', + }, }, }, }