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
The following commit(s) were added to refs/heads/add-cdc-connectors by this
push:
new 7547716878 Improve schema guessing, add docs
7547716878 is described below
commit 7547716878202fba2427a19bcc0e54500879238e
Author: Dominik Riemer <[email protected]>
AuthorDate: Thu May 7 21:13:01 2026 +0200
Improve schema guessing, add docs
---
.../streampipes-connectors-cdc/pom.xml | 5 +
.../cdc/adapter/mssql/MsSqlCdcAdapter.java | 118 +++++++++++++-
.../cdc/adapter/mssql/MsSqlMetadataClient.java | 144 ++++++++++++++++-
.../documentation.md | 142 +++++++++++++++++
.../icon.png | Bin 0 -> 58929 bytes
.../strings.en | 2 +-
.../cdc/adapter/mssql/MsSqlCdcAdapterTest.java | 173 +++++++++++++++++++++
7 files changed, 573 insertions(+), 11 deletions(-)
diff --git a/streampipes-extensions/streampipes-connectors-cdc/pom.xml
b/streampipes-extensions/streampipes-connectors-cdc/pom.xml
index 487a0ac203..5bc89ec9c8 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/pom.xml
+++ b/streampipes-extensions/streampipes-connectors-cdc/pom.xml
@@ -75,6 +75,11 @@
<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/adapter/mssql/MsSqlCdcAdapter.java
b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlCdcAdapter.java
index c0051721c0..c771b701f1 100644
---
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
@@ -18,6 +18,7 @@
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;
@@ -27,8 +28,10 @@ import
org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchema
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;
@@ -56,6 +59,7 @@ 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;
@@ -75,6 +79,7 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
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<>() {
};
@@ -86,6 +91,9 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
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());
@@ -114,6 +122,7 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
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))
@@ -121,7 +130,7 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
.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())
+ .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,
@@ -139,20 +148,35 @@ public class MsSqlCdcAdapter implements
StreamPipesAdapter, SupportsRuntimeConfi
metadataClient.validateSelectedTable(config);
this.temporalColumns = metadataClient.describeTemporalColumns(config);
this.configuredZoneId = config.getZoneId();
+ this.engineFailure = null;
+ this.connectorStarted = false;
+ this.stopRequested = false;
- Properties properties = buildDebeziumProperties(config);
+ 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();
@@ -177,6 +201,9 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
this.executorService = null;
this.temporalColumns = null;
this.configuredZoneId = null;
+ this.engineFailure = null;
+ this.connectorStarted = false;
+ this.stopRequested = false;
}
@Override
@@ -189,9 +216,9 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
.build();
}
- private Properties buildDebeziumProperties(MsSqlCdcAdapterConfig config) {
+ Properties buildDebeziumProperties(MsSqlCdcAdapterConfig config, String
elementId) {
Properties props = new Properties();
- String logicalName = createLogicalName(config);
+ 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");
@@ -204,6 +231,10 @@ public class MsSqlCdcAdapter implements
StreamPipesAdapter, SupportsRuntimeConfi
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()));
@@ -218,8 +249,10 @@ public class MsSqlCdcAdapter implements
StreamPipesAdapter, SupportsRuntimeConfi
return props;
}
- private String createLogicalName(MsSqlCdcAdapterConfig config) {
- return (config.getDatabase() + "_" + config.getTable())
+ 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();
@@ -261,7 +294,7 @@ public class MsSqlCdcAdapter implements StreamPipesAdapter,
SupportsRuntimeConfi
return normalized;
}
- private Object normalizeTemporalValue(Object value,
MsSqlMetadataClient.TemporalColumnMode temporalMode) {
+ Object normalizeTemporalValue(Object value,
MsSqlMetadataClient.TemporalColumnMode temporalMode) {
if (value == null) {
return null;
}
@@ -309,4 +342,75 @@ public class MsSqlCdcAdapter implements
StreamPipesAdapter, SupportsRuntimeConfi
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/MsSqlMetadataClient.java
b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/MsSqlMetadataClient.java
index 4d4d07d582..c2d581a5f0 100644
---
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
@@ -22,17 +22,27 @@ 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;
@@ -50,6 +60,8 @@ public class MsSqlMetadataClient {
+ "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);
@@ -95,9 +107,15 @@ public class MsSqlMetadataClient {
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)));
+ 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());
@@ -176,10 +194,41 @@ public class MsSqlMetadataClient {
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 sampleTemporalValue(temporalMode, zoneId);
+ return (double) sampleTemporalValue(temporalMode, zoneId);
}
return switch (descriptor.jdbcType()) {
@@ -191,6 +240,27 @@ public class MsSqlMetadataClient {
};
}
+ 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
? ""
@@ -247,6 +317,74 @@ public class MsSqlMetadataClient {
};
}
+ 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) {
}
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
index c73518e0f7..a54beb12af 100644
---
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
@@ -41,7 +41,7 @@ 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
+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) {
+ }
+ }
+}
