This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 060e660c3f feat(#4451): Add MSSQL CDC connector (#4452)
060e660c3f is described below
commit 060e660c3f489e280cb8ebb4a9b2668af41c8104
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon May 11 22:02:58 2026 +0200
feat(#4451): Add MSSQL CDC connector (#4452)
---
.../apache/streampipes/commons/constants/Envs.java | 2 +
.../commons/environment/DefaultEnvironment.java | 5 +
.../commons/environment/Environment.java | 2 +
streampipes-extensions/pom.xml | 1 +
.../streampipes-connectors-cdc/pom.xml | 85 +++++
.../connectors/cdc/CdcConnectorsModuleExport.java | 46 +++
.../cdc/adapter/mssql/MsSqlCdcAdapter.java | 416 +++++++++++++++++++++
.../cdc/adapter/mssql/MsSqlCdcAdapterConfig.java | 143 +++++++
.../cdc/adapter/mssql/MsSqlMetadataClient.java | 404 ++++++++++++++++++++
.../documentation.md | 142 +++++++
.../icon.png | Bin 0 -> 58929 bytes
.../strings.en | 47 +++
.../cdc/adapter/mssql/MsSqlCdcAdapterTest.java | 173 +++++++++
.../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 +
18 files changed, 1483 insertions(+), 1 deletion(-)
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index 7e5a47aa66..c608e4d486 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -103,6 +103,7 @@ public enum Envs {
SP_PULSAR_URL("SP_PULSAR_URL", "pulsar://localhost:6650"),
SP_CORE_EXTENSION_TRANSPORT_MODE("SP_CORE_EXTENSION_TRANSPORT_MODE", "http"),
+
SP_CORE_EXTENSIONS_TRANSPORT_MODE_TIMEOUT_SECONDS("SP_CORE_EXTENSIONS_TRANSPORT_MODE_TIMEOUT_SECONDS",
"20"),
SP_EXTENSION_TRANSPORT_MODE("SP_EXTENSION_TRANSPORT_MODE", "http"),
SP_EXTENSION_REQUEST_TOPIC_PREFIX(
"SP_EXTENSION_REQUEST_TOPIC_PREFIX",
@@ -155,6 +156,7 @@ public enum Envs {
// PLC4X connection cache
SP_PLC4X_CONN_MAX_WAIT_TIME_MS("SP_PLC4X_CONN_MAX_WAIT_TIME_MS", "20000"),
SP_PLC4X_CONN_MAX_LEASE_TIME_MS("SP_PLC4X_CONN_MAX_LEASE_TIME_MS", "4000"),
+ SP_CDC_MSSQL_POLL_INTERVAL_MS("SP_CDC_MSSQL_POLL_INTERVAL_MS", "1000"),
// Retention Local File
SP_RETENTION_LOCAL_DIR("SP_RETENTION_LOCAL_DIR", "./ArchivedData"),
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
index 76799749d6..edcd692de4 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
@@ -427,6 +427,11 @@ public class DefaultEnvironment implements Environment {
return new IntEnvironmentVariable(Envs.SP_PLC4X_CONN_MAX_LEASE_TIME_MS);
}
+ @Override
+ public IntEnvironmentVariable getMsSqlCdcPollIntervalMs() {
+ return new IntEnvironmentVariable(Envs.SP_CDC_MSSQL_POLL_INTERVAL_MS);
+ }
+
@Override
public BooleanEnvironmentVariable getFileLoggingEnabled() {
return new BooleanEnvironmentVariable(Envs.SP_LOGGING_FILE_ENABLED);
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
index 3bf54cce08..56a7d98c4d 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
@@ -202,6 +202,8 @@ public interface Environment {
IntEnvironmentVariable getPlc4xMaxLeaseTimeMs();
+ IntEnvironmentVariable getMsSqlCdcPollIntervalMs();
+
BooleanEnvironmentVariable getFileLoggingEnabled();
BooleanEnvironmentVariable getConsoleLoggingEnabled();
diff --git a/streampipes-extensions/pom.xml b/streampipes-extensions/pom.xml
index 749c3bbf24..50bb621557 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-filewatcher</module>
<module>streampipes-connectors-influx</module>
<module>streampipes-connectors-kafka</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..5bc89ec9c8
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-cdc/pom.xml
@@ -0,0 +1,85 @@
+<?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>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </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..c771b701f1
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapter.java
@@ -0,0 +1,416 @@
+/*
+ * 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.environment.Environments;
+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.monitoring.IExtensionsLogger;
+import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig;
+import org.apache.streampipes.model.connect.guess.SampleData;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
+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.CountDownLatch;
+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";
+ static final long ENGINE_STARTUP_TIMEOUT_SECONDS = 10;
+
+ 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;
+ private volatile Throwable engineFailure;
+ private volatile boolean connectorStarted;
+ private volatile boolean stopRequested;
+
+ 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)
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
+ .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),
ZoneOffset.UTC.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();
+ this.engineFailure = null;
+ this.connectorStarted = false;
+ this.stopRequested = false;
+
+ CountDownLatch startupLatch = new CountDownLatch(1);
+ Properties properties = buildDebeziumProperties(config,
extractor.getAdapterDescription().getElementId());
+ this.engine = DebeziumEngine.create(Json.class)
+ .using(properties)
+ .using((success, message, error) -> handleEngineCompletion(
+ success,
+ message,
+ error,
+ adapterRuntimeContext.getLogger(),
+ startupLatch
+ ))
+ .using(createConnectorCallback(startupLatch))
+ .notifying(record -> processRecord(record, collector))
+ .build();
+
+ this.executorService = Executors.newSingleThreadExecutor();
+ this.executorService.execute(engine);
+ awaitStartup(startupLatch, adapterRuntimeContext.getLogger());
+ }
+
+ @Override
+ public void onAdapterStopped(IAdapterParameterExtractor extractor,
+ IAdapterRuntimeContext adapterRuntimeContext)
throws AdapterException {
+ this.stopRequested = true;
+
+ 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;
+ this.engineFailure = null;
+ this.connectorStarted = false;
+ this.stopRequested = false;
+ }
+
+ @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();
+ }
+
+ Properties buildDebeziumProperties(MsSqlCdcAdapterConfig config, String
elementId) {
+ Properties props = new Properties();
+ String logicalName = createLogicalName(config, elementId);
+ 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(
+ "poll.interval.ms",
+
String.valueOf(Environments.getEnvironment().getMsSqlCdcPollIntervalMs().getValueOrDefault())
+ );
+
+ 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;
+ }
+
+ String createLogicalName(MsSqlCdcAdapterConfig config, String elementId) {
+ String effectiveElementId = elementId == null || elementId.isBlank() ?
"adapter" : elementId;
+
+ return (config.getDatabase() + "_" + config.getTable() + "_" +
effectiveElementId)
+ .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;
+ }
+
+ 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;
+ }
+
+ private DebeziumEngine.ConnectorCallback
createConnectorCallback(CountDownLatch startupLatch) {
+ return new DebeziumEngine.ConnectorCallback() {
+ @Override
+ public void connectorStarted() {
+ connectorStarted = true;
+ startupLatch.countDown();
+ }
+ };
+ }
+
+ void handleEngineCompletion(boolean success,
+ String message,
+ Throwable error,
+ IExtensionsLogger logger,
+ CountDownLatch startupLatch) {
+ if (error != null) {
+ this.engineFailure = error;
+ } else if (!success) {
+ this.engineFailure = new AdapterException(
+ "Debezium engine stopped unexpectedly" + (message == null ||
message.isBlank() ? "." : ": " + message)
+ );
+ }
+
+ if (startupLatch != null) {
+ startupLatch.countDown();
+ }
+
+ if (stopRequested) {
+ return;
+ }
+
+ if (engineFailure != null) {
+ logger.error("Debezium engine failed for MSSQL CDC adapter.",
asException(engineFailure));
+ } else if (success) {
+ logger.warn("Debezium Engine Stopped", "The embedded Debezium engine
stopped unexpectedly.");
+ } else {
+ logger.warn(
+ "Debezium Engine Stopped",
+ message == null || message.isBlank() ? "The embedded Debezium engine
stopped unexpectedly." : message
+ );
+ }
+ }
+
+ private void awaitStartup(CountDownLatch startupLatch, IExtensionsLogger
logger) throws AdapterException {
+ try {
+ boolean completed = startupLatch.await(ENGINE_STARTUP_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ if (engineFailure != null) {
+ throw new AdapterException("Failed to start Debezium engine: " +
engineFailure.getMessage(), engineFailure);
+ }
+
+ if (!completed || !connectorStarted) {
+ logger.warn(
+ "Debezium Startup Delayed",
+ "The embedded Debezium engine did not confirm startup within "
+ + ENGINE_STARTUP_TIMEOUT_SECONDS + " seconds."
+ );
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AdapterException("Interrupted while waiting for Debezium
engine startup.", e);
+ }
+ }
+
+ private Exception asException(Throwable throwable) {
+ if (throwable instanceof Exception exception) {
+ return exception;
+ }
+
+ return new AdapterException(throwable.getMessage(), throwable);
+ }
+}
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..c2d581a5f0
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlMetadataClient.java
@@ -0,0 +1,404 @@
+/*
+ * 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.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.Date;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Base64;
+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";
+
+ private static final String SAMPLE_ROW_QUERY_TEMPLATE = "SELECT TOP 1 * FROM
[%s].[%s]";
+
+ 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);
+ ZoneId zoneId = config.getZoneId();
+ Map<String, Object> sample = fetchRealSampleRow(connection, tableRef,
columns, zoneId);
+
+ if (sample == null) {
+ sample = new LinkedHashMap<>();
+ for (Map.Entry<String, ColumnDescriptor> entry : columns.entrySet()) {
+ sample.put(entry.getKey(), placeholderForType(entry.getValue(),
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 Map<String, Object> fetchRealSampleRow(Connection connection,
+ TableRef tableRef,
+ Map<String, ColumnDescriptor>
columns,
+ ZoneId zoneId) throws
SQLException {
+ String query = SAMPLE_ROW_QUERY_TEMPLATE.formatted(
+ escapeIdentifier(tableRef.schema()),
+ escapeIdentifier(tableRef.table())
+ );
+
+ try (PreparedStatement statement = connection.prepareStatement(query);
+ ResultSet resultSet = statement.executeQuery()) {
+ if (!resultSet.next()) {
+ return null;
+ }
+
+ ResultSetMetaData metaData = resultSet.getMetaData();
+ Map<String, Object> sample = new LinkedHashMap<>();
+ for (int columnIndex = 1; columnIndex <= metaData.getColumnCount();
columnIndex++) {
+ String columnName = metaData.getColumnLabel(columnIndex);
+ ColumnDescriptor descriptor = columns.getOrDefault(
+ columnName,
+ new ColumnDescriptor(columnName,
metaData.getColumnType(columnIndex), metaData.getColumnTypeName(columnIndex), 0)
+ );
+ Object rawValue = resultSet.getObject(columnIndex);
+ sample.put(columnName, sampleValue(rawValue, descriptor, zoneId));
+ }
+
+ return sample;
+ }
+ }
+
+ private Object placeholderForType(ColumnDescriptor descriptor, ZoneId
zoneId) {
+ TemporalColumnMode temporalMode = getTemporalColumnMode(descriptor);
+ if (temporalMode != null) {
+ return (double) 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 Object sampleValue(Object rawValue, ColumnDescriptor descriptor,
ZoneId zoneId) {
+ if (rawValue == null) {
+ return placeholderForType(descriptor, zoneId);
+ }
+
+ TemporalColumnMode temporalMode = getTemporalColumnMode(descriptor);
+ if (temporalMode != null) {
+ return (double) temporalValue(rawValue, temporalMode, zoneId);
+ }
+
+ if (rawValue instanceof BigDecimal decimalValue) {
+ return decimalValue.doubleValue();
+ }
+
+ if (rawValue instanceof byte[] bytes) {
+ return Base64.getEncoder().encodeToString(bytes);
+ }
+
+ return rawValue;
+ }
+
+ 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 long temporalValue(Object rawValue, TemporalColumnMode temporalMode,
ZoneId zoneId) {
+ return switch (temporalMode) {
+ case DATE_EPOCH_MILLIS -> toDateEpochMillis(rawValue, zoneId);
+ case TIME_MILLIS, TIME_MICROS, TIME_NANOS -> toTimeMillis(rawValue);
+ case TIMESTAMP_MILLIS, TIMESTAMP_MICROS, TIMESTAMP_NANOS ->
toTimestampEpochMillis(rawValue, zoneId);
+ case DATETIMEOFFSET_MILLIS -> toDateTimeOffsetEpochMillis(rawValue,
zoneId);
+ };
+ }
+
+ private long toDateEpochMillis(Object rawValue, ZoneId zoneId) {
+ if (rawValue instanceof Date dateValue) {
+ return
dateValue.toLocalDate().atStartOfDay(zoneId).toInstant().toEpochMilli();
+ } else if (rawValue instanceof LocalDate localDate) {
+ return localDate.atStartOfDay(zoneId).toInstant().toEpochMilli();
+ }
+
+ return
LocalDate.parse(rawValue.toString()).atStartOfDay(zoneId).toInstant().toEpochMilli();
+ }
+
+ private long toTimeMillis(Object rawValue) {
+ LocalTime localTime;
+ if (rawValue instanceof Time timeValue) {
+ localTime = timeValue.toLocalTime();
+ } else if (rawValue instanceof LocalTime parsedLocalTime) {
+ localTime = parsedLocalTime;
+ } else {
+ localTime = LocalTime.parse(rawValue.toString());
+ }
+
+ return localTime.toNanoOfDay() / 1_000_000;
+ }
+
+ private long toTimestampEpochMillis(Object rawValue, ZoneId zoneId) {
+ if (rawValue instanceof Timestamp timestampValue) {
+ return
timestampValue.toLocalDateTime().atZone(zoneId).toInstant().toEpochMilli();
+ } else if (rawValue instanceof LocalDateTime localDateTime) {
+ return localDateTime.atZone(zoneId).toInstant().toEpochMilli();
+ } else if (rawValue instanceof Instant instant) {
+ return instant.toEpochMilli();
+ }
+
+ String stringValue = rawValue.toString();
+ try {
+ return Instant.parse(stringValue).toEpochMilli();
+ } catch (DateTimeParseException ignored) {
+ return LocalDateTime.parse(stringValue.replace(' ',
'T')).atZone(zoneId).toInstant().toEpochMilli();
+ }
+ }
+
+ private long toDateTimeOffsetEpochMillis(Object rawValue, ZoneId zoneId) {
+ if (rawValue instanceof OffsetDateTime offsetDateTime) {
+ return offsetDateTime.toInstant().toEpochMilli();
+ } else if (rawValue instanceof Timestamp timestampValue) {
+ return timestampValue.toInstant().toEpochMilli();
+ }
+
+ String stringValue = rawValue.toString().replace(' ', 'T');
+ try {
+ return OffsetDateTime.parse(stringValue).toInstant().toEpochMilli();
+ } catch (DateTimeParseException ignored) {
+ return
LocalDateTime.parse(stringValue).atZone(zoneId).toInstant().toEpochMilli();
+ }
+ }
+
+ private String escapeIdentifier(String identifier) {
+ return identifier.replace("]", "]]");
+ }
+
+ 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/documentation.md
b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/documentation.md
new file mode 100644
index 0000000000..1566a3bea1
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/documentation.md
@@ -0,0 +1,142 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+## Microsoft SQL Server CDC Adapter
+
+***
+
+## Description
+
+This adapter uses the embedded Debezium Engine to capture change events from a
CDC-enabled Microsoft SQL Server table.
+
+Current behavior:
+
+* only newly inserted rows are emitted
+* updates and deletes are ignored
+* the output event is flat and contains only the row fields
+* Debezium metadata is not included in emitted events
+* adapter state is kept in memory only
+
+This means the adapter starts at the current log position and does not read
old rows when it starts.
+If the adapter is stopped, its position is lost and rows written while it is
down can be missed.
+
+***
+
+## Required Input
+
+None.
+
+***
+
+## SQL Server Prerequisites
+
+The adapter expects the following SQL Server setup:
+
+* CDC is enabled on the selected database
+* CDC is enabled on the selected table
+* SQL Server Agent is running
+* the configured SQL Server user can access the database metadata and captured
CDC data
+
+The configuration dialog only lists CDC-enabled tables from the selected
database.
+
+***
+
+## Configuration
+
+### Host
+
+The SQL Server host name or IP address.
+
+### Port
+
+The SQL Server TCP port, for example `1433`.
+
+### Database
+
+The name of the CDC-enabled SQL Server database.
+
+### Username / Password
+
+Credentials used to connect to SQL Server.
+
+### Encrypt Connection
+
+Enable TLS for the SQL Server connection.
+
+### Trust Server Certificate
+
+Trust the SQL Server certificate without validating its chain.
+This is useful for local or test environments with self-signed certificates.
+
+### Timezone
+
+Timezone used for SQL Server temporal types without embedded timezone
information, such as `DATETIME` and `DATETIME2`.
+
+Use an IANA timezone ID such as:
+
+* `UTC`
+* `Europe/Berlin`
+* `America/New_York`
+
+The default is `UTC`.
+
+### Table
+
+Select one CDC-enabled table from the configured database.
+
+***
+
+## Output Behavior
+
+Each insert event produces one StreamPipes event.
+
+Example output:
+
+```json
+{
+ "id": 42,
+ "temperature": 23.5,
+ "pressure": 1007.2,
+ "createdAt": 1778154600000
+}
+```
+
+Type handling:
+
+* `DECIMAL` and `NUMERIC` values are emitted as JSON numbers
+* `DATE`, `DATETIME`, `SMALLDATETIME`, `DATETIME2`, and `DATETIMEOFFSET`
values are normalized to epoch milliseconds
+* `TIME` values are normalized to milliseconds past midnight
+* binary values are emitted as Base64 strings
+
+Runtime polling:
+
+* Debezium polling is controlled through the environment variable
`SP_CDC_MSSQL_POLL_INTERVAL_MS`
+* the default polling interval is `1000` milliseconds
+
+***
+
+## Limitations
+
+Current limitations:
+
+* single-table capture only
+* insert events only
+* no snapshot of existing rows
+* no offset persistence across adapter restarts
+* no schema-history persistence across adapter restarts
+* timestamp semantics for downstream sinks still depend on the field selection
in StreamPipes
diff --git
a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/icon.png
b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/icon.png
new file mode 100644
index 0000000000..0a6ff414b7
Binary files /dev/null and
b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql/icon.png
differ
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..a54beb12af
--- /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. Defaults to UTC.
+
+table.title=Table
+table.description=Select a CDC-enabled table from the configured database
diff --git
a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterTest.java
b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterTest.java
new file mode 100644
index 0000000000..90b2a015eb
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapterTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.monitoring.IExtensionsLogger;
+import org.apache.streampipes.model.monitoring.SpLogMessage;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.Option;
+import
org.apache.streampipes.model.staticproperty.RuntimeResolvableOneOfStaticProperty;
+import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
+import org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty;
+import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class MsSqlCdcAdapterTest {
+
+ @Test
+ void configExtractionAllowsMissingTableDuringRuntimeResolution() {
+ RuntimeResolvableOneOfStaticProperty tableProperty =
+ new
RuntimeResolvableOneOfStaticProperty(MsSqlCdcAdapterConfig.TABLE_KEY, "", "");
+ tableProperty.setOptions(List.of(new Option("dbo.Measurements")));
+
+ MsSqlCdcAdapterConfig config = MsSqlCdcAdapterConfig.from(
+ StaticPropertyExtractor.from(List.of(
+ textProperty(MsSqlCdcAdapterConfig.HOST_KEY, "localhost"),
+ textProperty(MsSqlCdcAdapterConfig.PORT_KEY, "1433"),
+ textProperty(MsSqlCdcAdapterConfig.DATABASE_KEY, "testdb"),
+ textProperty(MsSqlCdcAdapterConfig.USERNAME_KEY, "sa"),
+ secretProperty(MsSqlCdcAdapterConfig.PASSWORD_KEY, "secret"),
+ toggleProperty(MsSqlCdcAdapterConfig.ENCRYPT_KEY, true),
+ toggleProperty(MsSqlCdcAdapterConfig.TRUST_SERVER_CERTIFICATE_KEY,
false),
+ textProperty(MsSqlCdcAdapterConfig.TIMEZONE_KEY, "UTC"),
+ tableProperty
+ )),
+ false
+ );
+
+ assertNull(config.getTable());
+ }
+
+ @Test
+ void buildDebeziumPropertiesUsesDecimalHandlingAndElementId() {
+ MsSqlCdcAdapter adapter = new MsSqlCdcAdapter();
+ MsSqlCdcAdapterConfig config = new MsSqlCdcAdapterConfig(
+ "localhost",
+ 1433,
+ "testdb",
+ "sa",
+ "secret",
+ "dbo.Measurements",
+ true,
+ false,
+ "UTC"
+ );
+
+ Properties properties = adapter.buildDebeziumProperties(config,
"adapter-42");
+
+ assertEquals("double", properties.getProperty("decimal.handling.mode"));
+ assertEquals("1000", properties.getProperty("poll.interval.ms"));
+ assertEquals("testdb_dbo_measurements_adapter_42",
properties.getProperty("topic.prefix"));
+ assertEquals("testdb_dbo_measurements_adapter_42-engine",
properties.getProperty("name"));
+ }
+
+ @Test
+ void timestampNormalizationUsesConfiguredTimezone() throws Exception {
+ MsSqlCdcAdapter adapter = new MsSqlCdcAdapter();
+ Field configuredZoneId =
MsSqlCdcAdapter.class.getDeclaredField("configuredZoneId");
+ configuredZoneId.setAccessible(true);
+ configuredZoneId.set(adapter, ZoneId.of("Europe/Berlin"));
+
+ Object normalized = adapter.normalizeTemporalValue(
+ 1_778_161_800_000L,
+ MsSqlMetadataClient.TemporalColumnMode.TIMESTAMP_MILLIS
+ );
+
+ long expected = LocalDateTime.of(2026, 5, 7, 13, 50)
+ .atZone(ZoneId.of("Europe/Berlin"))
+ .toInstant()
+ .toEpochMilli();
+
+ assertEquals(expected, normalized);
+ }
+
+ @Test
+ void engineCompletionReportsFailuresToRuntimeLogger() throws Exception {
+ MsSqlCdcAdapter adapter = new MsSqlCdcAdapter();
+ TestLogger logger = new TestLogger();
+ RuntimeException cause = new RuntimeException("boom");
+
+ adapter.handleEngineCompletion(false, "connector failed", cause, logger,
new CountDownLatch(1));
+
+ Field engineFailure =
MsSqlCdcAdapter.class.getDeclaredField("engineFailure");
+ engineFailure.setAccessible(true);
+
+ assertNotNull(engineFailure.get(adapter));
+ assertEquals("Debezium engine failed for MSSQL CDC adapter.",
logger.errorDetails);
+ assertEquals(cause, logger.errorException);
+ }
+
+ private static FreeTextStaticProperty textProperty(String internalName,
String value) {
+ FreeTextStaticProperty property = FreeTextStaticProperty.of(internalName,
value);
+ property.setInternalName(internalName);
+ return property;
+ }
+
+ private static SecretStaticProperty secretProperty(String internalName,
String value) {
+ SecretStaticProperty property = new SecretStaticProperty(internalName, "",
"");
+ property.setValue(value);
+ return property;
+ }
+
+ private static SlideToggleStaticProperty toggleProperty(String internalName,
boolean selected) {
+ SlideToggleStaticProperty property = new
SlideToggleStaticProperty(internalName, "", "", selected);
+ property.setSelected(selected);
+ return property;
+ }
+
+ private static class TestLogger implements IExtensionsLogger {
+ private String errorDetails;
+ private Exception errorException;
+
+ @Override
+ public void log(SpLogMessage logMessage) {
+ }
+
+ @Override
+ public void error(Exception e) {
+ this.errorException = e;
+ }
+
+ @Override
+ public void error(String details, Exception e) {
+ this.errorDetails = details;
+ this.errorException = e;
+ }
+
+ @Override
+ public void info(String title, String details) {
+ }
+
+ @Override
+ public void warn(String title, String details) {
+ }
+ }
+}
diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
index 47e559cda0..eddbe63a50 100644
--- a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
@@ -202,6 +202,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 179f9b4ede..b82ec7630e 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.filewatcher.FileWatcherConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
@@ -87,7 +88,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 a165ea8b03..1368085007 100644
--- a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
@@ -79,6 +79,11 @@
<artifactId>streampipes-connectors-filewatcher</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 ef031f35f3..689724dac2 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.filewatcher.FileWatcherConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
@@ -71,6 +72,7 @@ public class AllExtensionsInit extends
StreamPipesExtensionsServiceBase {
new CamelAzureConnectorsModuleExport(),
new FileWatcherConnectorsModuleExport(),
+ 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 {
