This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch
4367-warn-on-incompatible-dataset-schema-changes-affecting-measurements-in-pipelines
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4367-warn-on-incompatible-dataset-schema-changes-affecting-measurements-in-pipelines
by this push:
new 47e0ce2e3a Add unit tests
47e0ce2e3a is described below
commit 47e0ce2e3a07f043823faae3d90f030f57e4143a
Author: Sven Oehler <[email protected]>
AuthorDate: Wed May 6 17:39:00 2026 +0200
Add unit tests
---
.../DataExplorerSchemaManagementTest.java | 18 +-
.../v2/pipeline/MeasurementChangeDetectorTest.java | 203 +++++++++++++++++++++
.../MeasurementChangeValidationStepTest.java | 113 ++++++++++++
.../update/PipelineUpdateCoordinatorTest.java | 124 ++++++++++++-
4 files changed, 447 insertions(+), 11 deletions(-)
diff --git
a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
index 95ae273e62..0dbecba6f5 100644
---
a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
+++
b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
@@ -37,6 +37,7 @@ import java.util.List;
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;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -145,17 +146,16 @@ public class DataExplorerSchemaManagementTest {
var newMeasure =
getNewMeasure(DataLakeMeasureSchemaUpdateStrategy.EXTEND_EXISTING_SCHEMA);
- var resultMeasure =
schemaManagement.createOrUpdateMeasurement(newMeasure,null);
- assertEquals(newMeasure.getMeasureName(), resultMeasure.getMeasureName());
- verify(dataLakeStorageMock, Mockito.times(1)).updateElement(any());
+ var exception = assertThrows(
+ RuntimeException.class,
+ () -> schemaManagement.createOrUpdateMeasurement(newMeasure, null)
+ );
+
assertEquals(
- 2,
- resultMeasure.getEventSchema()
- .getEventProperties()
- .size()
+ "Can't save measurement with critical field changes: newProperty ("
+ + XSD.INTEGER + " -> " + XSD.STRING + ")",
+ exception.getMessage()
);
- assertTrue(containsPropertyWithName(resultMeasure, OLD_PROPERTY));
- assertTrue(containsPropertyWithName(resultMeasure, NEW_PROPERTY));
}
private EventProperty getEventProperty(
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetectorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetectorTest.java
new file mode 100644
index 0000000000..db952620c4
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetectorTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.List;
+
+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 MeasurementChangeDetectorTest {
+
+ private static final String STREAM_ID = "stream-1";
+ private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+
+ private final MeasurementChangeDetector detector = new
MeasurementChangeDetector();
+
+ @Test
+ void
hasCriticalMeasurementFieldChange_ShouldIgnoreChangesWithoutDatabaseSink() {
+ var pipeline =
makePipeline(makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)));
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.STRING));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertFalse(result);
+ }
+
+ @Test
+ void
hasCriticalMeasurementFieldChange_ShouldDetectCriticalChangeForDataLakeSink() {
+ var pipeline = makePipeline(
+ makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
+ makeDataLakeSink()
+ );
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.STRING));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertTrue(result);
+ }
+
+ @Test
+ void
hasCriticalMeasurementFieldChange_ShouldDetectCriticalChangeForDatabaseSinkCategory()
{
+ var pipeline = makePipeline(
+ makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
+ makeDatabaseSink()
+ );
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.BOOLEAN));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertTrue(result);
+ }
+
+ @Test
+ void findCriticalMeasurementFieldChanges_ShouldReturnChangedNamesAndTypes() {
+ var existingSchema = makeSchema(
+ makeMeasurementProperty("temperature", XSD.INTEGER),
+ makeMeasurementProperty("pressure", XSD.FLOAT)
+ );
+ var updatedSchema = makeSchema(
+ makeMeasurementProperty("temperature", XSD.STRING),
+ makeMeasurementProperty("pressure", XSD.BOOLEAN)
+ );
+
+ var result = detector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+
+ assertEquals(2, result.size());
+ assertEquals("temperature", result.get(0).runtimeName());
+ assertEquals(XSD.INTEGER.toString(), result.get(0).existingType());
+ assertEquals(XSD.STRING.toString(), result.get(0).updatedType());
+ assertEquals("pressure", result.get(1).runtimeName());
+ assertEquals(XSD.FLOAT.toString(), result.get(1).existingType());
+ assertEquals(XSD.BOOLEAN.toString(), result.get(1).updatedType());
+ }
+
+ @Test
+ void hasCriticalMeasurementFieldChange_ShouldIgnoreIntegerToLongChange() {
+ var pipeline = makePipeline(
+ makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
+ makeDatabaseSink()
+ );
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.LONG));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertFalse(result);
+ }
+
+ @Test
+ void hasCriticalMeasurementFieldChange_ShouldIgnoreFloatToDoubleChange() {
+ var pipeline = makePipeline(
+ makeSchema(makeMeasurementProperty("temperature", XSD.FLOAT)),
+ makeDatabaseSink()
+ );
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.DOUBLE));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertFalse(result);
+ }
+
+ @Test
+ void hasCriticalMeasurementFieldChange_ShouldIgnoreAddedAndRemovedFields() {
+ var pipeline = makePipeline(
+ makeSchema(
+ makeMeasurementProperty("temperature", XSD.INTEGER),
+ makeMeasurementProperty("pressure", XSD.DOUBLE)
+ ),
+ makeDatabaseSink()
+ );
+ var updatedSchema = makeSchema(
+ makeMeasurementProperty("temperature", XSD.INTEGER),
+ makeMeasurementProperty("humidity", XSD.DOUBLE)
+ );
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertFalse(result);
+ }
+
+ @Test
+ void hasCriticalMeasurementFieldChange_ShouldIgnoreDimensionFieldChanges() {
+ var pipeline = makePipeline(
+ makeSchema(makeDimensionProperty("machineId", XSD.INTEGER)),
+ makeDatabaseSink()
+ );
+ var updatedSchema = makeSchema(makeDimensionProperty("machineId",
XSD.STRING));
+
+ var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+
+ assertFalse(result);
+ }
+
+ private Pipeline makePipeline(EventSchema eventSchema,
+ DataSinkInvocation... actions) {
+ var dataStream = new SpDataStream();
+ dataStream.setElementId(STREAM_ID);
+ dataStream.setEventSchema(eventSchema);
+
+ var pipeline = new Pipeline();
+ pipeline.setStreams(List.of(dataStream));
+ pipeline.setActions(List.of(actions));
+ return pipeline;
+ }
+
+ private DataSinkInvocation makeDataLakeSink() {
+ var sink = new DataSinkInvocation();
+ sink.setAppId(DATA_LAKE_SINK_APP_ID);
+ return sink;
+ }
+
+ private DataSinkInvocation makeDatabaseSink() {
+ var sink = new DataSinkInvocation();
+ sink.setCategory(List.of(DataSinkType.DATABASE.name()));
+ return sink;
+ }
+
+ private EventSchema makeSchema(EventPropertyPrimitive... properties) {
+ return new EventSchema(List.of(properties));
+ }
+
+ private EventPropertyPrimitive makeMeasurementProperty(String runtimeName,
+ URI runtimeType) {
+ var property = new EventPropertyPrimitive(runtimeType.toString(),
runtimeName, "", "");
+ property.setPropertyScope(PropertyScope.MEASUREMENT_PROPERTY.name());
+ return property;
+ }
+
+ private EventPropertyPrimitive makeDimensionProperty(String runtimeName,
+ URI runtimeType) {
+ var property = new EventPropertyPrimitive(runtimeType.toString(),
runtimeName, "", "");
+ property.setPropertyScope(PropertyScope.DIMENSION_PROPERTY.name());
+ return property;
+ }
+}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStepTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStepTest.java
new file mode 100644
index 0000000000..8668bec49e
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStepTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class MeasurementChangeValidationStepTest {
+
+ private final MeasurementChangeValidationStep step = new
MeasurementChangeValidationStep();
+
+ @Test
+ void apply_ShouldAddValidationInfoForCriticalDatabaseMeasurementChange() {
+ var source = makeStream(makeSchema(makeMeasurementProperty("temperature",
XSD.STRING)));
+ var target =
makeDatabaseSink(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ var validationInfos = new ArrayList<PipelineElementValidationInfo>();
+
+ step.apply(source, target, Set.of(target), validationInfos);
+
+ assertEquals(1, validationInfos.size());
+ assertEquals(
+ MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED
+ + ": temperature (" + XSD.INTEGER + " -> " + XSD.STRING + ")",
+ validationInfos.get(0).getMessage()
+ );
+ }
+
+ @Test
+ void apply_ShouldIgnoreNonDatabaseSinks() {
+ var source = makeStream(makeSchema(makeMeasurementProperty("temperature",
XSD.STRING)));
+ var target = makeSink(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ var validationInfos = new ArrayList<PipelineElementValidationInfo>();
+
+ step.apply(source, target, Set.of(target), validationInfos);
+
+ assertTrue(validationInfos.isEmpty());
+ }
+
+ @Test
+ void apply_ShouldIgnoreNonCriticalStorageTypeChanges() {
+ var source = makeStream(makeSchema(makeMeasurementProperty("temperature",
XSD.LONG)));
+ var target =
makeDatabaseSink(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ var validationInfos = new ArrayList<PipelineElementValidationInfo>();
+
+ step.apply(source, target, Set.of(target), validationInfos);
+
+ assertTrue(validationInfos.isEmpty());
+ }
+
+ private SpDataStream makeStream(EventSchema eventSchema) {
+ var stream = new SpDataStream();
+ stream.setEventSchema(eventSchema);
+ return stream;
+ }
+
+ private DataSinkInvocation makeDatabaseSink(EventSchema inputSchema) {
+ var sink = makeSink(inputSchema);
+ sink.setCategory(List.of(DataSinkType.DATABASE.name()));
+ return sink;
+ }
+
+ private DataSinkInvocation makeSink(EventSchema inputSchema) {
+ var inputStream = new SpDataStream();
+ inputStream.setEventSchema(inputSchema);
+
+ var sink = new DataSinkInvocation();
+ sink.setInputStreams(List.of(inputStream));
+ return sink;
+ }
+
+ private EventSchema makeSchema(EventPropertyPrimitive... properties) {
+ return new EventSchema(List.of(properties));
+ }
+
+ private EventPropertyPrimitive makeMeasurementProperty(String runtimeName,
+ URI runtimeType) {
+ var property = new EventPropertyPrimitive(runtimeType.toString(),
runtimeName, "", "");
+ property.setPropertyScope(PropertyScope.MEASUREMENT_PROPERTY.name());
+ return property;
+ }
+}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
index 638a64fc24..e15511347e 100644
---
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
@@ -21,27 +21,33 @@ package org.apache.streampipes.manager.pipeline.update;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.execution.PipelineExecutor;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
+import
org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeValidationStep;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.model.message.PipelineModificationMessage;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
import org.apache.streampipes.model.pipeline.PipelineModification;
import org.apache.streampipes.model.pipeline.PipelineModificationResult;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.storage.api.core.INoSqlStorage;
import org.apache.streampipes.storage.api.pipeline.IPipelineStorage;
import org.apache.streampipes.storage.couchdb.CouchDbStorageManager;
+import org.apache.streampipes.vocabulary.XSD;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
+import java.net.URI;
import java.util.ArrayList;
import java.util.List;
@@ -57,6 +63,8 @@ import static org.mockito.Mockito.when;
class PipelineUpdateCoordinatorTest {
+ private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+
@Test
void updatePipelines_ShouldRestartRunningPipelinesForDataStreamUpdates() {
var requestManager = mock(ExtensionServiceRequestManager.class);
@@ -77,7 +85,8 @@ class PipelineUpdateCoordinatorTest {
mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) -> {
verifiedPipelines.add((Pipeline) context.arguments().get(0));
when(mock.verifyPipeline()).thenReturn(modificationMessage);
- when(mock.makeModifiedPipeline()).thenReturn(new
PipelineModificationResult(modifiedPipeline, List.of()));
+ when(mock.makeModifiedPipeline(modificationMessage))
+ .thenReturn(new
PipelineModificationResult(modifiedPipeline, List.of()));
});
MockedConstruction<CouchDbStorageManager> storageManagerConstruction =
mockConstruction(CouchDbStorageManager.class, (mock, context) ->
@@ -126,7 +135,8 @@ class PipelineUpdateCoordinatorTest {
mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) -> {
verifiedPipelines.add((Pipeline) context.arguments().get(0));
when(mock.verifyPipeline()).thenReturn(modificationMessage);
- when(mock.makeModifiedPipeline()).thenReturn(new
PipelineModificationResult(modifiedPipeline, List.of()));
+ when(mock.makeModifiedPipeline(modificationMessage))
+ .thenReturn(new
PipelineModificationResult(modifiedPipeline, List.of()));
});
MockedConstruction<CouchDbStorageManager> storageManagerConstruction =
mockConstruction(CouchDbStorageManager.class, (mock, context) ->
@@ -157,6 +167,56 @@ class PipelineUpdateCoordinatorTest {
}
}
+ @Test
+ void
updatePipelines_ShouldMarkPipelineRequiringAttentionForCriticalMeasurementFieldChange()
{
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ var coordinator = new PipelineUpdateCoordinator(requestManager);
+ var adapterDescription = makeAdapter("stream-1", "Updated adapter");
+
adapterDescription.getDataStream().setEventSchema(makeSchema(makeMeasurementProperty("temperature",
XSD.STRING)));
+
+ var storedPipeline = makePipeline("pipeline-1", "Pipeline", true,
"stream-1", "Old stream");
+
storedPipeline.getStreams().get(0).setEventSchema(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ storedPipeline.setActions(List.of(makeDataLakeSink()));
+
+ var modifiedPipeline = makePipeline("pipeline-1", "Pipeline", true,
"stream-1", "Updated adapter");
+ var measurementUpdateInfo = PipelineElementValidationInfo.info(
+ measurementUpdateRequiredMessage());
+ var modificationMessage = new
PipelineModificationMessage(List.of(validModification("sepa-1",
measurementUpdateInfo)));
+ var pipelineStorage = mock(IPipelineStorage.class);
+
+ try (MockedStatic<PipelineManager> pipelineManager =
mockStatic(PipelineManager.class);
+ MockedConstruction<PipelineVerificationHandlerV2>
verificationHandlerConstruction =
+ mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) -> {
+ when(mock.verifyPipeline()).thenReturn(modificationMessage);
+ when(mock.makeModifiedPipeline(modificationMessage))
+ .thenReturn(new
PipelineModificationResult(modifiedPipeline, List.of()));
+ });
+ MockedConstruction<CouchDbStorageManager> storageManagerConstruction =
+ mockConstruction(CouchDbStorageManager.class, (mock, context) ->
+
when(mock.getPipelineStorageAPI()).thenReturn(pipelineStorage));
+ MockedConstruction<PipelineExecutor> executorConstruction =
+ mockConstruction(PipelineExecutor.class)) {
+
+ pipelineManager.when(() ->
PipelineManager.getPipelinesContainingElements("stream-1"))
+ .thenReturn(List.of(storedPipeline));
+ pipelineManager.when(() -> PipelineManager.getPipeline("pipeline-1"))
+ .thenReturn(storedPipeline);
+
+ coordinator.updatePipelines(adapterDescription);
+
+ assertEquals(1, verificationHandlerConstruction.constructed().size());
+ assertEquals(1, storageManagerConstruction.constructed().size());
+
+ var pipelineCaptor = ArgumentCaptor.forClass(Pipeline.class);
+ verify(pipelineStorage).updateElement(pipelineCaptor.capture());
+ assertEquals(PipelineHealthStatus.HANDLE_MEASUREMENT_UPDATE,
pipelineCaptor.getValue().getHealthStatus());
+ assertFalse(pipelineCaptor.getValue().isValid());
+
+ assertEquals(1, executorConstruction.constructed().size());
+ verify(executorConstruction.constructed().get(0)).stopPipeline(true);
+ }
+ }
+
@Test
void checkPipelineMigrations_ShouldUseUpdatedDataStreamValues() {
var requestManager = mock(ExtensionServiceRequestManager.class);
@@ -226,6 +286,37 @@ class PipelineUpdateCoordinatorTest {
}
}
+ @Test
+ void
checkPipelineMigrations_ShouldDisableAutoMigrationForCriticalMeasurementFieldChange()
{
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ var coordinator = new PipelineUpdateCoordinator(requestManager);
+ var adapterDescription = makeAdapter("stream-1", "Updated adapter");
+
adapterDescription.getDataStream().setEventSchema(makeSchema(makeMeasurementProperty("temperature",
XSD.STRING)));
+
+ var pipeline = makePipeline("pipeline-1", "Pipeline", false, "stream-1",
"Old stream");
+
pipeline.getStreams().get(0).setEventSchema(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ pipeline.setActions(List.of(makeDataLakeSink()));
+
+ var measurementUpdateInfo = PipelineElementValidationInfo.info(
+ measurementUpdateRequiredMessage());
+ var modificationMessage = new
PipelineModificationMessage(List.of(validModification("sepa-1",
measurementUpdateInfo)));
+
+ try (MockedStatic<PipelineManager> pipelineManager =
mockStatic(PipelineManager.class);
+ MockedConstruction<PipelineVerificationHandlerV2>
verificationHandlerConstruction =
+ mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) ->
+ when(mock.verifyPipeline()).thenReturn(modificationMessage)))
{
+
+ pipelineManager.when(() ->
PipelineManager.getPipelinesContainingElements("stream-1"))
+ .thenReturn(List.of(pipeline));
+
+ var result = coordinator.checkPipelineMigrations(adapterDescription);
+
+ assertEquals(1, verificationHandlerConstruction.constructed().size());
+ assertEquals(1, result.size());
+ assertFalse(result.get(0).isCanAutoMigrate());
+ }
+ }
+
private SpDataStream makeDataStream(String elementId, String name) {
var dataStream = new SpDataStream();
dataStream.setElementId(elementId);
@@ -256,6 +347,28 @@ class PipelineUpdateCoordinatorTest {
return pipeline;
}
+ private DataSinkInvocation makeDataLakeSink() {
+ var sink = new DataSinkInvocation();
+ sink.setAppId(DATA_LAKE_SINK_APP_ID);
+ return sink;
+ }
+
+ private EventSchema makeSchema(EventPropertyPrimitive... eventProperties) {
+ return new EventSchema(List.of(eventProperties));
+ }
+
+ private EventPropertyPrimitive makeMeasurementProperty(String runtimeName,
+ URI runtimeType) {
+ var property = new EventPropertyPrimitive(runtimeType.toString(),
runtimeName, "", "");
+ property.setPropertyScope(PropertyScope.MEASUREMENT_PROPERTY.name());
+ return property;
+ }
+
+ private String measurementUpdateRequiredMessage() {
+ return MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED
+ + ": temperature (" + XSD.INTEGER + " -> " + XSD.STRING + ")";
+ }
+
private DataProcessorInvocation makeSepa(String elementId, String name) {
var sepa = new DataProcessorInvocation();
sepa.setElementId(elementId);
@@ -270,6 +383,13 @@ class PipelineUpdateCoordinatorTest {
return modification;
}
+ private PipelineModification validModification(String elementId,
+ PipelineElementValidationInfo
validationInfo) {
+ var modification = validModification(elementId);
+ modification.setValidationInfos(List.of(validationInfo));
+ return modification;
+ }
+
private PipelineModification invalidModification(String elementId,
PipelineElementValidationInfo validationInfo) {
var modification = new PipelineModification();
