This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 49bc094e97 feat(#4367): Warn on incompatible dataset schema changes
affecting measurements in pipelines (#4449)
49bc094e97 is described below
commit 49bc094e9747582a6b8fe754d2c4193539a09ed6
Author: Sven Oehler <[email protected]>
AuthorDate: Fri May 8 09:23:13 2026 +0200
feat(#4367): Warn on incompatible dataset schema changes affecting
measurements in pipelines (#4449)
---
.../streampipes/client/http/HttpRequest.java | 37 ++++-
.../dataexplorer/DataExplorerSchemaManagement.java | 22 +++
.../DataExplorerSchemaManagementTest.java | 18 +--
.../model/pipeline/PipelineHealthStatus.java | 1 +
.../matching/PipelineVerificationHandlerV2.java | 13 +-
.../pipeline/CriticalMeasurementFieldChange.java | 10 +-
.../v2/pipeline/MeasurementChangeDetector.java | 110 +++++++++++++++
.../pipeline/MeasurementChangeValidationStep.java | 119 ++++++++++++++++
.../v2/pipeline/PipelineValidationSteps.java | 1 +
.../manager/matching/v2/pipeline/StorageType.java | 12 +-
.../pipeline/update/PipelineUpdateCoordinator.java | 21 ++-
.../v2/pipeline/MeasurementChangeDetectorTest.java | 121 +++++++++++++++++
.../MeasurementChangeValidationStepTest.java | 132 ++++++++++++++++++
.../update/PipelineUpdateCoordinatorTest.java | 124 ++++++++++++++++-
.../impl/datalake/DataLakeMeasureResource.java | 17 ++-
.../src/lib/model/gen/streampipes-model.ts | 8 +-
.../pipeline-details/pipeline-details.component.ts | 40 ++++++
.../pipeline-preview-meta.component.ts | 5 +-
.../pipeline-overview.component.html | 4 +-
.../measurement-update-dialog.component.html | 150 +++++++++++++++++++++
.../measurement-update-dialog.component.scss | 20 ++-
.../measurement-update-dialog.component.ts | 60 +++++++++
22 files changed, 1004 insertions(+), 41 deletions(-)
diff --git
a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
index cbcc8550e5..69e324c5b6 100644
---
a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
+++
b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
@@ -24,7 +24,10 @@ import org.apache.streampipes.client.serializer.Serializer;
import org.apache.streampipes.client.util.StreamPipesApiPath;
import org.apache.streampipes.commons.exceptions.SpHttpErrorStatusCode;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
@@ -109,7 +112,7 @@ public abstract class HttpRequest<K, V, T> {
case HttpStatus.SC_NOT_FOUND ->
throw new SpHttpErrorStatusCode(" 404 - The requested resource
could not be found.",
HttpStatus.SC_NOT_FOUND);
- default -> throw new SpHttpErrorStatusCode(status.getStatusCode() +
" - " + status.getReasonPhrase(),
+ default -> throw new SpHttpErrorStatusCode(makeErrorMessage(status,
response.getEntity()),
status.getStatusCode());
}
}
@@ -119,6 +122,38 @@ public abstract class HttpRequest<K, V, T> {
}
}
+ private String makeErrorMessage(StatusLine status,
+ HttpEntity entity) throws IOException {
+ String message;
+ if (!status.getReasonPhrase().isBlank()) {
+ message = status.getReasonPhrase();
+ } else {
+ String body = entity != null ? entityAsString(entity) : "";
+ message = extractErrorMessage(body);
+ }
+ return status.getStatusCode() + " - " + message;
+ }
+
+ private String extractErrorMessage(String responseBody) {
+ try {
+ JsonNode jsonNode =
JacksonSerializer.getObjectMapper().readTree(responseBody);
+
+ if (jsonNode.isTextual()) {
+ return jsonNode.asText();
+ } else if (jsonNode.hasNonNull("cause")) {
+ return jsonNode.get("cause").asText();
+ } else if (jsonNode.hasNonNull("title")) {
+ return jsonNode.get("title").asText();
+ } else if (jsonNode.hasNonNull("detail")) {
+ return jsonNode.get("detail").asText();
+ }
+ } catch (JsonProcessingException e) {
+ return responseBody;
+ }
+
+ return responseBody;
+ }
+
public void writeToFile(String fileLocation) throws SpRuntimeException {
String urlString = makeUrl();
diff --git
a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java
index cb3ad997b9..7a2076a879 100644
---
a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java
+++
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java
@@ -19,10 +19,12 @@
package org.apache.streampipes.dataexplorer;
import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement;
+import
org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeDetector;
import org.apache.streampipes.manager.permission.DataLakePermissionManager;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import
org.apache.streampipes.model.datalake.DataLakeMeasureSchemaUpdateStrategy;
import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.storage.api.core.CRUDStorage;
import java.util.ArrayList;
@@ -86,6 +88,7 @@ public class DataExplorerSchemaManagement implements
IDataExplorerSchemaManageme
DataLakeMeasure measure,
DataLakeMeasure existingMeasure) {
measure.setElementId(existingMeasure.getElementId());
+ checkFieldChanges(existingMeasure.getEventSchema(),
measure.getEventSchema());
if
(DataLakeMeasureSchemaUpdateStrategy.UPDATE_SCHEMA.equals(measure.getSchemaUpdateStrategy()))
{
// For the update schema strategy the old schema is overwritten with the
new one
updateMeasurement(measure);
@@ -197,4 +200,23 @@ public class DataExplorerSchemaManagement implements
IDataExplorerSchemaManageme
.values();
return new ArrayList<>(unifiedEventProperties);
}
+
+ private void checkFieldChanges(EventSchema existingSchema, EventSchema
schema) {
+ var criticalFieldChanges =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(
+ existingSchema,
+ schema
+ );
+ if (!criticalFieldChanges.isEmpty()) {
+ throw new RuntimeException(
+ "Can't save measurement with critical field changes: " +
criticalFieldChanges
+ .stream()
+ .map(change -> "%s (%s -> %s)".formatted(
+ change.runtimeName(),
+ change.existingType(),
+ change.updatedType()
+ ))
+ .collect(Collectors.joining(", "))
+ );
+ }
+ }
}
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-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
index 6fac07a465..b863e47ad7 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
@@ -20,5 +20,6 @@ package org.apache.streampipes.model.pipeline;
public enum PipelineHealthStatus {
OK,
REQUIRES_ATTENTION,
+ HANDLE_MEASUREMENT_UPDATE,
FAILURE
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
index d87f7a8139..d57c4af51f 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
@@ -57,7 +57,11 @@ public class PipelineVerificationHandlerV2 {
}
public PipelineModificationResult makeModifiedPipeline() {
- var result = verifyAndBuildGraphs(false);
+ return makeModifiedPipeline(verifyPipeline());
+ }
+
+ public PipelineModificationResult
makeModifiedPipeline(PipelineModificationMessage modificationMessage) {
+ var result = verifyAndBuildGraphs(modificationMessage, false);
var allElements = result.modifiedPipelineElements();
pipeline.setSepas(filterAndConvert(allElements,
DataProcessorInvocation.class));
pipeline.setActions(filterAndConvert(allElements,
DataSinkInvocation.class));
@@ -74,7 +78,12 @@ public class PipelineVerificationHandlerV2 {
}
public PipelineVerificationResult verifyAndBuildGraphs(boolean
ignoreUnconfigured) {
- var pipelineModifications = verifyPipeline().getPipelineModifications();
+ return verifyAndBuildGraphs(verifyPipeline(), ignoreUnconfigured);
+ }
+
+ public PipelineVerificationResult
verifyAndBuildGraphs(PipelineModificationMessage modificationMessage,
+ boolean
ignoreUnconfigured) {
+ var pipelineModifications = modificationMessage.getPipelineModifications();
var allElements = new AllElementsProvider(pipeline).getAllElements();
var validationInfos = new
ArrayList<ExtendedPipelineElementValidationInfo>();
var modifiedPipelineElements = new ArrayList<NamedStreamPipesEntity>();
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
similarity index 75%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
index 6fac07a465..250b55ab52 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
@@ -15,10 +15,10 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.model.pipeline;
-public enum PipelineHealthStatus {
- OK,
- REQUIRES_ATTENTION,
- FAILURE
+package org.apache.streampipes.manager.matching.v2.pipeline;
+
+public record CriticalMeasurementFieldChange(String runtimeName,
+ String existingType,
+ String updatedType) {
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
new file mode 100644
index 0000000000..a7026fb7a6
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
@@ -0,0 +1,110 @@
+/*
+ * 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.schema.EventProperty;
+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.SO;
+import org.apache.streampipes.vocabulary.XSD;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public final class MeasurementChangeDetector {
+
+ private MeasurementChangeDetector() {
+ }
+
+ public static List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(
+ EventSchema existingEventSchema,
+ EventSchema updatedEventSchema) {
+ var existingMeasurementFields = existingEventSchema
+ .getEventProperties()
+ .stream()
+ .filter(MeasurementChangeDetector::isMeasurementField)
+ .collect(Collectors.toMap(
+ EventProperty::getRuntimeName,
+ Function.identity(),
+ (existingProperty, duplicateProperty) -> existingProperty
+ ));
+
+ return updatedEventSchema
+ .getEventProperties()
+ .stream()
+ .filter(MeasurementChangeDetector::isMeasurementField)
+ .map(updatedProperty -> {
+ var existingProperty =
existingMeasurementFields.get(updatedProperty.getRuntimeName());
+ if (existingProperty != null &&
hasCriticalFieldTypeChange(existingProperty, updatedProperty)) {
+ return Optional.of(new CriticalMeasurementFieldChange(
+ updatedProperty.getRuntimeName(),
+ toDisplayType(existingProperty),
+ toDisplayType(updatedProperty)
+ ));
+ } else {
+ return Optional.<CriticalMeasurementFieldChange>empty();
+ }
+ })
+ .flatMap(Optional::stream)
+ .toList();
+ }
+
+ private static boolean isMeasurementField(EventProperty eventProperty) {
+ return
!PropertyScope.DIMENSION_PROPERTY.name().equals(eventProperty.getPropertyScope());
+ }
+
+ private static boolean hasCriticalFieldTypeChange(EventProperty
existingProperty,
+ EventProperty
updatedProperty) {
+ return toStorageType(existingProperty) != toStorageType(updatedProperty);
+ }
+
+ private static StorageType toStorageType(EventProperty eventProperty) {
+ if (eventProperty instanceof EventPropertyPrimitive primitiveProperty) {
+ return toPrimitiveStorageType(primitiveProperty.getRuntimeType());
+ } else {
+ return StorageType.STRING;
+ }
+ }
+
+ private static StorageType toPrimitiveStorageType(String runtimeType) {
+ if (XSD.INTEGER.toString().equals(runtimeType)
+ || XSD.LONG.toString().equals(runtimeType)) {
+ return StorageType.INTEGER;
+ } else if (XSD.FLOAT.toString().equals(runtimeType)
+ || XSD.DOUBLE.toString().equals(runtimeType)
+ || SO.NUMBER.equals(runtimeType)) {
+ return StorageType.FLOAT;
+ } else if (XSD.BOOLEAN.toString().equals(runtimeType)) {
+ return StorageType.BOOLEAN;
+ } else {
+ return StorageType.STRING;
+ }
+ }
+
+ private static String toDisplayType(EventProperty eventProperty) {
+ if (eventProperty instanceof EventPropertyPrimitive primitiveProperty) {
+ return primitiveProperty.getRuntimeType();
+ } else {
+ return eventProperty.getClass().getSimpleName();
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStep.java
new file mode 100644
index 0000000000..4d53c0d49c
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStep.java
@@ -0,0 +1,119 @@
+/*
+ * 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.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.schema.EventSchema;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+public class MeasurementChangeValidationStep extends
AbstractPipelineValidationStep {
+
+ public static final String MEASUREMENT_UPDATE_REQUIRED =
+ "Measurement field storage type changed. Manual measurement update
handling required";
+ private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos) {
+ if (target instanceof DataSinkInvocation dataSink &&
isDatabaseSink(dataSink)) {
+ var criticalFieldChanges = findCriticalMeasurementFieldChanges(source,
dataSink);
+ if (!criticalFieldChanges.isEmpty()) {
+
validationInfos.add(PipelineElementValidationInfo.error(makeValidationMessage(criticalFieldChanges)));
+ }
+ }
+
+ if (target.getInputStreams() != null && target.getInputStreams().size() >
1) {
+ this.visitorHistory.put(target.getDom(), 1);
+ }
+ }
+
+ private boolean isDatabaseSink(DataSinkInvocation dataSink) {
+ return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId())
+ ||
streamOf(dataSink.getCategory()).anyMatch(DataSinkType.DATABASE.name()::equals);
+ }
+
+ private List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(
+ NamedStreamPipesEntity source,
+ DataSinkInvocation dataSink) {
+ var existingEventSchema = getExistingDataSinkSchema(dataSink);
+ var updatedEventSchema = getUpdatedSourceSchema(source);
+
+ if (existingEventSchema.isPresent() && updatedEventSchema.isPresent()) {
+ return MeasurementChangeDetector.findCriticalMeasurementFieldChanges(
+ existingEventSchema.get(),
+ updatedEventSchema.get()
+ );
+ } else {
+ return List.of();
+ }
+ }
+
+ private String makeValidationMessage(
+ List<CriticalMeasurementFieldChange> criticalFieldChanges) {
+ return MEASUREMENT_UPDATE_REQUIRED + ": " + criticalFieldChanges
+ .stream()
+ .map(change -> "%s (%s -> %s)".formatted(
+ change.runtimeName(),
+ change.existingType(),
+ change.updatedType()
+ ))
+ .collect(Collectors.joining(", "));
+ }
+
+ private Optional<EventSchema> getExistingDataSinkSchema(DataSinkInvocation
dataSink) {
+ if (dataSink.getInputStreams() == null ||
dataSink.getInputStreams().size() <= getIndex(dataSink)) {
+ return Optional.empty();
+ }
+
+ return
Optional.ofNullable(dataSink.getInputStreams().get(getIndex(dataSink)).getEventSchema());
+ }
+
+ private Optional<EventSchema> getUpdatedSourceSchema(NamedStreamPipesEntity
source) {
+ if (source instanceof SpDataStream dataStream) {
+ return Optional.ofNullable(dataStream.getEventSchema());
+ } else if (source instanceof DataProcessorInvocation dataProcessor
+ && dataProcessor.getOutputStream() != null) {
+ return
Optional.ofNullable(dataProcessor.getOutputStream().getEventSchema());
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ private <T> Stream<T> streamOf(Iterable<T> iterable) {
+ if (iterable == null) {
+ return Stream.empty();
+ }
+
+ return StreamSupport.stream(iterable.spliterator(), false);
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
index 495a79c418..fb8e1c1681 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
@@ -27,6 +27,7 @@ public class PipelineValidationSteps {
public List<IPipelineValidationStep> collect(ExtensionServiceRequestManager
requestManager) {
return Arrays.asList(
+ new MeasurementChangeValidationStep(),
new PrepareStep(),
new ApplyGroundingStep(),
new SchemaValidationStep(),
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/StorageType.java
similarity index 86%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/StorageType.java
index 6fac07a465..07f10abcbf 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/StorageType.java
@@ -15,10 +15,12 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.model.pipeline;
-public enum PipelineHealthStatus {
- OK,
- REQUIRES_ATTENTION,
- FAILURE
+package org.apache.streampipes.manager.matching.v2.pipeline;
+
+enum StorageType {
+ INTEGER,
+ FLOAT,
+ BOOLEAN,
+ STRING
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
index 6de0a17945..228477dd2d 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
@@ -22,6 +22,7 @@ import
org.apache.streampipes.commons.prometheus.pipelines.PipelinesStats;
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.base.NamedStreamPipesEntity;
@@ -107,11 +108,15 @@ public class PipelineUpdateCoordinator {
var verificationHandler = new
PipelineVerificationHandlerV2(updatedPipeline, requestManager);
var modificationMessage = verificationHandler.verifyPipeline();
var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
- var modifiedPipeline =
verificationHandler.makeModifiedPipeline().pipeline();
+ var modifiedPipeline =
verificationHandler.makeModifiedPipeline(modificationMessage).pipeline();
var canAutoMigrate = canAutoMigrate(modificationMessage);
if (!canAutoMigrate) {
-
modifiedPipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
+ modifiedPipeline.setHealthStatus(
+ requiresMeasurementUpdate(modificationMessage)
+ ? PipelineHealthStatus.HANDLE_MEASUREMENT_UPDATE
+ : PipelineHealthStatus.REQUIRES_ATTENTION
+ );
PIPELINES_STATS.updatePipelineHealthState(
modifiedPipeline.getElementId(),
modifiedPipeline.getName(),
@@ -142,7 +147,8 @@ public class PipelineUpdateCoordinator {
var updatedPipeline = updatePipeline(pipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
try {
var modificationMessage = new
PipelineVerificationHandlerV2(updatedPipeline, requestManager).verifyPipeline();
- updateInfos.add(makeUpdateInfo(modificationMessage, updatedPipeline));
+ var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
+ updateInfos.add(updateInfo);
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -188,6 +194,15 @@ public class PipelineUpdateCoordinator {
&& modification.getValidationInfos().isEmpty());
}
+ private boolean requiresMeasurementUpdate(PipelineModificationMessage
modificationMessage) {
+ return modificationMessage
+ .getPipelineModifications()
+ .stream()
+ .flatMap(modification -> modification.getValidationInfos().stream())
+ .anyMatch(validationInfo -> validationInfo.getMessage() != null
+ &&
validationInfo.getMessage().startsWith(MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED));
+ }
+
private List<String> toNotification(PipelineUpdateInfo updateInfo,
String updateType) {
var notifications = new ArrayList<String>();
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..d5b99a2956
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetectorTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.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.assertTrue;
+
+class MeasurementChangeDetectorTest {
+
+ @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 =
MeasurementChangeDetector.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 findCriticalMeasurementFieldChanges_ShouldIgnoreIntegerToLongChange() {
+ var existingSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER));
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.LONG));
+
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreFloatToDoubleChange() {
+ var existingSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.FLOAT));
+ var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.DOUBLE));
+
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreAddedAndRemovedFields()
{
+ var existingSchema = makeSchema(
+ makeMeasurementProperty("temperature", XSD.INTEGER),
+ makeMeasurementProperty("pressure", XSD.DOUBLE)
+ );
+ var updatedSchema = makeSchema(
+ makeMeasurementProperty("temperature", XSD.INTEGER),
+ makeMeasurementProperty("humidity", XSD.DOUBLE)
+ );
+
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreDimensionFieldChanges()
{
+ var existingSchema = makeSchema(makeDimensionProperty("machineId",
XSD.INTEGER));
+ var updatedSchema = makeSchema(makeDimensionProperty("machineId",
XSD.STRING));
+
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+
+ assertTrue(result.isEmpty());
+ }
+
+ 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..559e209fdc
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStepTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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 static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+
+ 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_ShouldAddValidationInfoForDataLakeSinkMeasurementChange() {
+ var source = makeStream(makeSchema(makeMeasurementProperty("temperature",
XSD.STRING)));
+ var target =
makeDataLakeSink(makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER)));
+ var validationInfos = new ArrayList<PipelineElementValidationInfo>();
+
+ step.apply(source, target, Set.of(target), validationInfos);
+
+ assertEquals(1, validationInfos.size());
+ }
+
+ @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 DataSinkInvocation makeDataLakeSink(EventSchema inputSchema) {
+ var sink = makeSink(inputSchema);
+ sink.setAppId(DATA_LAKE_SINK_APP_ID);
+ 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();
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeMeasureResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeMeasureResource.java
index 5637d37692..28e12fc92e 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeMeasureResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeMeasureResource.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.rest.impl.datalake;
import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.monitoring.SpLogMessage;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@@ -51,12 +52,16 @@ public class DataLakeMeasureResource extends
AbstractDataLakeResource {
@PostMapping(produces = MediaType.APPLICATION_JSON_VALUE, consumes =
MediaType.APPLICATION_JSON_VALUE)
@PreAuthorize("this.hasWriteAuthority()")
- public ResponseEntity<DataLakeMeasure> addDataLake(@RequestBody
DataLakeMeasure dataLakeMeasure) {
- DataLakeMeasure result =
this.dataLakeMeasureManagement.createOrUpdateMeasurement(
- dataLakeMeasure,
- getAuthenticatedUserSid()
- );
- return ok(result);
+ public ResponseEntity<?> addDataLake(@RequestBody DataLakeMeasure
dataLakeMeasure) {
+ try {
+ DataLakeMeasure result =
this.dataLakeMeasureManagement.createOrUpdateMeasurement(
+ dataLakeMeasure,
+ getAuthenticatedUserSid()
+ );
+ return ok(result);
+ } catch (RuntimeException e) {
+ return badRequest(SpLogMessage.from(e));
+ }
}
/**
diff --git
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index 397faf0f6b..1c09e3e189 100644
---
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -4632,7 +4632,13 @@ export type OutputStrategyUnion =
| TransformOutputStrategy
| UserDefinedOutputStrategy;
-export type PipelineHealthStatus = 'OK' | 'REQUIRES_ATTENTION' | 'FAILURE';
+export type PipelineHealthStatus =
+ | 'OK'
+ | 'REQUIRES_ATTENTION'
+ | 'HANDLE_MEASUREMENT_UPDATE'
+ | 'FAILURE';
+
+export type MeasurementUpdateAction = 'edit-pipeline' | 'manage-datasets';
export type PropertyScope =
| 'HEADER_PROPERTY'
diff --git a/ui/src/app/pipeline-details/pipeline-details.component.ts
b/ui/src/app/pipeline-details/pipeline-details.component.ts
index 7a2448ede8..eb63b3e913 100644
--- a/ui/src/app/pipeline-details/pipeline-details.component.ts
+++ b/ui/src/app/pipeline-details/pipeline-details.component.ts
@@ -20,6 +20,7 @@ import { Component, OnDestroy, OnInit, ViewChild, inject }
from '@angular/core';
import { ActivatedRoute, Router } from '@angular/router';
import { AuthService } from '../services/auth.service';
import {
+ MeasurementUpdateAction,
Pipeline,
PipelineCanvasMetadata,
PipelineCanvasMetadataService,
@@ -55,6 +56,7 @@ import { PipelineDetailsToolbarComponent } from
'./components/pipeline-details-t
import { PipelineDetailsExpansionPanelComponent } from
'./components/pipeline-details-expansion-panel/pipeline-details-expansion-panel.component';
import { TranslatePipe } from '@ngx-translate/core';
import { PipelineOperationsService } from
'../pipelines/services/pipeline-operations.service';
+import { MeasurementUpdateDialogComponent } from
'../pipelines/dialog/measurement-update/measurement-update-dialog.component';
@Component({
selector: 'sp-pipeline-details-overview-component',
@@ -102,6 +104,7 @@ export class SpPipelineDetailsComponent implements OnInit,
OnDestroy {
currentUser$: Subscription;
autoRefresh$: Subscription;
private shortcutReg: ShortcutRegistration;
+ private measurementUpdateDialogOpened = false;
@ViewChild('pipelinePreviewComponent')
pipelinePreviewComponent: PipelinePreviewComponent;
@@ -171,6 +174,43 @@ export class SpPipelineDetailsComponent implements OnInit,
OnDestroy {
{ label: this.pipeline.name },
{ label: 'Overview' },
]);
+ this.openMeasurementUpdateDialogIfRequired();
+ }
+
+ openMeasurementUpdateDialogIfRequired(): void {
+ if (
+ this.measurementUpdateDialogOpened ||
+ this.pipeline.healthStatus !== 'HANDLE_MEASUREMENT_UPDATE'
+ ) {
+ return;
+ }
+
+ this.measurementUpdateDialogOpened = true;
+ const dialogRef = this.dialogService.open(
+ MeasurementUpdateDialogComponent,
+ {
+ panelType: PanelType.STANDARD_PANEL,
+ title: 'Measurement update required',
+ width: '50vw',
+ data: {
+ pipeline: this.pipeline,
+ },
+ },
+ );
+
+ dialogRef.afterClosed().subscribe(action => {
+ this.handleMeasurementUpdateAction(action);
+ });
+ }
+
+ handleMeasurementUpdateAction(action?: MeasurementUpdateAction): void {
+ if (action === 'edit-pipeline') {
+ this.pipelineOperationsService.showPipelineInEditor(
+ this.pipeline._id,
+ );
+ } else if (action === 'manage-datasets') {
+ this.router.navigate(['datasets']);
+ }
}
setupAutoRefresh(): void {
diff --git
a/ui/src/app/pipelines/components/pipeline-feature-card/pipeline-preview-meta/pipeline-preview-meta.component.ts
b/ui/src/app/pipelines/components/pipeline-feature-card/pipeline-preview-meta/pipeline-preview-meta.component.ts
index 77ea3b5929..1649099ab2 100644
---
a/ui/src/app/pipelines/components/pipeline-feature-card/pipeline-preview-meta/pipeline-preview-meta.component.ts
+++
b/ui/src/app/pipelines/components/pipeline-feature-card/pipeline-preview-meta/pipeline-preview-meta.component.ts
@@ -73,7 +73,10 @@ export class PipelinePreviewMetaComponent implements OnInit {
if (this.healthStatus === 'OK') {
this.healthStatusTone = 'success';
- } else if (this.healthStatus === 'REQUIRES_ATTENTION') {
+ } else if (
+ this.healthStatus === 'REQUIRES_ATTENTION' ||
+ this.healthStatus === 'HANDLE_MEASUREMENT_UPDATE'
+ ) {
this.healthStatusTone = 'warning';
} else if (this.healthStatus === 'FAILURE') {
this.healthStatusTone = 'error';
diff --git
a/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
b/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
index cb78aaa817..cee5e86ca3 100644
---
a/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
+++
b/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
@@ -51,7 +51,9 @@
}
@if (
pipeline.running &&
- pipeline.healthStatus === 'REQUIRES_ATTENTION'
+ (pipeline.healthStatus === 'REQUIRES_ATTENTION' ||
+ pipeline.healthStatus ===
+ 'HANDLE_MEASUREMENT_UPDATE')
) {
<div class="light light-yellow"></div>
}
diff --git
a/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.html
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.html
new file mode 100644
index 0000000000..cc9064c75f
--- /dev/null
+++
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.html
@@ -0,0 +1,150 @@
+<!--
+~ 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.
+~
+-->
+
+<div class="sp-dialog-container">
+ <div
+ class="sp-dialog-content p-md"
+ data-cy="measurement-update-dialog"
+ fxLayout="column"
+ fxLayoutGap="var(--space-lg)"
+ >
+ <div fxLayout="column" fxLayoutGap="var(--space-xs)">
+ @if (pipeline?.pipelineNotifications?.length > 0) {
+ <div
+ class="measurement-update-notifications mt-sm"
+ fxLayout="column"
+ fxLayoutGap="0.75rem"
+ >
+ @for (
+ notification of pipeline.pipelineNotifications;
+ track notification
+ ) {
+ <sp-alert-banner
+ type="warning"
+ title="Pipeline Notification"
+ [description]="notification"
+ ></sp-alert-banner>
+ }
+ </div>
+ }
+ <p class="measurement-update-description">
+ {{
+ 'A measurement field changed to a data type that conflicts
with existing data. This needs manual adjustments. You can handle this error by
following one of the given instructions:'
+ | translate
+ }}
+ </p>
+ </div>
+
+ <div
+ class="measurement-update-options"
+ fxLayout="column"
+ fxLayoutGap="0.75rem"
+ >
+ <div
+ class="measurement-update-option"
+ fxLayout="row"
+ fxLayoutAlign="start center"
+ fxLayoutGap="0.75rem"
+ >
+ <mat-icon>add_chart</mat-icon>
+ <span
+ class="measurement-update-option-text"
+ fxLayout="column"
+ fxLayoutGap="0.5rem"
+ >
+ <span class="font-bold">
+ {{ 'Save in a new data lake' | translate }}
+ </span>
+ <span class="text-small">
+ {{
+ 'Change the identifier of the "Data Lake" by
opening the pipeline element configuration to save the data in a new "Data
Lake".'
+ | translate
+ }}
+ </span>
+ </span>
+ </div>
+ <div
+ class="measurement-update-option"
+ fxLayout="row"
+ fxLayoutAlign="start center"
+ fxLayoutGap="0.75rem"
+ >
+ <mat-icon>drive_file_rename_outline</mat-icon>
+ <span
+ class="measurement-update-option-text"
+ fxLayout="column"
+ fxLayoutGap="0.5rem"
+ >
+ <span class="font-bold">
+ {{ 'Rename the changed fields' | translate }}
+ </span>
+ <span class="text-small">
+ {{
+ 'Rename the fields with changed data types by
using the "Field Renamer" data processor'
+ | translate
+ }}
+ </span>
+ </span>
+ </div>
+ <div
+ class="measurement-update-option"
+ fxLayout="row"
+ fxLayoutAlign="start center"
+ fxLayoutGap="0.75rem"
+ >
+ <mat-icon>delete_forever</mat-icon>
+ <span
+ class="measurement-update-option-text"
+ fxLayout="column"
+ fxLayoutGap="0.5rem"
+ >
+ <span class="font-bold">
+ {{ 'Delete historical data' | translate }}
+ </span>
+ <span class="text-small">
+ {{
+ 'Truncate the data of the dataset or delete it
completely. All historical data is lost!'
+ | translate
+ }}
+ </span>
+ </span>
+ </div>
+ </div>
+ </div>
+ <mat-divider></mat-divider>
+ <div
+ class="sp-dialog-actions actions-align-right"
+ fxLayout="row"
+ fxLayoutAlign="end center"
+ fxLayoutGap="0.5rem"
+ >
+ <button mat-flat-button class="mat-basic" (click)="close()">
+ {{ 'Close' | translate }}
+ </button>
+ <button mat-flat-button color="accent"
(click)="close('edit-pipeline')">
+ {{ 'Edit pipeline' | translate }}
+ </button>
+ <button
+ mat-flat-button
+ color="accent"
+ (click)="close('manage-datasets')"
+ >
+ {{ 'Manage datasets' | translate }}
+ </button>
+ </div>
+</div>
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.scss
similarity index 69%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
copy to
ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.scss
index 6fac07a465..6370d9f4e0 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineHealthStatus.java
+++
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.scss
@@ -15,10 +15,20 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.model.pipeline;
-public enum PipelineHealthStatus {
- OK,
- REQUIRES_ATTENTION,
- FAILURE
+.measurement-update-description {
+ margin: 0;
+}
+
+.measurement-update-option {
+ height: auto;
+ min-height: calc(var(--space-xl) * 3.5);
+ padding: var(--space-md);
+ text-align: left;
+ border: 1px solid var(--color-border-subtle);
+ border-radius: var(--button-border-radius);
+}
+
+.measurement-update-option-text {
+ white-space: normal;
}
diff --git
a/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.ts
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.ts
new file mode 100644
index 0000000000..8201e57699
--- /dev/null
+++
b/ui/src/app/pipelines/dialog/measurement-update/measurement-update-dialog.component.ts
@@ -0,0 +1,60 @@
+/*
+ * 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 { Component, Input, inject } from '@angular/core';
+import {
+ MeasurementUpdateAction,
+ Pipeline,
+} from '@streampipes/platform-services';
+import { DialogRef, SpAlertBannerComponent } from '@streampipes/shared-ui';
+import { MatButton } from '@angular/material/button';
+import { MatDivider } from '@angular/material/divider';
+import { MatIcon } from '@angular/material/icon';
+import { TranslatePipe } from '@ngx-translate/core';
+import {
+ LayoutAlignDirective,
+ LayoutDirective,
+ LayoutGapDirective,
+} from '@ngbracket/ngx-layout/flex';
+
+@Component({
+ selector: 'sp-measurement-update-dialog',
+ templateUrl: './measurement-update-dialog.component.html',
+ styleUrls: ['./measurement-update-dialog.component.scss'],
+ imports: [
+ LayoutAlignDirective,
+ LayoutDirective,
+ LayoutGapDirective,
+ MatButton,
+ MatDivider,
+ MatIcon,
+ SpAlertBannerComponent,
+ TranslatePipe,
+ ],
+})
+export class MeasurementUpdateDialogComponent {
+ private dialogRef =
+ inject<DialogRef<MeasurementUpdateDialogComponent>>(DialogRef);
+
+ @Input()
+ pipeline: Pipeline;
+
+ close(action?: MeasurementUpdateAction): void {
+ this.dialogRef.close(action);
+ }
+}
