You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2026/08/10 15:18:16 UTC
(streampipes) 02/02: feat: add MSSQL table polling adapter
This is an automated email from the ASF dual-hosted git repository.
tenthe pushed a commit to branch 4813-add-pull-based-mssql-streaming-adapter
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 43ca109ee8be46b76a2b70757c1a555f5abd73b7
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Mon Aug 10 15:17:31 2026 +0000
feat: add MSSQL table polling adapter
---
.../transformer/js/PolyglotTypeConverter.java | 3 +
.../AdapterWorkerSampleDataRequestManagement.java | 7 +-
...apterWorkerSampleDataRequestManagementTest.java | 51 ++++
.../adapter/mssql/polling/CheckpointSnapshot.java | 33 ---
.../cdc/adapter/mssql/polling/CheckpointStore.java | 38 ---
.../adapter/mssql/polling/FileCheckpointStore.java | 226 ---------------
.../mssql/polling/MsSqlPollingSettings.java | 10 -
.../adapter/mssql/polling/MsSqlTablePoller.java | 66 +----
.../mssql/polling/MsSqlTablePollingAdapter.java | 58 +---
.../mssql/polling/MsSqlTablePollingClient.java | 19 +-
.../mssql/polling/MsSqlTablePollingConfig.java | 10 +-
.../adapter/mssql/polling/MsSqlValueConverter.java | 64 +++++
.../cdc/adapter/mssql/polling/StartupMode.java | 25 --
.../documentation.md | 27 +-
.../icon.png | Bin 0 -> 30045 bytes
.../strings.en | 6 +-
.../mssql/polling/FileCheckpointStoreTest.java | 68 -----
.../mssql/polling/InMemoryCheckpointStore.java | 70 -----
.../mssql/polling/MsSqlTablePollerTest.java | 173 +++--------
.../MsSqlTablePollingClientIntegrationTest.java | 54 +---
.../MsSqlTablePollingConfigurationTest.java | 24 +-
.../mssql/polling/MsSqlValueConverterTest.java | 66 +++++
streampipes-integration-tests/pom.xml | 6 +
.../MsSqlTablePollingAdapterIntegrationTest.java | 317 +++++++++++++++++++++
...untime-resolvable-oneof-input.component.spec.ts | 67 +++++
...tic-runtime-resolvable-oneof-input.component.ts | 1 +
26 files changed, 673 insertions(+), 816 deletions(-)
diff --git a/streampipes-connect-transformer-js/src/main/java/org/apache/streampipes/connect/transformer/js/PolyglotTypeConverter.java b/streampipes-connect-transformer-js/src/main/java/org/apache/streampipes/connect/transformer/js/PolyglotTypeConverter.java
index d083bf6ce1..b511948021 100644
--- a/streampipes-connect-transformer-js/src/main/java/org/apache/streampipes/connect/transformer/js/PolyglotTypeConverter.java
+++ b/streampipes-connect-transformer-js/src/main/java/org/apache/streampipes/connect/transformer/js/PolyglotTypeConverter.java
@@ -121,6 +121,9 @@ public final class PolyglotTypeConverter {
if (hostObject != null && hostObject.getClass().isArray()) {
return getArrayValue(hostObject);
}
+ if (hostObject instanceof String || hostObject instanceof Number || hostObject instanceof Boolean) {
+ return hostObject;
+ }
}
if (value.hasHashEntries()) {
Map<String, Object> result = new LinkedHashMap<>();
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagement.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagement.java
index 9a4ff5a02d..88619142f2 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagement.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagement.java
@@ -47,7 +47,12 @@ public class AdapterWorkerSampleDataRequestManagement {
}
public SampleData getSampleData(AdapterDescription adapterDescription) throws AdapterException, ParseException {
- return getOrCreateSampleDataManagement().getSampleData(adapterDescription);
+ try {
+ return getOrCreateSampleDataManagement().getSampleData(adapterDescription);
+ } catch (RuntimeException e) {
+ String cause = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+ throw new AdapterException("Failed to get adapter sample data: " + cause, e);
+ }
}
private AdapterWorkerSampleDataManagement getOrCreateSampleDataManagement() {
diff --git a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagementTest.java b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagementTest.java
new file mode 100644
index 0000000000..60055198dd
--- /dev/null
+++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerSampleDataRequestManagementTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.management.connect;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class AdapterWorkerSampleDataRequestManagementTest {
+
+ @Test
+ void uncheckedAdapterFailureIsReportedAsAdapterException() throws Exception {
+ AdapterDescription adapterDescription = new AdapterDescription();
+ IllegalArgumentException failure = new IllegalArgumentException("No value selected for sequence-column.");
+ AdapterWorkerSampleDataManagement sampleDataManagement = mock(AdapterWorkerSampleDataManagement.class);
+ when(sampleDataManagement.getSampleData(adapterDescription)).thenThrow(failure);
+ AdapterWorkerSampleDataRequestManagement requestManagement =
+ new AdapterWorkerSampleDataRequestManagement(sampleDataManagement);
+
+ AdapterException result = assertThrows(
+ AdapterException.class,
+ () -> requestManagement.getSampleData(adapterDescription)
+ );
+
+ assertEquals("Failed to get adapter sample data: No value selected for sequence-column.", result.getMessage());
+ assertSame(failure, result.getCause());
+ }
+}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointSnapshot.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointSnapshot.java
deleted file mode 100644
index 7f50408f7f..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointSnapshot.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.polling;
-
-import java.math.BigDecimal;
-import java.util.Optional;
-
-public record CheckpointSnapshot(boolean present, Optional<BigDecimal> cursor, long revision) {
-
- public static CheckpointSnapshot absent(long revision) {
- return new CheckpointSnapshot(false, Optional.empty(), revision);
- }
-
- public static CheckpointSnapshot present(Optional<BigDecimal> cursor, long revision) {
- return new CheckpointSnapshot(true, cursor, revision);
- }
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointStore.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointStore.java
deleted file mode 100644
index 5e3a4c9e3c..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/CheckpointStore.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.polling;
-
-import java.math.BigDecimal;
-import java.util.List;
-import java.util.Optional;
-
-public interface CheckpointStore {
-
- CheckpointSnapshot load(String adapterElementId) throws Exception;
-
- Optional<CheckpointSnapshot> save(String adapterElementId,
- long expectedRevision,
- Optional<BigDecimal> cursor) throws Exception;
-
- boolean delete(String adapterElementId, long expectedRevision) throws Exception;
-
- Optional<List<MsSqlColumn>> loadExpectedSchema(String adapterElementId) throws Exception;
-
- void saveExpectedSchema(String adapterElementId, List<MsSqlColumn> schema) throws Exception;
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStore.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStore.java
deleted file mode 100644
index e3ea217d59..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStore.java
+++ /dev/null
@@ -1,226 +0,0 @@
-/*
- * 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.polling;
-
-import org.apache.streampipes.commons.environment.Environments;
-import org.apache.streampipes.serializers.json.JacksonSerializer;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.nio.channels.FileChannel;
-import java.nio.channels.FileLock;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
-import java.nio.file.StandardOpenOption;
-import java.util.List;
-import java.util.Optional;
-
-public class FileCheckpointStore implements CheckpointStore {
-
- private static final ObjectMapper MAPPER = JacksonSerializer.getObjectMapper();
-
- private final Path baseDirectory;
-
- public FileCheckpointStore() {
- this(resolveDefaultDirectory());
- }
-
- public FileCheckpointStore(Path baseDirectory) {
- this.baseDirectory = baseDirectory;
- }
-
- @Override
- public CheckpointSnapshot load(String adapterElementId) throws IOException {
- return withLock(adapterElementId, () -> loadUnlocked(adapterElementId));
- }
-
- @Override
- public Optional<CheckpointSnapshot> save(String adapterElementId,
- long expectedRevision,
- Optional<BigDecimal> cursor) throws IOException {
- return withLock(adapterElementId, () -> {
- PersistedCheckpoint persisted = loadPersistedUnlocked(adapterElementId);
- CheckpointSnapshot current = persisted.toSnapshot();
- if (current.revision() != expectedRevision) {
- return Optional.empty();
- }
-
- CheckpointSnapshot saved = CheckpointSnapshot.present(cursor, expectedRevision + 1);
- persisted.apply(saved);
- writeAtomically(adapterElementId, persisted);
- return Optional.of(saved);
- });
- }
-
- @Override
- public boolean delete(String adapterElementId, long expectedRevision) throws IOException {
- return withLock(adapterElementId, () -> {
- PersistedCheckpoint persisted = loadPersistedUnlocked(adapterElementId);
- CheckpointSnapshot current = persisted.toSnapshot();
- if (current.revision() != expectedRevision) {
- return false;
- }
-
- persisted.apply(CheckpointSnapshot.absent(expectedRevision + 1));
- writeAtomically(adapterElementId, persisted);
- return true;
- });
- }
-
- @Override
- public Optional<List<MsSqlColumn>> loadExpectedSchema(String adapterElementId) throws IOException {
- return withLock(adapterElementId, () -> {
- Path path = schemaPath(adapterElementId);
- if (!Files.exists(path)) {
- return Optional.empty();
- }
- return Optional.of(List.of(MAPPER.readValue(path.toFile(), MsSqlColumn[].class)));
- });
- }
-
- @Override
- public void saveExpectedSchema(String adapterElementId, List<MsSqlColumn> schema) throws IOException {
- withLock(adapterElementId, () -> {
- writeAtomically(schemaPath(adapterElementId), List.copyOf(schema));
- return null;
- });
- }
-
- public Path statePath(String adapterElementId) {
- return baseDirectory.resolve(sanitizeAdapterElementId(adapterElementId) + ".json");
- }
-
- public Path schemaPath(String adapterElementId) {
- return baseDirectory.resolve(sanitizeAdapterElementId(adapterElementId) + ".schema.json");
- }
-
- private CheckpointSnapshot loadUnlocked(String adapterElementId) throws IOException {
- return loadPersistedUnlocked(adapterElementId).toSnapshot();
- }
-
- private PersistedCheckpoint loadPersistedUnlocked(String adapterElementId) throws IOException {
- Path path = statePath(adapterElementId);
- if (!Files.exists(path)) {
- return PersistedCheckpoint.from(CheckpointSnapshot.absent(0));
- }
-
- return MAPPER.readValue(path.toFile(), PersistedCheckpoint.class);
- }
-
- private void writeAtomically(String adapterElementId, PersistedCheckpoint checkpoint) throws IOException {
- writeAtomically(statePath(adapterElementId), checkpoint);
- }
-
- private void writeAtomically(Path target, Object value) throws IOException {
- Files.createDirectories(baseDirectory);
- Path temporary = Files.createTempFile(baseDirectory, target.getFileName().toString(), ".tmp");
- try {
- MAPPER.writeValue(temporary.toFile(), value);
- Files.move(temporary, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
- } finally {
- Files.deleteIfExists(temporary);
- }
- }
-
- private <T> T withLock(String adapterElementId, IoSupplier<T> operation) throws IOException {
- Files.createDirectories(baseDirectory);
- Path lockPath = baseDirectory.resolve(sanitizeAdapterElementId(adapterElementId) + ".lock");
- try (FileChannel channel = FileChannel.open(
- lockPath,
- StandardOpenOption.CREATE,
- StandardOpenOption.WRITE
- ); FileLock ignored = channel.lock()) {
- return operation.get();
- }
- }
-
- private String sanitizeAdapterElementId(String adapterElementId) {
- if (adapterElementId == null || adapterElementId.isBlank()) {
- throw new IllegalArgumentException("Adapter element ID is required for checkpoint persistence.");
- }
- return adapterElementId.replaceAll("[^A-Za-z0-9._-]", "_");
- }
-
- private static Path resolveDefaultDirectory() {
- String baseDir = Environments.getEnvironment()
- .getExtAssetBaseDir()
- .getValueOrReturn(System.getProperty("user.home"));
- return Path.of(baseDir, ".streampipes", "service", "mssql-table-polling-checkpoints");
- }
-
- @FunctionalInterface
- private interface IoSupplier<T> {
- T get() throws IOException;
- }
-
- public static class PersistedCheckpoint {
- private boolean present;
- private String cursor;
- private long revision;
-
- public PersistedCheckpoint() {
- }
-
- static PersistedCheckpoint from(CheckpointSnapshot snapshot) {
- PersistedCheckpoint checkpoint = new PersistedCheckpoint();
- checkpoint.present = snapshot.present();
- checkpoint.cursor = snapshot.cursor().map(BigDecimal::toPlainString).orElse(null);
- checkpoint.revision = snapshot.revision();
- return checkpoint;
- }
-
- void apply(CheckpointSnapshot snapshot) {
- this.present = snapshot.present();
- this.cursor = snapshot.cursor().map(BigDecimal::toPlainString).orElse(null);
- this.revision = snapshot.revision();
- }
-
- CheckpointSnapshot toSnapshot() {
- Optional<BigDecimal> parsedCursor = cursor == null ? Optional.empty() : Optional.of(new BigDecimal(cursor));
- return new CheckpointSnapshot(present, parsedCursor, revision);
- }
-
- public boolean isPresent() {
- return present;
- }
-
- public void setPresent(boolean present) {
- this.present = present;
- }
-
- public String getCursor() {
- return cursor;
- }
-
- public void setCursor(String cursor) {
- this.cursor = cursor;
- }
-
- public long getRevision() {
- return revision;
- }
-
- public void setRevision(long revision) {
- this.revision = revision;
- }
- }
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlPollingSettings.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlPollingSettings.java
index b7d9058f69..e205750e61 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlPollingSettings.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlPollingSettings.java
@@ -18,22 +18,12 @@
package org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling;
-import java.math.BigDecimal;
-
public record MsSqlPollingSettings(
- StartupMode startupMode,
- BigDecimal customSequence,
int batchSize,
int maxRowsPerPoll
) {
public MsSqlPollingSettings {
- if (startupMode == null) {
- throw new IllegalArgumentException("Startup mode is required.");
- }
- if (startupMode == StartupMode.CUSTOM_SEQUENCE && customSequence == null) {
- throw new IllegalArgumentException("A custom sequence is required for CUSTOM_SEQUENCE startup mode.");
- }
if (batchSize <= 0) {
throw new IllegalArgumentException("Batch size must be positive.");
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePoller.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePoller.java
index 69f0c6d291..0774b7e31e 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePoller.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePoller.java
@@ -28,39 +28,33 @@ import java.util.function.Consumer;
public class MsSqlTablePoller {
- private final String adapterElementId;
private final MsSqlPollingSettings settings;
private final List<MsSqlColumn> expectedSchema;
private final String sequenceColumn;
private final MsSqlPollingRowSource rowSource;
- private final CheckpointStore checkpointStore;
private final Consumer<Map<String, Object>> collector;
private final IExtensionsLogger logger;
+ private Optional<BigDecimal> cursor;
private String reportedSchemaMismatch;
- private String reportedRegression;
- public MsSqlTablePoller(String adapterElementId,
- MsSqlPollingSettings settings,
+ public MsSqlTablePoller(MsSqlPollingSettings settings,
List<MsSqlColumn> expectedSchema,
String sequenceColumn,
+ Optional<BigDecimal> startupCursor,
MsSqlPollingRowSource rowSource,
- CheckpointStore checkpointStore,
Consumer<Map<String, Object>> collector,
IExtensionsLogger logger) {
- this.adapterElementId = adapterElementId;
this.settings = settings;
this.expectedSchema = List.copyOf(expectedSchema);
this.sequenceColumn = sequenceColumn;
this.rowSource = rowSource;
- this.checkpointStore = checkpointStore;
this.collector = collector;
this.logger = logger;
+ this.cursor = startupCursor;
}
public void poll() throws Exception {
- CheckpointSnapshot checkpoint = checkpointStore.load(adapterElementId);
-
try (MsSqlPollingRowSource.PollSession session = rowSource.openSession()) {
String mismatch = describeSchemaMismatch(expectedSchema, session.currentSchema());
if (mismatch != null) {
@@ -69,43 +63,21 @@ public class MsSqlTablePoller {
}
reportSchemaRecovery();
- if (!checkpoint.present()) {
- checkpoint = initializeCheckpoint(checkpoint, session);
- if (checkpoint == null) {
- return;
- }
- }
-
- Optional<BigDecimal> maximum = session.maximumSequence();
- if (isRegression(checkpoint.cursor(), maximum)) {
- reportRegression(checkpoint.cursor().orElseThrow(), maximum);
- return;
- }
- reportedRegression = null;
-
int rowsProcessed = 0;
while (rowsProcessed < settings.maxRowsPerPoll()) {
int requested = Math.min(settings.batchSize(), settings.maxRowsPerPoll() - rowsProcessed);
- List<MsSqlRow> batch = session.readAfter(checkpoint.cursor(), requested);
+ List<MsSqlRow> batch = session.readAfter(cursor, requested);
if (batch.isEmpty()) {
return;
}
- validateBatch(batch, checkpoint.cursor());
+ validateBatch(batch, cursor);
for (MsSqlRow row : batch) {
collector.accept(row.event());
}
BigDecimal highestSequence = batch.get(batch.size() - 1).sequence();
- Optional<CheckpointSnapshot> saved = checkpointStore.save(
- adapterElementId,
- checkpoint.revision(),
- Optional.of(highestSequence)
- );
- if (saved.isEmpty()) {
- return;
- }
- checkpoint = saved.orElseThrow();
+ cursor = Optional.of(highestSequence);
rowsProcessed += batch.size();
if (batch.size() < requested) {
@@ -115,22 +87,6 @@ public class MsSqlTablePoller {
}
}
- private CheckpointSnapshot initializeCheckpoint(CheckpointSnapshot absent,
- MsSqlPollingRowSource.PollSession session) throws Exception {
- Optional<BigDecimal> cursor = switch (settings.startupMode()) {
- case NEW_ROWS -> session.maximumSequence();
- case ALL_EXISTING -> Optional.empty();
- case CUSTOM_SEQUENCE -> Optional.of(settings.customSequence());
- };
-
- return checkpointStore.save(adapterElementId, absent.revision(), cursor).orElse(null);
- }
-
- private boolean isRegression(Optional<BigDecimal> checkpoint, Optional<BigDecimal> maximum) {
- return checkpoint.isPresent()
- && (maximum.isEmpty() || maximum.orElseThrow().compareTo(checkpoint.orElseThrow()) < 0);
- }
-
private void validateBatch(List<MsSqlRow> batch, Optional<BigDecimal> cursor) {
BigDecimal previous = cursor.orElse(null);
for (MsSqlRow row : batch) {
@@ -173,12 +129,4 @@ public class MsSqlTablePoller {
}
}
- private void reportRegression(BigDecimal checkpoint, Optional<BigDecimal> maximum) {
- String details = "Persisted sequence " + checkpoint + " is greater than current table maximum "
- + maximum.map(BigDecimal::toPlainString).orElse("<empty>") + ". Reset polling state to recover.";
- if (!details.equals(reportedRegression)) {
- logger.error("MSSQL sequence regression: " + details, new IllegalStateException(details));
- reportedRegression = details;
- }
- }
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapter.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapter.java
index de08e09e63..1ee660aea3 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapter.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapter.java
@@ -40,15 +40,14 @@ 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 org.apache.streampipes.sdk.helpers.Options;
-import org.apache.streampipes.sdk.helpers.Tuple2;
+import java.math.BigDecimal;
import java.time.ZoneOffset;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.BATCH_SIZE_KEY;
-import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.CUSTOM_SEQUENCE_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.DATABASE_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.DEFAULT_BATCH_SIZE;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.DEFAULT_MAX_ROWS_PER_POLL;
@@ -59,7 +58,6 @@ import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.pol
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.POLLING_INTERVAL_SECONDS_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.PORT_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.SEQUENCE_COLUMN_KEY;
-import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.STARTUP_MODE_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TABLE_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TIMEZONE_KEY;
import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TRUST_SERVER_CERTIFICATE_KEY;
@@ -69,20 +67,10 @@ public class MsSqlTablePollingAdapter implements StreamPipesAdapter, IPullAdapte
public static final String ID = "org.apache.streampipes.connect.cdc.adapter.mssql.polling";
- private final CheckpointStore checkpointStore;
-
private PullAdapterScheduler scheduler;
private MsSqlTablePoller poller;
private int pollingIntervalSeconds;
- public MsSqlTablePollingAdapter() {
- this(new FileCheckpointStore());
- }
-
- MsSqlTablePollingAdapter(CheckpointStore checkpointStore) {
- this.checkpointStore = checkpointStore;
- }
-
@Override
public StaticProperty resolveConfiguration(String staticPropertyInternalName,
IStaticPropertyExtractor extractor) throws SpConfigurationException {
@@ -120,7 +108,7 @@ public class MsSqlTablePollingAdapter implements StreamPipesAdapter, IPullAdapte
return AdapterConfigurationBuilder.create(ID, 0, MsSqlTablePollingAdapter::new)
.withLocales(Locales.EN)
- .withAssets(ExtensionAssetType.DOCUMENTATION)
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.requiredTextParameter(Labels.withId(HOST_KEY))
.requiredIntegerParameter(Labels.withId(PORT_KEY), 1433)
.requiredTextParameter(Labels.withId(DATABASE_KEY))
@@ -131,15 +119,6 @@ public class MsSqlTablePollingAdapter implements StreamPipesAdapter, IPullAdapte
.requiredTextParameter(Labels.withId(TIMEZONE_KEY), ZoneOffset.UTC.getId())
.requiredSingleValueSelectionFromContainer(Labels.withId(TABLE_KEY), connectionDependencies)
.requiredSingleValueSelectionFromContainer(Labels.withId(SEQUENCE_COLUMN_KEY), sequenceDependencies)
- .requiredSingleValueSelection(
- Labels.withId(STARTUP_MODE_KEY),
- Options.from(
- new Tuple2<>("New rows only", StartupMode.NEW_ROWS.name()),
- new Tuple2<>("All existing rows", StartupMode.ALL_EXISTING.name()),
- new Tuple2<>("Custom sequence", StartupMode.CUSTOM_SEQUENCE.name())
- )
- )
- .requiredTextParameter(Labels.withId(CUSTOM_SEQUENCE_KEY), "0")
.requiredIntegerParameter(Labels.withId(POLLING_INTERVAL_SECONDS_KEY), 5)
.requiredIntegerParameter(Labels.withId(BATCH_SIZE_KEY), DEFAULT_BATCH_SIZE)
.requiredIntegerParameter(Labels.withId(MAX_ROWS_PER_POLL_KEY), DEFAULT_MAX_ROWS_PER_POLL)
@@ -162,31 +141,22 @@ public class MsSqlTablePollingAdapter implements StreamPipesAdapter, IPullAdapte
MsSqlTablePollingClient client = new MsSqlTablePollingClient(config);
client.validateConfiguration();
- List<MsSqlColumn> currentSchema = client.describeSchema();
- List<MsSqlColumn> expectedSchema;
- try {
- expectedSchema = checkpointStore.loadExpectedSchema(extractor.getAdapterDescription().getElementId())
- .orElseThrow(() -> new AdapterException(
- "No captured SQL Server schema is available. Request schema detection before starting the adapter."
- ));
- } catch (AdapterException e) {
- throw e;
+ List<MsSqlColumn> currentSchema;
+ Optional<BigDecimal> startupCursor;
+ try (MsSqlPollingRowSource.PollSession startupSession = client.openSession()) {
+ currentSchema = startupSession.currentSchema();
+ startupCursor = startupSession.maximumSequence();
} catch (Exception e) {
- throw new AdapterException("Failed to load the captured SQL Server schema: " + e.getMessage(), e);
- }
- String initialMismatch = MsSqlTablePoller.describeSchemaMismatch(expectedSchema, currentSchema);
- if (initialMismatch != null) {
- throw new AdapterException("SQL Server table schema changed since schema discovery: " + initialMismatch);
+ throw new AdapterException("Failed to capture the SQL Server startup watermark: " + e.getMessage(), e);
}
this.pollingIntervalSeconds = config.pollingIntervalSeconds();
this.poller = new MsSqlTablePoller(
- extractor.getAdapterDescription().getElementId(),
config.pollingSettings(),
- expectedSchema,
+ currentSchema,
config.sequenceColumn(),
+ startupCursor,
client,
- checkpointStore,
collector::collect,
adapterRuntimeContext.getLogger()
);
@@ -210,12 +180,6 @@ public class MsSqlTablePollingAdapter implements StreamPipesAdapter, IPullAdapte
MsSqlTablePollingConfig config = MsSqlTablePollingConfig.from(extractor.getStaticPropertyExtractor());
MsSqlTablePollingClient client = new MsSqlTablePollingClient(config);
client.validateConfiguration();
- List<MsSqlColumn> expectedSchema = client.describeSchema();
- try {
- checkpointStore.saveExpectedSchema(extractor.getAdapterDescription().getElementId(), expectedSchema);
- } catch (Exception e) {
- throw new AdapterException("Failed to persist the captured SQL Server schema: " + e.getMessage(), e);
- }
return SampleDataBuilder.create().sample(client.sampleRow()).build();
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClient.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClient.java
index fc239e161c..d770f21d55 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClient.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClient.java
@@ -69,7 +69,7 @@ public class MsSqlTablePollingClient implements MsSqlPollingRowSource {
+ "OR (ty.name IN ('decimal', 'numeric') AND c.scale = 0)) "
+ "AND EXISTS (SELECT 1 FROM sys.indexes i "
+ "JOIN sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id "
- + "WHERE i.object_id = t.object_id AND i.is_unique_constraint = 1 "
+ + "WHERE i.object_id = t.object_id AND (i.is_primary_key = 1 OR i.is_unique_constraint = 1) "
+ "AND ic.column_id = c.column_id AND ic.key_ordinal = 1 "
+ "AND NOT EXISTS (SELECT 1 FROM sys.index_columns other "
+ "WHERE other.object_id = i.object_id AND other.index_id = i.index_id AND other.key_ordinal > 1)) "
@@ -221,7 +221,7 @@ public class MsSqlTablePollingClient implements MsSqlPollingRowSource {
return columns;
}
- private Object convertValue(Object value, MsSqlColumn column) {
+ private Object convertValue(Object value, MsSqlColumn column) throws SQLException {
if (value == null) {
return null;
}
@@ -259,7 +259,7 @@ public class MsSqlTablePollingClient implements MsSqlPollingRowSource {
return value.toString();
}
}
- return value;
+ return MsSqlValueConverter.convert(value, column);
}
private boolean isDateTimeOffset(MsSqlColumn column) {
@@ -322,7 +322,7 @@ public class MsSqlTablePollingClient implements MsSqlPollingRowSource {
try (ResultSet resultSet = statement.executeQuery()) {
List<MsSqlRow> rows = new ArrayList<>();
ResultSetMetaData resultMetadata = resultSet.getMetaData();
- Map<String, MsSqlColumn> columnsByName = new LinkedHashMap<>();
+ Map<String, MsSqlColumn> columnsByName = new java.util.TreeMap<>(String.CASE_INSENSITIVE_ORDER);
schema.forEach(column -> columnsByName.put(column.name(), column));
while (resultSet.next()) {
Map<String, Object> event = new LinkedHashMap<>();
@@ -330,15 +330,20 @@ public class MsSqlTablePollingClient implements MsSqlPollingRowSource {
for (int index = 1; index <= resultMetadata.getColumnCount(); index++) {
String name = resultMetadata.getColumnLabel(index);
MsSqlColumn column = columnsByName.get(name);
+ if (column == null) {
+ throw new SQLException(
+ "Could not map SQL Server result column " + name + " to the captured table schema."
+ );
+ }
Object rawValue = resultSet.getObject(index);
- Object converted = column == null ? rawValue : convertValue(rawValue, column);
- if (name.equals(config.sequenceColumn())) {
+ Object converted = convertValue(rawValue, column);
+ if (column.name().equalsIgnoreCase(config.sequenceColumn())) {
rowSequence = exactDecimal(rawValue).orElseThrow(
() -> new SQLException("Sequence column contains null: " + config.sequenceColumn())
);
converted = rowSequence;
}
- event.put(name, converted);
+ event.put(column.name(), converted);
}
if (rowSequence == null) {
throw new SQLException("Sequence column is missing from query result: " + config.sequenceColumn());
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfig.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfig.java
index a13a9f8658..ee076dc56b 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfig.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfig.java
@@ -22,7 +22,6 @@ import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import org.apache.streampipes.model.staticproperty.Option;
import org.apache.streampipes.model.staticproperty.RuntimeResolvableOneOfStaticProperty;
-import java.math.BigDecimal;
import java.time.ZoneId;
import java.time.ZoneOffset;
@@ -37,8 +36,6 @@ public record MsSqlTablePollingConfig(
String timezone,
MsSqlTableIdentifier table,
String sequenceColumn,
- StartupMode startupMode,
- BigDecimal customSequence,
int pollingIntervalSeconds,
int batchSize,
int maxRowsPerPoll
@@ -54,8 +51,6 @@ public record MsSqlTablePollingConfig(
public static final String TIMEZONE_KEY = "timezone";
public static final String TABLE_KEY = "table";
public static final String SEQUENCE_COLUMN_KEY = "sequence-column";
- public static final String STARTUP_MODE_KEY = "startup-mode";
- public static final String CUSTOM_SEQUENCE_KEY = "custom-sequence";
public static final String POLLING_INTERVAL_SECONDS_KEY = "polling-interval-seconds";
public static final String BATCH_SIZE_KEY = "batch-size";
public static final String MAX_ROWS_PER_POLL_KEY = "max-rows-per-poll";
@@ -70,7 +65,6 @@ public record MsSqlTablePollingConfig(
public static MsSqlTablePollingConfig from(IStaticPropertyExtractor extractor,
boolean requireTable,
boolean requireSequenceColumn) {
- String customSequence = extractor.singleValueParameter(CUSTOM_SEQUENCE_KEY, String.class);
return new MsSqlTablePollingConfig(
extractor.singleValueParameter(HOST_KEY, String.class),
extractor.singleValueParameter(PORT_KEY, Integer.class),
@@ -82,8 +76,6 @@ public record MsSqlTablePollingConfig(
extractor.singleValueParameter(TIMEZONE_KEY, String.class),
selectedTable(extractor, requireTable),
selectedRuntimeValue(extractor, SEQUENCE_COLUMN_KEY, requireSequenceColumn),
- StartupMode.valueOf(extractor.selectedSingleValue(STARTUP_MODE_KEY, String.class)),
- customSequence == null || customSequence.isBlank() ? null : new BigDecimal(customSequence),
extractor.singleValueParameter(POLLING_INTERVAL_SECONDS_KEY, Integer.class),
extractor.singleValueParameter(BATCH_SIZE_KEY, Integer.class),
extractor.singleValueParameter(MAX_ROWS_PER_POLL_KEY, Integer.class)
@@ -91,7 +83,7 @@ public record MsSqlTablePollingConfig(
}
public MsSqlPollingSettings pollingSettings() {
- return new MsSqlPollingSettings(startupMode, customSequence, batchSize, maxRowsPerPoll);
+ return new MsSqlPollingSettings(batchSize, maxRowsPerPoll);
}
public ZoneId zoneId() {
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverter.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverter.java
new file mode 100644
index 0000000000..d2c5b27e87
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverter.java
@@ -0,0 +1,64 @@
+/*
+ * 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.polling;
+
+import java.math.BigDecimal;
+import java.sql.SQLException;
+import java.sql.Types;
+
+final class MsSqlValueConverter {
+
+ private MsSqlValueConverter() {
+ }
+
+ static Object convert(Object value, MsSqlColumn column) throws SQLException {
+ if (value == null) {
+ return null;
+ }
+
+ try {
+ return switch (column.jdbcType()) {
+ case Types.DECIMAL, Types.NUMERIC -> new BigDecimal(value.toString());
+ case Types.BIGINT -> new BigDecimal(value.toString()).longValueExact();
+ case Types.INTEGER -> new BigDecimal(value.toString()).intValueExact();
+ case Types.SMALLINT -> new BigDecimal(value.toString()).shortValueExact();
+ case Types.TINYINT -> new BigDecimal(value.toString()).intValueExact();
+ case Types.DOUBLE, Types.FLOAT, Types.REAL -> Double.parseDouble(value.toString());
+ case Types.BIT, Types.BOOLEAN -> Boolean.parseBoolean(value.toString());
+ default -> convertScalar(value, column);
+ };
+ } catch (NumberFormatException | ArithmeticException e) {
+ throw new SQLException(
+ "Could not convert SQL Server column " + column.name() + " with JDBC type " + column.typeName()
+ + " to a primitive value: " + value,
+ e
+ );
+ }
+ }
+
+ private static Object convertScalar(Object value, MsSqlColumn column) throws SQLException {
+ if (value instanceof String || value instanceof Number || value instanceof Boolean || value instanceof Character) {
+ return value;
+ }
+ throw new SQLException(
+ "Could not convert SQL Server column " + column.name() + " with JDBC type " + column.typeName()
+ + " to a primitive value. JDBC returned " + value.getClass().getName() + "."
+ );
+ }
+}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/StartupMode.java b/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/StartupMode.java
deleted file mode 100644
index 2ac4ca32a2..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/StartupMode.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.polling;
-
-public enum StartupMode {
- NEW_ROWS,
- ALL_EXISTING,
- CUSTOM_SEQUENCE
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/documentation.md b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/documentation.md
index 007d6c3588..ca99c484e8 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/documentation.md
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/documentation.md
@@ -29,22 +29,15 @@ order. Updates, deletes, views, joins, and late lower-sequence commits are not s
### Startup behavior
-* **New rows only** records the current maximum when state is absent.
-* **All existing rows** reads the table from its first sequence value.
-* **Custom sequence** reads rows whose sequence is strictly greater than the configured exact value.
+When the adapter starts, it records the current maximum sequence in memory. It emits only rows with a greater sequence
+value. Rows that exist before this watermark, including rows inserted while the adapter was stopped, are not replayed.
+The startup watermark is the activation boundary; it is captured before the polling scheduler starts.
-An existing checkpoint always wins. Deleting the adapter's checkpoint reapplies the configured startup mode.
+### Delivery
-### Delivery and state
-
-Rows are read using ordered keyset pagination and emitted individually. A checkpoint advances only after every row in a
-batch has been handed to the collector. A crash between handoff and checkpoint persistence can replay at most one batch,
-so delivery at this boundary is at least once.
-
-The default file-backed state is stored below the extension asset directory in
-`.streampipes/service/mssql-table-polling-checkpoints`. Each checkpoint JSON file exposes the exact cursor as a string and
-a revision; the separate `.schema.json` file preserves the schema captured during configuration. Delete the checkpoint
-JSON through the checkpoint store or while the adapter is stopped to reset cursor state safely.
+Rows are read using ordered keyset pagination and emitted individually. The in-memory cursor advances only after every
+row in a batch has been handed to the collector. A failed handoff retries that batch while the adapter remains running.
+Restarting the adapter captures a fresh watermark and intentionally provides no replay guarantee.
### Polling limits and schema changes
@@ -52,9 +45,9 @@ The batch size defaults to 500 and the maximum rows per poll defaults to 10,000.
allowed interval with `SP_MSSQL_POLLING_MIN_INTERVAL_SECONDS` (default: 1). Startup fails if the configured interval is
below this value.
-The output contains every source column, including the sequence, and no synthetic timestamp. Schema changes pause
-emission and checkpoint advancement until the original column order and JDBC types return. Sequence regression caused by
-truncation or reseeding also pauses emission and requires an explicit state reset.
+The output contains every source column, including the sequence, and no synthetic timestamp. Schema changes during an
+active run pause emission until the original column order and JDBC types return. A restart captures the current schema as
+the new baseline. Sequence regression caused by truncation or reseeding is not detected in this version.
SQL temporal values are emitted as epoch milliseconds and binary values as Base64 strings, matching the MSSQL connector
conventions. Each poll uses and closes one JDBC connection. Connection login and query execution have finite timeouts.
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/icon.png b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/icon.png
new file mode 100644
index 0000000000..ee3781f2f5
Binary files /dev/null and b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/icon.png differ
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/strings.en b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/strings.en
index 97b31eed4b..a5061fee0d 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/strings.en
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/main/resources/org.apache.streampipes.connect.cdc.adapter.mssql.polling/strings.en
@@ -38,13 +38,9 @@ table.title=Base Table
table.description=Ordinary user table to poll; views and system tables are excluded
sequence-column.title=Sequence Column
sequence-column.description=Non-null exact integer column backed by a single-column unique constraint
-startup-mode.title=Startup Mode
-startup-mode.description=Choose what the adapter emits when no checkpoint state exists
-custom-sequence.title=Custom Sequence
-custom-sequence.description=Exact sequence boundary used by Custom sequence mode; rows must be strictly greater
polling-interval-seconds.title=Polling Interval (seconds)
polling-interval-seconds.description=Fixed delay between polls
batch-size.title=Batch Size
-batch-size.description=Maximum rows read and checkpointed in one batch
+batch-size.description=Maximum rows read and handed to the collector in one batch
max-rows-per-poll.title=Maximum Rows per Poll
max-rows-per-poll.description=Maximum backlog rows handed off during one scheduled invocation
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStoreTest.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStoreTest.java
deleted file mode 100644
index 30f6a8c710..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/FileCheckpointStoreTest.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.polling;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-import java.math.BigDecimal;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.Optional;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-class FileCheckpointStoreTest {
-
- private static final java.util.List<MsSqlColumn> SCHEMA = java.util.List.of(
- new MsSqlColumn("id", java.sql.Types.BIGINT, "bigint", 19, 0, false)
- );
-
- @TempDir
- Path directory;
-
- @Test
- void preservesExactCursorAndRejectsSaveAfterDeletion() throws Exception {
- FileCheckpointStore store = new FileCheckpointStore(directory);
- BigDecimal cursor = new BigDecimal("99999999999999999999999999999999999999");
-
- CheckpointSnapshot saved = store.save("urn:adapter/one", 0, Optional.of(cursor)).orElseThrow();
- assertEquals(cursor, store.load("urn:adapter/one").cursor().orElseThrow());
- assertTrue(Files.readString(directory.resolve("urn_adapter_one.json")).contains(cursor.toPlainString()));
- assertTrue(store.delete("urn:adapter/one", saved.revision()));
- assertFalse(store.save("urn:adapter/one", saved.revision(), Optional.of(BigDecimal.TEN)).isPresent());
- assertFalse(store.load("urn:adapter/one").present());
-
- assertEquals(saved.revision() + 1, store.load("urn:adapter/one").revision());
- }
-
- @Test
- void preservesGuessedSchemaAcrossStoreInstancesAndCheckpointChanges() throws Exception {
- FileCheckpointStore first = new FileCheckpointStore(directory);
- first.saveExpectedSchema("adapter-1", SCHEMA);
- CheckpointSnapshot saved = first.save("adapter-1", 0, Optional.of(BigDecimal.TEN)).orElseThrow();
- assertTrue(first.delete("adapter-1", saved.revision()));
- Files.delete(first.statePath("adapter-1"));
-
- FileCheckpointStore restarted = new FileCheckpointStore(directory);
- assertEquals(SCHEMA, restarted.loadExpectedSchema("adapter-1").orElseThrow());
- }
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/InMemoryCheckpointStore.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/InMemoryCheckpointStore.java
deleted file mode 100644
index 39c75f5ce6..0000000000
--- a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/InMemoryCheckpointStore.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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.polling;
-
-import java.math.BigDecimal;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-
-class InMemoryCheckpointStore implements CheckpointStore {
-
- private final Map<String, CheckpointSnapshot> snapshots = new HashMap<>();
- private final Map<String, java.util.List<MsSqlColumn>> schemas = new HashMap<>();
-
- @Override
- public synchronized CheckpointSnapshot load(String adapterElementId) {
- return snapshots.getOrDefault(adapterElementId, CheckpointSnapshot.absent(0));
- }
-
- @Override
- public synchronized Optional<CheckpointSnapshot> save(String adapterElementId,
- long expectedRevision,
- Optional<BigDecimal> cursor) {
- CheckpointSnapshot current = load(adapterElementId);
- if (current.revision() != expectedRevision) {
- return Optional.empty();
- }
-
- CheckpointSnapshot saved = CheckpointSnapshot.present(cursor, expectedRevision + 1);
- snapshots.put(adapterElementId, saved);
- return Optional.of(saved);
- }
-
- @Override
- public synchronized boolean delete(String adapterElementId, long expectedRevision) {
- CheckpointSnapshot current = load(adapterElementId);
- if (current.revision() != expectedRevision) {
- return false;
- }
-
- snapshots.put(adapterElementId, CheckpointSnapshot.absent(expectedRevision + 1));
- return true;
- }
-
- @Override
- public synchronized Optional<java.util.List<MsSqlColumn>> loadExpectedSchema(String adapterElementId) {
- return Optional.ofNullable(schemas.get(adapterElementId)).map(java.util.List::copyOf);
- }
-
- @Override
- public synchronized void saveExpectedSchema(String adapterElementId, java.util.List<MsSqlColumn> schema) {
- schemas.put(adapterElementId, java.util.List.copyOf(schema));
- }
-}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollerTest.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollerTest.java
index b700dd8b23..cbee2d208a 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollerTest.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollerTest.java
@@ -33,9 +33,7 @@ import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
class MsSqlTablePollerTest {
@@ -45,83 +43,36 @@ class MsSqlTablePollerTest {
);
@Test
- void allExistingEmitsRowsAndCheckpointsCompletedBatch() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
- TestRowSource source = new TestRowSource(List.of(
- row("9223372036854775808", "first"),
- row("9223372036854775809", "second")
- ));
- List<Map<String, Object>> emitted = new ArrayList<>();
- MsSqlTablePoller poller = new MsSqlTablePoller(
- "adapter-1",
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 2, 10),
- SCHEMA,
- "sequence_id",
- source,
- store,
- emitted::add,
- new TestLogger()
- );
-
- poller.poll();
-
- assertEquals(List.of("first", "second"), emitted.stream().map(event -> event.get("value")).toList());
- assertEquals(
- new BigDecimal("9223372036854775809"),
- store.load("adapter-1").cursor().orElseThrow()
- );
- assertEquals(1, source.openCount);
- assertEquals(1, source.closeCount);
- }
-
- @Test
- void newRowsStartsAtCurrentMaximumAndReadsRowsAddedLater() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
+ void emitsOnlyRowsAddedAfterStartupWatermark() throws Exception {
TestRowSource source = new TestRowSource(new ArrayList<>(List.of(row("10", "existing"))));
List<Map<String, Object>> emitted = new ArrayList<>();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.NEW_ROWS, null, 2, 10), source, store, emitted::add, new TestLogger()
- );
+ MsSqlTablePoller poller = poller(Optional.of(new BigDecimal("10")), source, emitted::add);
- poller.poll();
source.rows.add(row("11", "new"));
poller.poll();
- assertEquals(List.of("new"), emitted.stream().map(event -> event.get("value")).toList());
- assertEquals(new BigDecimal("11"), store.load("adapter-1").cursor().orElseThrow());
+ assertEquals(List.of("new"), values(emitted));
}
@Test
- void customSequenceIsExclusiveAndExactAboveLongRange() throws Exception {
- BigDecimal custom = new BigDecimal("9223372036854775808");
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
- TestRowSource source = new TestRowSource(List.of(
- row(custom.toPlainString(), "boundary"),
- row("9223372036854775809", "after")
- ));
+ void emptyTableAtStartupEmitsItsFirstRow() throws Exception {
+ TestRowSource source = new TestRowSource(new ArrayList<>());
List<Map<String, Object>> emitted = new ArrayList<>();
+ MsSqlTablePoller poller = poller(Optional.empty(), source, emitted::add);
- poller(
- new MsSqlPollingSettings(StartupMode.CUSTOM_SEQUENCE, custom, 1, 10),
- source,
- store,
- emitted::add,
- new TestLogger()
- ).poll();
+ source.rows.add(row("1", "first"));
+ poller.poll();
- assertEquals(List.of("after"), emitted.stream().map(event -> event.get("value")).toList());
+ assertEquals(List.of("first"), values(emitted));
}
@Test
- void maxRowsPerPollContinuesBacklogOnNextInvocation() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
+ void advancesInMemoryCursorAcrossBatchesAndPolls() throws Exception {
TestRowSource source = new TestRowSource(List.of(
row("1", "one"), row("2", "two"), row("3", "three"), row("4", "four"), row("5", "five")
));
List<Map<String, Object>> emitted = new ArrayList<>();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 2, 4), source, store, emitted::add, new TestLogger()
- );
+ MsSqlTablePoller poller = poller(Optional.empty(), new MsSqlPollingSettings(2, 4), source, emitted::add);
poller.poll();
assertEquals(List.of("one", "two", "three", "four"), values(emitted));
@@ -129,42 +80,33 @@ class MsSqlTablePollerTest {
assertEquals(List.of("one", "two", "three", "four", "five"), values(emitted));
assertEquals(2, source.openCount);
+ assertEquals(2, source.closeCount);
}
@Test
- void collectorFailureReplaysWholeUncheckpointedBatch() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
+ void collectorFailureDoesNotAdvanceTheInMemoryCursor() throws Exception {
TestRowSource source = new TestRowSource(List.of(row("1", "one"), row("2", "two")));
AtomicBoolean failOnce = new AtomicBoolean(true);
List<Map<String, Object>> emitted = new ArrayList<>();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 2, 10),
- source,
- store,
- event -> {
- emitted.add(event);
- if (failOnce.compareAndSet(true, false)) {
- throw new IllegalStateException("collector failed");
- }
- },
- new TestLogger()
- );
+ MsSqlTablePoller poller = poller(Optional.empty(), source, event -> {
+ emitted.add(event);
+ if (failOnce.compareAndSet(true, false)) {
+ throw new IllegalStateException("collector failed");
+ }
+ });
assertThrows(IllegalStateException.class, poller::poll);
- assertTrue(store.load("adapter-1").cursor().isEmpty());
poller.poll();
assertEquals(List.of("one", "one", "two"), values(emitted));
- assertEquals(new BigDecimal("2"), store.load("adapter-1").cursor().orElseThrow());
}
@Test
void schemaMismatchIsSuppressedUntilDistinctAndReportsRecovery() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
TestRowSource source = new TestRowSource(List.of(row("1", "one")));
TestLogger logger = new TestLogger();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 1, 10), source, store, event -> { }, logger
+ MsSqlTablePoller poller = new MsSqlTablePoller(
+ new MsSqlPollingSettings(1, 10), SCHEMA, "sequence_id", Optional.empty(), source, event -> { }, logger
);
source.schema = List.of(SCHEMA.get(0));
@@ -179,60 +121,18 @@ class MsSqlTablePollerTest {
assertEquals(1, logger.infos.size());
}
- @Test
- void sequenceRegressionPausesAndPreservesCheckpoint() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
- CheckpointSnapshot saved = store.save("adapter-1", 0, Optional.of(new BigDecimal("100"))).orElseThrow();
- TestRowSource source = new TestRowSource(List.of(row("10", "old")));
- TestLogger logger = new TestLogger();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 1, 10), source, store, event -> { }, logger
- );
-
- poller.poll();
- poller.poll();
-
- assertEquals(1, logger.errors.size());
- assertEquals(saved, store.load("adapter-1"));
- }
-
- @Test
- void deletionRejectsStaleSaveAndReappliesStartupMode() throws Exception {
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
- TestRowSource source = new TestRowSource(List.of(row("1", "one")));
- List<Map<String, Object>> emitted = new ArrayList<>();
- AtomicBoolean deleted = new AtomicBoolean();
- MsSqlTablePoller poller = poller(
- new MsSqlPollingSettings(StartupMode.NEW_ROWS, null, 1, 10),
- source,
- store,
- event -> {
- emitted.add(event);
- if (deleted.compareAndSet(false, true)) {
- CheckpointSnapshot current = store.load("adapter-1");
- assertTrue(store.delete("adapter-1", current.revision()));
- }
- },
- new TestLogger()
- );
- CheckpointSnapshot initial = store.save("adapter-1", 0, Optional.empty()).orElseThrow();
- assertTrue(initial.present());
-
- poller.poll();
- assertFalse(store.load("adapter-1").present());
- poller.poll();
-
- assertEquals(List.of("one"), values(emitted));
- assertEquals(new BigDecimal("1"), store.load("adapter-1").cursor().orElseThrow());
+ private static MsSqlTablePoller poller(Optional<BigDecimal> startupCursor,
+ TestRowSource source,
+ java.util.function.Consumer<Map<String, Object>> collector) {
+ return poller(startupCursor, new MsSqlPollingSettings(2, 10), source, collector);
}
- private static MsSqlTablePoller poller(MsSqlPollingSettings settings,
- MsSqlPollingRowSource source,
- CheckpointStore store,
- java.util.function.Consumer<Map<String, Object>> collector,
- IExtensionsLogger logger) {
+ private static MsSqlTablePoller poller(Optional<BigDecimal> startupCursor,
+ MsSqlPollingSettings settings,
+ TestRowSource source,
+ java.util.function.Consumer<Map<String, Object>> collector) {
return new MsSqlTablePoller(
- "adapter-1", settings, SCHEMA, "sequence_id", source, store, collector, logger
+ settings, SCHEMA, "sequence_id", startupCursor, source, collector, new TestLogger()
);
}
@@ -296,21 +196,22 @@ class MsSqlTablePollerTest {
}
@Override
- public void error(Exception e) {
+ public void error(Exception exception) {
+ errors.add(exception.getMessage());
}
@Override
- public void error(String details, Exception e) {
- errors.add(details);
+ public void error(String title, Exception exception) {
+ errors.add(title + exception.getMessage());
}
@Override
- public void info(String title, String details) {
- infos.add(details);
+ public void info(String title, String description) {
+ infos.add(title + description);
}
@Override
- public void warn(String title, String details) {
+ public void warn(String title, String description) {
}
}
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClientIntegrationTest.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClientIntegrationTest.java
index 3cb23fda01..38a67c1192 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClientIntegrationTest.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingClientIntegrationTest.java
@@ -18,9 +18,6 @@
package org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling;
-import org.apache.streampipes.extensions.api.monitoring.IExtensionsLogger;
-import org.apache.streampipes.model.monitoring.SpLogMessage;
-
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeEach;
@@ -31,9 +28,7 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.sql.Types;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.UUID;
@@ -62,6 +57,7 @@ class MsSqlTablePollingClientIntegrationTest {
Statement statement = connection.createStatement()) {
statement.execute("CREATE SCHEMA " + MsSqlTablePollingClient.quoteIdentifier(table.schema()));
statement.execute("CREATE TABLE " + table.displayName() + " ("
+ + "primary_sequence BIGINT IDENTITY PRIMARY KEY, "
+ "sequence_id DECIMAL(38, 0) NOT NULL, "
+ "recorded_at DATETIME2(3) NULL, "
+ "payload VARBINARY(16) NULL, "
@@ -96,7 +92,10 @@ class MsSqlTablePollingClientIntegrationTest {
assertTrue(client.discoverTables().stream().anyMatch(option -> table.encode().equals(option.getInternalName())));
assertFalse(client.discoverTables().stream().anyMatch(option -> option.getName().contains(viewName)));
- assertEquals(List.of("sequence_id"), client.discoverSequenceColumns().stream().map(option -> option.getName()).toList());
+ assertEquals(
+ List.of("primary_sequence", "sequence_id"),
+ client.discoverSequenceColumns().stream().map(option -> option.getName()).toList()
+ );
assertEquals(BigDecimal.ONE, client.sampleRow().get("sequence_id"));
insertRows(client);
@@ -115,22 +114,6 @@ class MsSqlTablePollingClientIntegrationTest {
assertEquals("BAUG", rows.get(0).event().get("payload"));
}
- InMemoryCheckpointStore store = new InMemoryCheckpointStore();
- store.save("adapter-live", 0, Optional.of(new BigDecimal("99999999999999999999999999999999999999")));
- TestLogger logger = new TestLogger();
- List<Map<String, Object>> emitted = new ArrayList<>();
- new MsSqlTablePoller(
- "adapter-live",
- new MsSqlPollingSettings(StartupMode.ALL_EXISTING, null, 10, 10),
- schema,
- "sequence_id",
- client,
- store,
- emitted::add,
- logger
- ).poll();
- assertTrue(emitted.isEmpty());
- assertEquals(1, logger.errors);
}
private void insertRows(MsSqlTablePollingClient client) throws Exception {
@@ -162,8 +145,6 @@ class MsSqlTablePollingClientIntegrationTest {
"UTC",
selectedTable,
"sequence_id",
- StartupMode.ALL_EXISTING,
- null,
1,
10,
10
@@ -175,29 +156,4 @@ class MsSqlTablePollingClientIntegrationTest {
return value == null ? defaultValue : value;
}
- private static class TestLogger implements IExtensionsLogger {
- private int errors;
-
- @Override
- public void log(SpLogMessage logMessage) {
- }
-
- @Override
- public void error(Exception e) {
- errors++;
- }
-
- @Override
- public void error(String details, Exception e) {
- errors++;
- }
-
- @Override
- public void info(String title, String details) {
- }
-
- @Override
- public void warn(String title, String details) {
- }
- }
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfigurationTest.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfigurationTest.java
index 973f394d81..3aa87f833c 100644
--- a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfigurationTest.java
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingConfigurationTest.java
@@ -20,8 +20,6 @@ package org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling;
import org.junit.jupiter.api.Test;
-import java.math.BigDecimal;
-
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -30,27 +28,24 @@ class MsSqlTablePollingConfigurationTest {
@Test
void intervalAtAndAboveAdministrativeMinimumIsAccepted() {
- assertDoesNotThrow(() -> config(2, 10, 10, StartupMode.NEW_ROWS, null).validate(2));
- assertDoesNotThrow(() -> config(3, 10, 10, StartupMode.NEW_ROWS, null).validate(2));
+ assertDoesNotThrow(() -> config(2, 10, 10).validate(2));
+ assertDoesNotThrow(() -> config(3, 10, 10).validate(2));
}
@Test
void intervalBelowAdministrativeMinimumIsRejected() {
assertThrows(IllegalArgumentException.class, () ->
- config(1, 10, 10, StartupMode.NEW_ROWS, null).validate(2)
+ config(1, 10, 10).validate(2)
);
}
@Test
- void invalidBatchLimitsAndMissingCustomCursorAreRejected() {
- assertThrows(IllegalArgumentException.class, () ->
- config(2, 0, 10, StartupMode.NEW_ROWS, null).validate(1)
- );
+ void invalidBatchLimitsAreRejected() {
assertThrows(IllegalArgumentException.class, () ->
- config(2, 11, 10, StartupMode.NEW_ROWS, null).validate(1)
+ config(2, 0, 10).validate(1)
);
assertThrows(IllegalArgumentException.class, () ->
- config(2, 10, 10, StartupMode.CUSTOM_SEQUENCE, null).validate(1)
+ config(2, 11, 10).validate(1)
);
}
@@ -77,9 +72,7 @@ class MsSqlTablePollingConfigurationTest {
private static MsSqlTablePollingConfig config(int interval,
int batchSize,
- int maxRows,
- StartupMode startupMode,
- BigDecimal customSequence) {
+ int maxRows) {
return new MsSqlTablePollingConfig(
"localhost",
1433,
@@ -91,11 +84,10 @@ class MsSqlTablePollingConfigurationTest {
"UTC",
new MsSqlTableIdentifier("dbo", "measurements"),
"sequence_id",
- startupMode,
- customSequence,
interval,
batchSize,
maxRows
);
}
+
}
diff --git a/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverterTest.java b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverterTest.java
new file mode 100644
index 0000000000..85e4636bbe
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-cdc/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlValueConverterTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.polling;
+
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class MsSqlValueConverterTest {
+
+ @Test
+ void normalizesDriverSpecificDecimalValuesToJsonNumbers() throws Exception {
+ MsSqlColumn column = new MsSqlColumn("MENGE", Types.DECIMAL, "decimal", 18, 3, false);
+
+ Object converted = MsSqlValueConverter.convert(new NumericValue("108.000"), column);
+
+ assertEquals(new BigDecimal("108.000"), converted);
+ var jsonValue = JacksonSerializer.getObjectMapper()
+ .readTree(JacksonSerializer.getObjectMapper().writeValueAsString(Map.of("MENGE", converted)))
+ .get("MENGE");
+ assertTrue(jsonValue.isNumber());
+ assertEquals(0, new BigDecimal("108.000").compareTo(jsonValue.decimalValue()));
+ }
+
+ @Test
+ void rejectsUnsupportedJdbcObjectsWithColumnAndJavaType() {
+ MsSqlColumn column = new MsSqlColumn("MENGE", Types.OTHER, "sql_variant", 0, 0, false);
+
+ var exception = assertThrows(SQLException.class, () -> MsSqlValueConverter.convert(new Object(), column));
+
+ assertTrue(exception.getMessage().contains("MENGE"));
+ assertTrue(exception.getMessage().contains(Object.class.getName()));
+ }
+
+ private record NumericValue(String value) {
+ @Override
+ public String toString() {
+ return value;
+ }
+ }
+}
diff --git a/streampipes-integration-tests/pom.xml b/streampipes-integration-tests/pom.xml
index b392d2875e..67f3015579 100644
--- a/streampipes-integration-tests/pom.xml
+++ b/streampipes-integration-tests/pom.xml
@@ -105,6 +105,12 @@
<version>0.99.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-connectors-cdc</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-pipeline-management</artifactId>
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapterIntegrationTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapterIntegrationTest.java
new file mode 100644
index 0000000000..9c18f5de32
--- /dev/null
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/extensions/connectors/cdc/adapter/mssql/polling/MsSqlTablePollingAdapterIntegrationTest.java
@@ -0,0 +1,317 @@
+/*
+ * 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.polling;
+
+import org.apache.streampipes.connect.transformer.api.TransformationEngines;
+import org.apache.streampipes.connect.transformer.js.GraalJsScriptEngine;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
+import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
+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.management.connect.adapter.model.EventCollector;
+import org.apache.streampipes.extensions.management.connect.adapter.model.pipeline.AdapterPipeline;
+import org.apache.streampipes.model.connect.TransformationConfig;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.staticproperty.Option;
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableOneOfStaticProperty;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.BATCH_SIZE_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.DATABASE_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.ENCRYPT_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.HOST_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.MAX_ROWS_PER_POLL_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.PASSWORD_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.POLLING_INTERVAL_SECONDS_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.PORT_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.SEQUENCE_COLUMN_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TABLE_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TIMEZONE_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.TRUST_SERVER_CERTIFICATE_KEY;
+import static org.apache.streampipes.extensions.connectors.cdc.adapter.mssql.polling.MsSqlTablePollingConfig.USERNAME_KEY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@EnabledIfEnvironmentVariable(named = "SP_TEST_MSSQL_HOST", matches = ".+")
+class MsSqlTablePollingAdapterIntegrationTest {
+
+ private static final String DATABASE = valueOrDefault("SP_TEST_MSSQL_DATABASE", "streampipes_demo");
+ private static final String USERNAME = valueOrDefault("SP_TEST_MSSQL_USERNAME", "sa");
+ private static final String PASSWORD = valueOrDefault("SP_TEST_MSSQL_PASSWORD", "");
+ private static final int PORT = Integer.parseInt(valueOrDefault("SP_TEST_MSSQL_PORT", "1433"));
+
+ @Test
+ void previewAndPullEmitTheSameScalarTypesForRowsAddedAfterStartup() throws Exception {
+ String tableName = "sp_poll_it_" + UUID.randomUUID().toString().replace("-", "");
+ String adapterId = "sp:mssql-poll-it:" + UUID.randomUUID();
+ IAdapterParameterExtractor extractor = extractor(tableName, adapterId);
+ IAdapterRuntimeContext runtimeContext = runtimeContext();
+ List<Map<String, Object>> events = new CopyOnWriteArrayList<>();
+
+ createTable(tableName);
+ try {
+ insertRow(tableName, 10L, 537_190_230L, new BigDecimal("108.000"), "first");
+ insertRow(tableName, 11L, 537_190_234L, new BigDecimal("122.000"), "second");
+
+ MsSqlTablePollingAdapter previewAdapter = new MsSqlTablePollingAdapter();
+ Map<String, Object> preview = previewAdapter.onSampleDataRequested(
+ extractor,
+ mock(IAdapterGuessSchemaContext.class)
+ ).getSamples().get(0);
+ assertScalarTypes(preview);
+
+ MsSqlTablePollingAdapter firstRun = new MsSqlTablePollingAdapter();
+ try {
+ firstRun.onAdapterStarted(extractor, event -> events.add(new HashMap<>(event)), runtimeContext);
+ insertRow(tableName, 12L, 537_190_229L, new BigDecimal("254.000"), "after-startup");
+ awaitEventCount(events, 1);
+ } finally {
+ firstRun.onAdapterStopped(extractor, runtimeContext);
+ }
+
+ assertEquals(
+ List.of(new BigDecimal("12")),
+ events.stream().map(event -> event.get("poll_sequence")).toList()
+ );
+ events.forEach(this::assertScalarTypes);
+ events.forEach(this::assertJsonContainsNumbers);
+
+ insertRow(tableName, 13L, 537_190_229L, new BigDecimal("254.000"), "before-restart");
+ events.clear();
+
+ MsSqlTablePollingAdapter restarted = new MsSqlTablePollingAdapter();
+ try {
+ restarted.onAdapterStarted(extractor, event -> events.add(new HashMap<>(event)), runtimeContext);
+ insertRow(tableName, 14L, 537_190_229L, new BigDecimal("255.000"), "after-restart");
+ awaitEventCount(events, 1);
+ } finally {
+ restarted.onAdapterStopped(extractor, runtimeContext);
+ }
+
+ assertEquals(new BigDecimal("14"), events.get(0).get("poll_sequence"));
+ assertEquals("after-restart", events.get(0).get("event_note"));
+ assertFalse(events.get(0).values().stream().anyMatch(Map.class::isInstance));
+ } finally {
+ dropTable(tableName);
+ }
+ }
+
+ @Test
+ void scriptedRuntimePipelineKeepsExactNumericColumnsAsJsonNumbers() throws Exception {
+ TransformationEngines.INSTANCE.registerEngine(GraalJsScriptEngine::new);
+ String tableName = "sp_poll_script_it_" + UUID.randomUUID().toString().replace("-", "");
+ String adapterId = "sp:mssql-poll-script-it:" + UUID.randomUUID();
+ IAdapterParameterExtractor extractor = extractor(tableName, adapterId);
+ IAdapterRuntimeContext runtimeContext = runtimeContext();
+ List<byte[]> serializedEvents = new CopyOnWriteArrayList<>();
+
+ createTable(tableName);
+ try {
+ TransformationConfig transformationConfig = TransformationConfig.withDefaultScript();
+ transformationConfig.setScriptActive(true);
+ extractor.getAdapterDescription().setTransformationConfig(transformationConfig);
+
+ IAdapterPipelineElement serializationSink = event -> {
+ serializedEvents.add(new JsonDataFormatDefinition().fromMap(event));
+ return event;
+ };
+ AdapterPipeline pipeline = new AdapterPipeline(
+ List.of(),
+ transformationConfig,
+ serializationSink,
+ null,
+ new EventSchema()
+ );
+ EventCollector collector = new EventCollector(pipeline, runtimeContext);
+ MsSqlTablePollingAdapter adapter = new MsSqlTablePollingAdapter();
+ try {
+ adapter.onAdapterStarted(extractor, collector, runtimeContext);
+ insertRow(tableName, 21L, 734_820_116L, new BigDecimal("73.250"), "scripted-event");
+ awaitSerializedEventCount(serializedEvents, 1);
+ } finally {
+ adapter.onAdapterStopped(extractor, runtimeContext);
+ collector.close();
+ }
+
+ var root = JacksonSerializer.getObjectMapper().readTree(serializedEvents.get(0));
+ assertTrue(root.get("poll_sequence").isIntegralNumber());
+ assertTrue(root.get("asset_reference").isIntegralNumber());
+ assertTrue(root.get("measurement_amount").isNumber());
+ assertFalse(root.get("poll_sequence").isObject());
+ assertFalse(root.get("measurement_amount").isObject());
+ } finally {
+ dropTable(tableName);
+ }
+ }
+
+ private void assertScalarTypes(Map<String, Object> event) {
+ assertInstanceOf(BigDecimal.class, event.get("poll_sequence"));
+ assertInstanceOf(Long.class, event.get("asset_reference"));
+ assertEquals(
+ new BigDecimal("108.000").scale(),
+ assertInstanceOf(BigDecimal.class, event.get("measurement_amount")).scale()
+ );
+ assertInstanceOf(Long.class, event.get("captured_at"));
+ assertInstanceOf(String.class, event.get("event_note"));
+ }
+
+ private void assertJsonContainsNumbers(Map<String, Object> event) {
+ byte[] json = new JsonDataFormatDefinition().fromMap(event);
+ try {
+ var root = JacksonSerializer.getObjectMapper().readTree(json);
+ assertTrue(root.get("poll_sequence").isIntegralNumber());
+ assertTrue(root.get("asset_reference").isIntegralNumber());
+ assertTrue(root.get("measurement_amount").isNumber());
+ assertFalse(root.get("poll_sequence").isObject());
+ assertFalse(root.get("measurement_amount").isObject());
+ } catch (Exception e) {
+ throw new AssertionError("Could not inspect serialized adapter event", e);
+ }
+ }
+
+ private void awaitEventCount(List<Map<String, Object>> events, int expected) throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (events.size() < expected && System.nanoTime() < deadline) {
+ Thread.sleep(50);
+ }
+ assertEquals(expected, events.size(), "Timed out waiting for MSSQL polling events");
+ }
+
+ private void awaitSerializedEventCount(List<byte[]> events, int expected) throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (events.size() < expected && System.nanoTime() < deadline) {
+ Thread.sleep(50);
+ }
+ assertEquals(expected, events.size(), "Timed out waiting for serialized MSSQL polling events");
+ }
+
+ private IAdapterParameterExtractor extractor(String tableName, String adapterId) {
+ IStaticPropertyExtractor properties = mock(IStaticPropertyExtractor.class);
+ when(properties.singleValueParameter(HOST_KEY, String.class)).thenReturn(System.getenv("SP_TEST_MSSQL_HOST"));
+ when(properties.singleValueParameter(PORT_KEY, Integer.class)).thenReturn(PORT);
+ when(properties.singleValueParameter(DATABASE_KEY, String.class)).thenReturn(DATABASE);
+ when(properties.singleValueParameter(USERNAME_KEY, String.class)).thenReturn(USERNAME);
+ when(properties.secretValue(PASSWORD_KEY)).thenReturn(PASSWORD);
+ when(properties.slideToggleValue(ENCRYPT_KEY)).thenReturn(true);
+ when(properties.slideToggleValue(TRUST_SERVER_CERTIFICATE_KEY)).thenReturn(true);
+ when(properties.singleValueParameter(TIMEZONE_KEY, String.class)).thenReturn("UTC");
+ when(properties.getStaticPropertyByName(TABLE_KEY, RuntimeResolvableOneOfStaticProperty.class))
+ .thenReturn(selectedRuntimeProperty("dbo." + tableName));
+ when(properties.getStaticPropertyByName(SEQUENCE_COLUMN_KEY, RuntimeResolvableOneOfStaticProperty.class))
+ .thenReturn(selectedRuntimeProperty("poll_sequence"));
+ when(properties.singleValueParameter(POLLING_INTERVAL_SECONDS_KEY, Integer.class)).thenReturn(60);
+ when(properties.singleValueParameter(BATCH_SIZE_KEY, Integer.class)).thenReturn(10);
+ when(properties.singleValueParameter(MAX_ROWS_PER_POLL_KEY, Integer.class)).thenReturn(100);
+
+ AdapterDescription description = new AdapterDescription();
+ description.setElementId(adapterId);
+
+ IAdapterParameterExtractor extractor = mock(IAdapterParameterExtractor.class);
+ when(extractor.getStaticPropertyExtractor()).thenReturn(properties);
+ when(extractor.getAdapterDescription()).thenReturn(description);
+ return extractor;
+ }
+
+ private RuntimeResolvableOneOfStaticProperty selectedRuntimeProperty(String value) {
+ Option option = new Option(value);
+ option.setSelected(true);
+ RuntimeResolvableOneOfStaticProperty property = new RuntimeResolvableOneOfStaticProperty();
+ property.setOptions(new ArrayList<>(List.of(option)));
+ return property;
+ }
+
+ private IAdapterRuntimeContext runtimeContext() {
+ IAdapterRuntimeContext context = mock(IAdapterRuntimeContext.class);
+ when(context.getLogger()).thenReturn(mock(IExtensionsLogger.class));
+ return context;
+ }
+
+ private void createTable(String tableName) throws Exception {
+ try (Connection connection = connection(); Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE dbo." + tableName + " ("
+ + "poll_sequence BIGINT NOT NULL PRIMARY KEY, "
+ + "asset_reference BIGINT NOT NULL, "
+ + "measurement_amount DECIMAL(18, 3) NOT NULL, "
+ + "captured_at DATETIME2(3) NOT NULL, "
+ + "event_note NVARCHAR(100) NOT NULL)");
+ }
+ }
+
+ private void insertRow(String tableName,
+ long pollSequence,
+ long assetReference,
+ BigDecimal amount,
+ String eventNote) throws Exception {
+ String sql = "INSERT INTO dbo." + tableName
+ + " (poll_sequence, asset_reference, measurement_amount, captured_at, event_note) "
+ + "VALUES (?, ?, ?, ?, ?)";
+ try (Connection connection = connection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+ statement.setLong(1, pollSequence);
+ statement.setLong(2, assetReference);
+ statement.setBigDecimal(3, amount);
+ statement.setTimestamp(4, Timestamp.from(Instant.parse("2026-08-10T10:15:30Z").plusSeconds(pollSequence)));
+ statement.setString(5, eventNote);
+ statement.executeUpdate();
+ }
+ }
+
+ private void dropTable(String tableName) throws Exception {
+ try (Connection connection = connection(); Statement statement = connection.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS dbo." + tableName);
+ }
+ }
+
+ private Connection connection() throws Exception {
+ String url = "jdbc:sqlserver://" + System.getenv("SP_TEST_MSSQL_HOST") + ":" + PORT
+ + ";databaseName=" + DATABASE + ";encrypt=true;trustServerCertificate=true;loginTimeout=15";
+ return DriverManager.getConnection(url, USERNAME, PASSWORD);
+ }
+
+ private static String valueOrDefault(String name, String defaultValue) {
+ String value = System.getenv(name);
+ return value == null || value.isBlank() ? defaultValue : value;
+ }
+
+}
diff --git a/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.spec.ts b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.spec.ts
new file mode 100644
index 0000000000..efb6678cca
--- /dev/null
+++ b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-oneof-input/static-runtime-resolvable-oneof-input.component.spec.ts
@@ -0,0 +1,67 @@
+/*
+ * 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.
+ *
+ */
+
+import '@angular/compiler';
+import { Injector, runInInjectionContext } from '@angular/core';
+import { UntypedFormGroup } from '@angular/forms';
+import {
+ Option,
+ RuntimeResolvableOneOfStaticProperty,
+} from '@streampipes/platform-services';
+import { beforeEach, describe, expect, it } from 'vitest';
+import { StaticPropertyUtilService } from '../static-property-util.service';
+import { RuntimeResolvableService } from '../static-runtime-resolvable-input/runtime-resolvable.service';
+import { StaticRuntimeResolvableOneOfInputComponent } from './static-runtime-resolvable-oneof-input.component';
+
+describe('StaticRuntimeResolvableOneOfInputComponent', () => {
+ let component: StaticRuntimeResolvableOneOfInputComponent;
+
+ beforeEach(() => {
+ const injector = Injector.create({
+ providers: [
+ { provide: RuntimeResolvableService, useValue: {} },
+ { provide: StaticPropertyUtilService, useValue: {} },
+ ],
+ });
+ component = runInInjectionContext(
+ injector,
+ () => new StaticRuntimeResolvableOneOfInputComponent(),
+ );
+ });
+
+ it('marks a selected runtime option as configured', () => {
+ const option = new Option();
+ option.selected = true;
+ const staticProperty = new RuntimeResolvableOneOfStaticProperty();
+ staticProperty.internalName = 'table';
+ staticProperty.options = [option];
+ component.staticProperty = staticProperty;
+ component.parentForm = new UntypedFormGroup({});
+ const completedConfigurations = [];
+ component.completedConfigurationsEmitter.subscribe(configuration =>
+ completedConfigurations.push(configuration),
+ );
+
+ component.ngOnInit();
+
+ expect(completedConfigurations).toContainEqual({
+ staticPropertyInternalName: 'table',
+ configured: true,
+ });
+ });
+});
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 8f64958019..eec1ef0ec4 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
@@ -135,5 +135,6 @@ export class StaticRuntimeResolvableOneOfInputComponent
this.parentForm.controls[this.staticProperty.internalName].setErrors(
error,
);
+ this.applyCompletedConfiguration(error === undefined);
}
}