This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch add-cdc-connectors in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit cd714253b2e6771e1ed8b057d59739148af91f95 Author: Dominik Riemer <[email protected]> AuthorDate: Thu May 7 14:55:12 2026 +0200 feat: Add MSSQL CDC connector --- streampipes-extensions/pom.xml | 1 + .../streampipes-connectors-cdc/pom.xml | 80 ++++++ .../connectors/cdc/CdcConnectorsModuleExport.java | 46 +++ .../cdc/adapter/mssql/MsSqlCdcAdapter.java | 312 +++++++++++++++++++++ .../cdc/adapter/mssql/MsSqlCdcAdapterConfig.java | 143 ++++++++++ .../cdc/adapter/mssql/MsSqlMetadataClient.java | 266 ++++++++++++++++++ .../strings.en | 47 ++++ .../streampipes-extensions-all-iiot/pom.xml | 6 + .../extensions/all/iiot/AllExtensionsIIoTInit.java | 4 +- .../streampipes-extensions-all-jvm/pom.xml | 5 + .../extensions/all/jvm/AllExtensionsInit.java | 2 + ...tic-runtime-resolvable-oneof-input.component.ts | 1 + 12 files changed, 912 insertions(+), 1 deletion(-) diff --git a/streampipes-extensions/pom.xml b/streampipes-extensions/pom.xml index d540061ea2..2916a15a79 100644 --- a/streampipes-extensions/pom.xml +++ b/streampipes-extensions/pom.xml @@ -35,6 +35,7 @@ <module>streampipes-connect-adapters-iiot</module> <module>streampipes-connectors-camel</module> + <module>streampipes-connectors-cdc</module> <module>streampipes-connectors-influx</module> <module>streampipes-connectors-kafka</module> <module>streampipes-connectors-mqtt</module> diff --git a/streampipes-extensions/streampipes-connectors-cdc/pom.xml b/streampipes-extensions/streampipes-connectors-cdc/pom.xml new file mode 100644 index 0000000000..487a0ac203 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/pom.xml @@ -0,0 +1,80 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You 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. + ~ + --> + +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-extensions</artifactId> + <version>0.99.0-SNAPSHOT</version> + </parent> + + <artifactId>streampipes-connectors-cdc</artifactId> + + <properties> + <debezium.version>3.4.3.Final</debezium.version> + </properties> + + <dependencies> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-extensions-api</artifactId> + <version>0.99.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-extensions-management</artifactId> + <version>0.99.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-sdk</artifactId> + <version>0.99.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>io.debezium</groupId> + <artifactId>debezium-api</artifactId> + <version>${debezium.version}</version> + </dependency> + <dependency> + <groupId>io.debezium</groupId> + <artifactId>debezium-embedded</artifactId> + <version>${debezium.version}</version> + <exclusions> + <exclusion> + <groupId>io.github.classgraph</groupId> + <artifactId>classgraph</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>io.debezium</groupId> + <artifactId>debezium-connector-sqlserver</artifactId> + <version>${debezium.version}</version> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </dependency> + </dependencies> + +</project> diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/CdcConnectorsModuleExport.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/CdcConnectorsModuleExport.java new file mode 100644 index 0000000000..604eb97105 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/CdcConnectorsModuleExport.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.streampipes.extensions.connectors.cdc; + +import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter; +import org.apache.streampipes.extensions.api.declarer.IExtensionModuleExport; +import org.apache.streampipes.extensions.api.migration.IModelMigrator; +import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement; +import org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapter; + +import java.util.Collections; +import java.util.List; + +public class CdcConnectorsModuleExport implements IExtensionModuleExport { + + @Override + public List<StreamPipesAdapter> adapters() { + return List.of(new MsSqlCdcAdapter()); + } + + @Override + public List<IStreamPipesPipelineElement<?>> pipelineElements() { + return Collections.emptyList(); + } + + @Override + public List<IModelMigrator<?, ?>> migrators() { + return Collections.emptyList(); + } +} diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapter.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapter.java new file mode 100644 index 0000000000..c0051721c0 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapter.java @@ -0,0 +1,312 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.streampipes.extensions.connectors.cdc.adapter.mssql; + +import org.apache.streampipes.commons.exceptions.SpConfigurationException; +import org.apache.streampipes.commons.exceptions.connect.AdapterException; +import org.apache.streampipes.extensions.api.connect.IAdapterConfiguration; +import org.apache.streampipes.extensions.api.connect.IEventCollector; +import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter; +import org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchemaContext; +import org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeContext; +import org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor; +import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor; +import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig; +import org.apache.streampipes.model.connect.guess.SampleData; +import org.apache.streampipes.model.staticproperty.RuntimeResolvableOneOfStaticProperty; +import org.apache.streampipes.model.staticproperty.StaticProperty; +import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder; +import org.apache.streampipes.sdk.builder.adapter.SampleDataBuilder; +import org.apache.streampipes.sdk.helpers.Labels; +import org.apache.streampipes.sdk.helpers.Locales; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.debezium.engine.ChangeEvent; +import io.debezium.engine.DebeziumEngine; +import io.debezium.engine.format.Json; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.format.DateTimeParseException; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.DATABASE_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.ENCRYPT_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.HOST_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.PASSWORD_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.PORT_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.TABLE_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.TIMEZONE_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.TRUST_SERVER_CERTIFICATE_KEY; +import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.MsSqlCdcAdapterConfig.USERNAME_KEY; + +public class MsSqlCdcAdapter implements StreamPipesAdapter, SupportsRuntimeConfig { + + private static final Logger LOG = LoggerFactory.getLogger(MsSqlCdcAdapter.class); + + public static final String ID = "org.apache.streampipes.connect.cdc.adapter.mssql"; + + private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() { + }; + + private final ObjectMapper mapper; + private final MsSqlMetadataClient metadataClient; + + private DebeziumEngine<ChangeEvent<String, String>> engine; + private ExecutorService executorService; + private Map<String, MsSqlMetadataClient.TemporalColumnMode> temporalColumns; + private ZoneId configuredZoneId; + + public MsSqlCdcAdapter() { + this(new ObjectMapper(), new MsSqlMetadataClient()); + } + + MsSqlCdcAdapter(ObjectMapper mapper, MsSqlMetadataClient metadataClient) { + this.mapper = mapper; + this.metadataClient = metadataClient; + } + + @Override + public StaticProperty resolveConfiguration(String staticPropertyInternalName, + IStaticPropertyExtractor extractor) throws SpConfigurationException { + if (!TABLE_KEY.equals(staticPropertyInternalName)) { + return null; + } + + MsSqlCdcAdapterConfig config = MsSqlCdcAdapterConfig.from(extractor, false); + RuntimeResolvableOneOfStaticProperty tableProperty = + extractor.getStaticPropertyByName(TABLE_KEY, RuntimeResolvableOneOfStaticProperty.class); + tableProperty.setOptions(metadataClient.discoverTables(config)); + return tableProperty; + } + + @Override + public IAdapterConfiguration declareConfig() { + return AdapterConfigurationBuilder.create(ID, 0, MsSqlCdcAdapter::new) + .withLocales(Locales.EN) + .requiredTextParameter(Labels.withId(HOST_KEY)) + .requiredIntegerParameter(Labels.withId(PORT_KEY), 1433) + .requiredTextParameter(Labels.withId(DATABASE_KEY)) + .requiredTextParameter(Labels.withId(USERNAME_KEY)) + .requiredSecret(Labels.withId(PASSWORD_KEY)) + .requiredSlideToggle(Labels.withId(ENCRYPT_KEY), true) + .requiredSlideToggle(Labels.withId(TRUST_SERVER_CERTIFICATE_KEY), false) + .requiredTextParameter(Labels.withId(TIMEZONE_KEY), ZoneId.systemDefault().getId()) + .requiredSingleValueSelectionFromContainer( + Labels.withId(TABLE_KEY), + java.util.List.of(HOST_KEY, PORT_KEY, DATABASE_KEY, USERNAME_KEY, PASSWORD_KEY, ENCRYPT_KEY, + TIMEZONE_KEY, + TRUST_SERVER_CERTIFICATE_KEY) + ) + .buildConfiguration(); + } + + @Override + public void onAdapterStarted(IAdapterParameterExtractor extractor, + IEventCollector collector, + IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException { + MsSqlCdcAdapterConfig config = MsSqlCdcAdapterConfig.from(extractor.getStaticPropertyExtractor()); + metadataClient.validateSelectedTable(config); + this.temporalColumns = metadataClient.describeTemporalColumns(config); + this.configuredZoneId = config.getZoneId(); + + Properties properties = buildDebeziumProperties(config); + this.engine = DebeziumEngine.create(Json.class) + .using(properties) + .notifying(record -> processRecord(record, collector)) + .build(); + + this.executorService = Executors.newSingleThreadExecutor(); + this.executorService.execute(engine); + } + + @Override + public void onAdapterStopped(IAdapterParameterExtractor extractor, + IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException { + if (engine != null) { + try { + engine.close(); + } catch (IOException e) { + throw new AdapterException("Failed to stop Debezium engine: " + e.getMessage(), e); + } + } + + if (executorService != null) { + executorService.shutdown(); + try { + if (!executorService.awaitTermination(15, TimeUnit.SECONDS)) { + LOG.warn("Debezium engine executor did not shut down within 15 seconds."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AdapterException("Interrupted while waiting for Debezium engine shutdown.", e); + } + } + + this.engine = null; + this.executorService = null; + this.temporalColumns = null; + this.configuredZoneId = null; + } + + @Override + public SampleData onSampleDataRequested(IAdapterParameterExtractor extractor, + IAdapterGuessSchemaContext adapterGuessSchemaContext) throws AdapterException { + MsSqlCdcAdapterConfig config = MsSqlCdcAdapterConfig.from(extractor.getStaticPropertyExtractor()); + metadataClient.validateSelectedTable(config); + return SampleDataBuilder.create() + .sample(metadataClient.generateSample(config)) + .build(); + } + + private Properties buildDebeziumProperties(MsSqlCdcAdapterConfig config) { + Properties props = new Properties(); + String logicalName = createLogicalName(config); + props.setProperty("name", logicalName + "-engine"); + props.setProperty("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector"); + props.setProperty("offset.storage", "org.apache.kafka.connect.storage.MemoryOffsetBackingStore"); + props.setProperty("schema.history.internal", "io.debezium.relational.history.MemorySchemaHistory"); + props.setProperty("converter.schemas.enable", "false"); + props.setProperty("key.converter.schemas.enable", "false"); + props.setProperty("value.converter.schemas.enable", "false"); + props.setProperty("include.schema.changes", "false"); + props.setProperty("tombstones.on.delete", "false"); + props.setProperty("snapshot.mode", "no_data"); + props.setProperty("tasks.max", "1"); + props.setProperty("decimal.handling.mode", "double"); + + props.setProperty("database.hostname", config.getHost()); + props.setProperty("database.port", String.valueOf(config.getPort())); + props.setProperty("database.user", config.getUsername()); + props.setProperty("database.password", config.getPassword()); + props.setProperty("database.names", config.getDatabase()); + props.setProperty("database.encrypt", String.valueOf(config.getEncrypt())); + props.setProperty("database.trustServerCertificate", String.valueOf(config.getTrustServerCertificate())); + props.setProperty("topic.prefix", logicalName); + props.setProperty("table.include.list", config.getTable()); + + return props; + } + + private String createLogicalName(MsSqlCdcAdapterConfig config) { + return (config.getDatabase() + "_" + config.getTable()) + .replace('.', '_') + .replaceAll("[^A-Za-z0-9_]", "_") + .toLowerCase(); + } + + private void processRecord(ChangeEvent<String, String> record, IEventCollector collector) { + if (record == null || record.value() == null) { + return; + } + + try { + JsonNode root = mapper.readTree(record.value()); + if (!"c".equals(root.path("op").asText())) { + return; + } + + JsonNode afterNode = root.path("after"); + if (!afterNode.isObject()) { + return; + } + + collector.collect(normalizeRow(mapper.convertValue(afterNode, MAP_TYPE))); + } catch (Exception e) { + LOG.error("Failed to process Debezium change event", e); + } + } + + private Map<String, Object> normalizeRow(Map<String, Object> row) { + if (row == null || row.isEmpty() || temporalColumns == null || temporalColumns.isEmpty()) { + return row; + } + + Map<String, Object> normalized = new LinkedHashMap<>(row); + temporalColumns.forEach((columnName, temporalMode) -> { + if (normalized.containsKey(columnName)) { + normalized.put(columnName, normalizeTemporalValue(normalized.get(columnName), temporalMode)); + } + }); + return normalized; + } + + private Object normalizeTemporalValue(Object value, MsSqlMetadataClient.TemporalColumnMode temporalMode) { + if (value == null) { + return null; + } + + return switch (temporalMode) { + case DATE_EPOCH_MILLIS -> normalizeDateValue(asLong(value)); + case TIME_MILLIS -> asLong(value); + case TIME_MICROS -> asLong(value) / 1_000L; + case TIME_NANOS -> asLong(value) / 1_000_000L; + case TIMESTAMP_MILLIS -> normalizeTimestampValue(asLong(value), 1L); + case TIMESTAMP_MICROS -> normalizeTimestampValue(asLong(value), 1_000L); + case TIMESTAMP_NANOS -> normalizeTimestampValue(asLong(value), 1_000_000L); + case DATETIMEOFFSET_MILLIS -> parseOffsetDateTime(value); + }; + } + + private long normalizeDateValue(long epochDays) { + LocalDate localDate = LocalDate.ofEpochDay(epochDays); + return localDate.atStartOfDay(getConfiguredZoneId()).toInstant().toEpochMilli(); + } + + private long normalizeTimestampValue(long rawValue, long divisor) { + long epochMillis = rawValue / divisor; + LocalDateTime localDateTime = Instant.ofEpochMilli(epochMillis).atOffset(ZoneOffset.UTC).toLocalDateTime(); + return localDateTime.atZone(getConfiguredZoneId()).toInstant().toEpochMilli(); + } + + private long asLong(Object value) { + if (value instanceof Number number) { + return number.longValue(); + } + + return Long.parseLong(String.valueOf(value)); + } + + private Object parseOffsetDateTime(Object value) { + try { + return OffsetDateTime.parse(String.valueOf(value)).toInstant().toEpochMilli(); + } catch (DateTimeParseException e) { + LOG.warn("Failed to parse DATETIMEOFFSET value '{}', forwarding raw value.", value); + return value; + } + } + + private ZoneId getConfiguredZoneId() { + return configuredZoneId == null ? ZoneOffset.UTC : configuredZoneId; + } +} diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterConfig.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterConfig.java new file mode 100644 index 0000000000..3d1374067a --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterConfig.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.streampipes.extensions.connectors.cdc.adapter.mssql; + +import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor; +import org.apache.streampipes.model.staticproperty.Option; +import org.apache.streampipes.model.staticproperty.RuntimeResolvableOneOfStaticProperty; + +import java.time.ZoneId; +import java.time.ZoneOffset; + +public class MsSqlCdcAdapterConfig { + + public static final String HOST_KEY = "host"; + public static final String PORT_KEY = "port"; + public static final String DATABASE_KEY = "database"; + public static final String USERNAME_KEY = "username"; + public static final String PASSWORD_KEY = "password"; + public static final String TABLE_KEY = "table"; + public static final String ENCRYPT_KEY = "encrypt"; + public static final String TRUST_SERVER_CERTIFICATE_KEY = "trust-server-certificate"; + public static final String TIMEZONE_KEY = "timezone"; + + private final String host; + private final Integer port; + private final String database; + private final String username; + private final String password; + private final String table; + private final Boolean encrypt; + private final Boolean trustServerCertificate; + private final String timezone; + + public MsSqlCdcAdapterConfig(String host, + Integer port, + String database, + String username, + String password, + String table, + Boolean encrypt, + Boolean trustServerCertificate, + String timezone) { + this.host = host; + this.port = port; + this.database = database; + this.username = username; + this.password = password; + this.table = table; + this.encrypt = encrypt; + this.trustServerCertificate = trustServerCertificate; + this.timezone = timezone; + } + + public static MsSqlCdcAdapterConfig from(IStaticPropertyExtractor extractor) { + return from(extractor, true); + } + + public static MsSqlCdcAdapterConfig from(IStaticPropertyExtractor extractor, boolean requireTableSelection) { + return new MsSqlCdcAdapterConfig( + extractor.singleValueParameter(HOST_KEY, String.class), + extractor.singleValueParameter(PORT_KEY, Integer.class), + extractor.singleValueParameter(DATABASE_KEY, String.class), + extractor.singleValueParameter(USERNAME_KEY, String.class), + extractor.secretValue(PASSWORD_KEY), + extractTable(extractor, requireTableSelection), + extractor.slideToggleValue(ENCRYPT_KEY), + extractor.slideToggleValue(TRUST_SERVER_CERTIFICATE_KEY), + extractor.singleValueParameter(TIMEZONE_KEY, String.class) + ); + } + + private static String extractTable(IStaticPropertyExtractor extractor, boolean requireTableSelection) { + if (requireTableSelection) { + return extractor.selectedSingleValue(TABLE_KEY, String.class); + } + + RuntimeResolvableOneOfStaticProperty tableProperty = + extractor.getStaticPropertyByName(TABLE_KEY, RuntimeResolvableOneOfStaticProperty.class); + + return tableProperty.getOptions() + .stream() + .filter(Option::isSelected) + .findFirst() + .map(Option::getName) + .orElse(null); + } + + public String getHost() { + return host; + } + + public Integer getPort() { + return port; + } + + public String getDatabase() { + return database; + } + + public String getUsername() { + return username; + } + + public String getPassword() { + return password; + } + + public String getTable() { + return table; + } + + public Boolean getEncrypt() { + return encrypt; + } + + public Boolean getTrustServerCertificate() { + return trustServerCertificate; + } + + public String getTimezone() { + return timezone; + } + + public ZoneId getZoneId() { + return timezone == null || timezone.isBlank() ? ZoneOffset.UTC : ZoneId.of(timezone); + } +} diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlMetadataClient.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlMetadataClient.java new file mode 100644 index 0000000000..4d4d07d582 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlMetadataClient.java @@ -0,0 +1,266 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.streampipes.extensions.connectors.cdc.adapter.mssql; + +import org.apache.streampipes.commons.exceptions.SpConfigurationException; +import org.apache.streampipes.commons.exceptions.connect.AdapterException; +import org.apache.streampipes.model.staticproperty.Option; + +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Types; +import java.time.LocalDate; +import java.time.LocalTime; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +public class MsSqlMetadataClient { + + private static final String CDC_ENABLED_QUERY = + "SELECT is_cdc_enabled FROM sys.databases WHERE name = DB_NAME()"; + + private static final String CDC_TABLES_QUERY = + "SELECT s.name AS schema_name, t.name AS table_name " + + "FROM sys.tables t " + + "JOIN sys.schemas s ON t.schema_id = s.schema_id " + + "WHERE t.is_ms_shipped = 0 AND t.is_tracked_by_cdc = 1 " + + "ORDER BY s.name, t.name"; + + public List<Option> discoverTables(MsSqlCdcAdapterConfig config) throws SpConfigurationException { + try (Connection connection = openConnection(config)) { + validateCdcEnabled(connection); + + List<Option> options = new ArrayList<>(); + try (PreparedStatement statement = connection.prepareStatement(CDC_TABLES_QUERY); + ResultSet resultSet = statement.executeQuery()) { + while (resultSet.next()) { + options.add(new Option(resultSet.getString("schema_name") + "." + resultSet.getString("table_name"))); + } + } + + if (options.isEmpty()) { + throw new SpConfigurationException("No CDC-enabled tables found in database " + config.getDatabase()); + } + + return options; + } catch (SQLException e) { + throw new SpConfigurationException("Failed to resolve SQL Server tables: " + e.getMessage(), e); + } + } + + public void validateSelectedTable(MsSqlCdcAdapterConfig config) throws AdapterException { + try (Connection connection = openConnection(config)) { + validateCdcEnabled(connection); + + List<Option> options = discoverTables(config); + boolean containsTable = options.stream().anyMatch(option -> option.getName().equals(config.getTable())); + if (!containsTable) { + throw new AdapterException("Selected table is not CDC-enabled or does not exist: " + config.getTable()); + } + } catch (SQLException e) { + throw new AdapterException("Failed to validate SQL Server configuration: " + e.getMessage(), e); + } catch (SpConfigurationException e) { + throw new AdapterException(e.getMessage(), e); + } + } + + public Map<String, Object> generateSample(MsSqlCdcAdapterConfig config) throws AdapterException { + TableRef tableRef = parseTable(config.getTable()); + + try (Connection connection = openConnection(config)) { + validateCdcEnabled(connection); + + Map<String, ColumnDescriptor> columns = describeColumns(connection, tableRef); + Map<String, Object> sample = new LinkedHashMap<>(); + ZoneId zoneId = config.getZoneId(); + columns.forEach((columnName, descriptor) -> sample.put(columnName, placeholderForType(descriptor, zoneId))); + + if (sample.isEmpty()) { + throw new AdapterException("No columns found for table " + config.getTable()); + } + + return sample; + } catch (SQLException | SpConfigurationException e) { + throw new AdapterException("Failed to generate sample data: " + e.getMessage(), e); + } + } + + public Connection openConnection(MsSqlCdcAdapterConfig config) throws SQLException { + String url = "jdbc:sqlserver://" + config.getHost() + ":" + config.getPort() + + ";databaseName=" + config.getDatabase() + + ";encrypt=" + config.getEncrypt() + + ";trustServerCertificate=" + config.getTrustServerCertificate(); + + return DriverManager.getConnection(url, config.getUsername(), config.getPassword()); + } + + public Map<String, TemporalColumnMode> describeTemporalColumns(MsSqlCdcAdapterConfig config) throws AdapterException { + TableRef tableRef = parseTable(config.getTable()); + + try (Connection connection = openConnection(config)) { + validateCdcEnabled(connection); + + Map<String, TemporalColumnMode> temporalColumns = new LinkedHashMap<>(); + Map<String, ColumnDescriptor> columns = describeColumns(connection, tableRef); + columns.forEach((columnName, descriptor) -> { + TemporalColumnMode temporalMode = getTemporalColumnMode(descriptor); + if (temporalMode != null) { + temporalColumns.put(columnName, temporalMode); + } + }); + return temporalColumns; + } catch (SQLException | SpConfigurationException e) { + throw new AdapterException("Failed to inspect SQL Server table metadata: " + e.getMessage(), e); + } + } + + private void validateCdcEnabled(Connection connection) throws SQLException, SpConfigurationException { + try (PreparedStatement statement = connection.prepareStatement(CDC_ENABLED_QUERY); + ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next() || !resultSet.getBoolean(1)) { + throw new SpConfigurationException("CDC is not enabled on database " + connection.getCatalog()); + } + } + } + + private TableRef parseTable(String table) throws AdapterException { + if (table == null || table.isBlank()) { + throw new AdapterException("No table selected."); + } + + String[] tokens = table.split("\\.", 2); + if (tokens.length != 2) { + throw new AdapterException("Selected table must be of the form schema.table: " + table); + } + + return new TableRef(tokens[0], tokens[1]); + } + + private Map<String, ColumnDescriptor> describeColumns(Connection connection, + TableRef tableRef) throws SQLException { + DatabaseMetaData metaData = connection.getMetaData(); + Map<String, ColumnDescriptor> columns = new LinkedHashMap<>(); + try (ResultSet resultSet = metaData.getColumns(null, tableRef.schema(), tableRef.table(), null)) { + while (resultSet.next()) { + String columnName = resultSet.getString("COLUMN_NAME"); + int jdbcType = resultSet.getInt("DATA_TYPE"); + String typeName = resultSet.getString("TYPE_NAME"); + int decimalDigits = resultSet.getInt("DECIMAL_DIGITS"); + columns.put(columnName, new ColumnDescriptor(columnName, jdbcType, typeName, decimalDigits)); + } + } + return columns; + } + + private Object placeholderForType(ColumnDescriptor descriptor, ZoneId zoneId) { + TemporalColumnMode temporalMode = getTemporalColumnMode(descriptor); + if (temporalMode != null) { + return sampleTemporalValue(temporalMode, zoneId); + } + + return switch (descriptor.jdbcType()) { + case Types.BIGINT, Types.INTEGER, Types.SMALLINT, Types.TINYINT -> 1; + case Types.DECIMAL, Types.DOUBLE, Types.FLOAT, Types.NUMERIC, Types.REAL -> 1.0; + case Types.BIT, Types.BOOLEAN -> true; + case Types.BINARY, Types.LONGVARBINARY, Types.VARBINARY -> "AQID"; + default -> "example"; + }; + } + + private TemporalColumnMode getTemporalColumnMode(ColumnDescriptor descriptor) { + String normalizedTypeName = descriptor.typeName() == null + ? "" + : descriptor.typeName().toLowerCase(Locale.ENGLISH); + + return switch (normalizedTypeName) { + case "date" -> TemporalColumnMode.DATE_EPOCH_MILLIS; + case "time" -> { + if (descriptor.decimalDigits() <= 3) { + yield TemporalColumnMode.TIME_MILLIS; + } else if (descriptor.decimalDigits() <= 6) { + yield TemporalColumnMode.TIME_MICROS; + } else { + yield TemporalColumnMode.TIME_NANOS; + } + } + case "datetime", "smalldatetime" -> TemporalColumnMode.TIMESTAMP_MILLIS; + case "datetime2" -> { + if (descriptor.decimalDigits() <= 3) { + yield TemporalColumnMode.TIMESTAMP_MILLIS; + } else if (descriptor.decimalDigits() <= 6) { + yield TemporalColumnMode.TIMESTAMP_MICROS; + } else { + yield TemporalColumnMode.TIMESTAMP_NANOS; + } + } + case "datetimeoffset" -> TemporalColumnMode.DATETIMEOFFSET_MILLIS; + default -> { + if (descriptor.jdbcType() == Types.DATE) { + yield TemporalColumnMode.DATE_EPOCH_MILLIS; + } else if (descriptor.jdbcType() == Types.TIME || descriptor.jdbcType() == Types.TIME_WITH_TIMEZONE) { + yield TemporalColumnMode.TIME_MILLIS; + } else if (descriptor.jdbcType() == Types.TIMESTAMP + || descriptor.jdbcType() == Types.TIMESTAMP_WITH_TIMEZONE) { + yield TemporalColumnMode.TIMESTAMP_MILLIS; + } + yield null; + } + }; + } + + private Object sampleTemporalValue(TemporalColumnMode temporalMode, ZoneId zoneId) { + return switch (temporalMode) { + case DATE_EPOCH_MILLIS -> + LocalDate.of(2026, 5, 7).atStartOfDay(zoneId).toInstant().toEpochMilli(); + case TIME_MILLIS, TIME_MICROS, TIME_NANOS -> + LocalTime.of(10, 15, 30, 123_000_000).toNanoOfDay() / 1_000_000; + case TIMESTAMP_MILLIS, TIMESTAMP_MICROS, TIMESTAMP_NANOS, DATETIMEOFFSET_MILLIS -> + LocalDate.of(2026, 5, 7) + .atTime(10, 15, 30, 123_000_000) + .atZone(zoneId) + .toInstant() + .toEpochMilli(); + }; + } + + private record TableRef(String schema, String table) { + } + + public enum TemporalColumnMode { + DATE_EPOCH_MILLIS, + TIME_MILLIS, + TIME_MICROS, + TIME_NANOS, + TIMESTAMP_MILLIS, + TIMESTAMP_MICROS, + TIMESTAMP_NANOS, + DATETIMEOFFSET_MILLIS + } + + private record ColumnDescriptor(String columnName, int jdbcType, String typeName, int decimalDigits) { + } +} diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/strings.en b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/strings.en new file mode 100644 index 0000000000..c73518e0f7 --- /dev/null +++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/strings.en @@ -0,0 +1,47 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. +# + + +org.apache.streampipes.connect.cdc.adapter.mssql.title=Microsoft SQL Server CDC +org.apache.streampipes.connect.cdc.adapter.mssql.description=Captures newly inserted rows from a CDC-enabled Microsoft SQL Server table + +host.title=Host +host.description=SQL Server hostname or IP address + +port.title=Port +port.description=SQL Server TCP port + +database.title=Database +database.description=Name of the CDC-enabled SQL Server database + +username.title=Username +username.description=SQL Server user name + +password.title=Password +password.description=SQL Server password + +encrypt.title=Encrypt Connection +encrypt.description=Use TLS when connecting to SQL Server + +trust-server-certificate.title=Trust Server Certificate +trust-server-certificate.description=Trust the SQL Server certificate without validating its chain + +timezone.title=Timezone +timezone.description=Timezone to assume for SQL Server DATETIME and DATETIME2 values without timezone information, for example Europe/Berlin + +table.title=Table +table.description=Select a CDC-enabled table from the configured database diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml index 4507aff55c..319995a00b 100644 --- a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml +++ b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml @@ -197,6 +197,12 @@ <version>0.99.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-connectors-cdc</artifactId> + <version>0.99.0-SNAPSHOT</version> + </dependency> + <dependency> <groupId>org.bouncycastle</groupId> <artifactId>bcprov-jdk18on</artifactId> diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java index fb64c58b37..9290d205da 100644 --- a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java +++ b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java @@ -20,6 +20,7 @@ package org.apache.streampipes.extensions.all.iiot; import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport; import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport; +import org.apache.streampipes.extensions.connectors.cdc.CdcConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport; @@ -85,7 +86,8 @@ public class AllExtensionsIIoTInit extends StreamPipesExtensionsServiceBase { new DatabaseSinksExtensionModuleExport(), new InternalSinksExtensionModuleExports(), new NotificationsExtensionModuleExport(), - new LlmExtensionModuleExport() + new LlmExtensionModuleExport(), + new CdcConnectorsModuleExport() ) .registerRuntimeProvider(new StandaloneStreamPipesRuntimeProvider()) .registerMessagingProtocols( diff --git a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml index f6da620463..4e30d10153 100644 --- a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml +++ b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml @@ -74,6 +74,11 @@ <artifactId>streampipes-connectors-camel-azure</artifactId> <version>0.99.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-connectors-cdc</artifactId> + <version>0.99.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-connectors-influx</artifactId> diff --git a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java index e39dbdf0d8..ab7f2d6ea3 100644 --- a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java +++ b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java @@ -21,6 +21,7 @@ import org.apache.streampipes.connect.GeneralAdaptersExtensionModuleExport; import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport; import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.camel.azure.CamelAzureConnectorsModuleExport; +import org.apache.streampipes.extensions.connectors.cdc.CdcConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport; import org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport; @@ -69,6 +70,7 @@ public class AllExtensionsInit extends StreamPipesExtensionsServiceBase { new IIoTAdaptersExtensionModuleExport(), new CamelAzureConnectorsModuleExport(), + new CdcConnectorsModuleExport(), new InfluxConnectorsModuleExport(), new KafkaConnectorsModuleExport(), new MqttConnectorsModuleExport(), diff --git a/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.ts b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.ts index 2dbb534d10..203439b179 100644 --- a/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.ts +++ b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.ts @@ -79,6 +79,7 @@ export class StaticRuntimeResolvableOneOfInputComponent } } this.staticProperty.options = staticProperty.options; + this.performValidation(); } isOptionSelected(): boolean {
