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 04a1ce42f2 Add new pipeline health status
04a1ce42f2 is described below
commit 04a1ce42f2b03ef1ef4a1b5a0075ba159d4f9445
Author: Sven Oehler <[email protected]>
AuthorDate: Wed May 6 10:26:25 2026 +0200
Add new pipeline health status
---
.../model/pipeline/PipelineHealthStatus.java | 1 +
.../matching/PipelineVerificationHandlerV2.java | 13 ++-
.../v2/pipeline/MeasurementChangeDetector.java} | 24 ++++--
.../pipeline/MeasurementChangeValidationStep.java | 92 ++++++++++++++++++++++
.../v2/pipeline/PipelineValidationSteps.java | 1 +
.../pipeline/update/PipelineUpdateCoordinator.java | 33 ++++----
6 files changed, 138 insertions(+), 26 deletions(-)
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-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
similarity index 86%
rename from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
rename to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
index 700c104f6f..10847fa8c3 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineMeasurementChangeDetector.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeDetector.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.manager.pipeline.update;
+package org.apache.streampipes.manager.matching.v2.pipeline;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.SpDataStream;
@@ -29,13 +29,13 @@ 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;
import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
-public class PipelineMeasurementChangeDetector {
+public class MeasurementChangeDetector {
private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
@@ -53,13 +53,13 @@ public class PipelineMeasurementChangeDetector {
}
private boolean hasDatabaseSink(Pipeline pipeline) {
- return pipeline.getActions().stream()
+ return streamOf(pipeline.getActions())
.anyMatch(this::isDatabaseSink);
}
- private boolean isDatabaseSink(DataSinkInvocation dataSink) {
+ public boolean isDatabaseSink(DataSinkInvocation dataSink) {
return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId())
- ||
dataSink.getCategory().stream().anyMatch(DataSinkType.DATABASE.name()::equals);
+ ||
streamOf(dataSink.getCategory()).anyMatch(DataSinkType.DATABASE.name()::equals);
}
private Optional<EventSchema> getEventSchema(Pipeline pipeline,
@@ -72,8 +72,8 @@ public class PipelineMeasurementChangeDetector {
.map(SpDataStream::getEventSchema);
}
- private boolean hasCriticalMeasurementFieldChange(EventSchema
existingEventSchema,
- EventSchema
updatedEventSchema) {
+ public boolean hasCriticalMeasurementFieldChange(EventSchema
existingEventSchema,
+ EventSchema
updatedEventSchema) {
var existingMeasurementFields = existingEventSchema
.getEventProperties()
.stream()
@@ -132,4 +132,12 @@ public class PipelineMeasurementChangeDetector {
BOOLEAN,
STRING
}
+
+ 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/MeasurementChangeValidationStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStep.java
new file mode 100644
index 0000000000..dce0582d33
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/MeasurementChangeValidationStep.java
@@ -0,0 +1,92 @@
+/*
+ * 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.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;
+
+public class MeasurementChangeValidationStep extends
AbstractPipelineValidationStep {
+
+ public static final String MEASUREMENT_UPDATE_REQUIRED =
+ "Measurement field storage type changed. Manual measurement update
handling required";
+
+ private final MeasurementChangeDetector detector;
+
+ public MeasurementChangeValidationStep() {
+ this(new MeasurementChangeDetector());
+ }
+
+ MeasurementChangeValidationStep(MeasurementChangeDetector detector) {
+ this.detector = detector;
+ }
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos) {
+ if (target instanceof DataSinkInvocation dataSink
+ && detector.isDatabaseSink(dataSink)
+ && hasCriticalMeasurementFieldChange(source, dataSink)) {
+
validationInfos.add(PipelineElementValidationInfo.error(MEASUREMENT_UPDATE_REQUIRED));
+ }
+
+ if (target.getInputStreams() != null && target.getInputStreams().size() >
1) {
+ this.visitorHistory.put(target.getDom(), 1);
+ }
+ }
+
+ private boolean hasCriticalMeasurementFieldChange(NamedStreamPipesEntity
source,
+ DataSinkInvocation
dataSink) {
+ var existingEventSchema = getExistingDataSinkSchema(dataSink);
+ var updatedEventSchema = getUpdatedSourceSchema(source);
+
+ return existingEventSchema.isPresent()
+ && updatedEventSchema.isPresent()
+ &&
detector.hasCriticalMeasurementFieldChange(existingEventSchema.get(),
updatedEventSchema.get());
+ }
+
+ 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();
+ }
+ }
+}
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-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 1f81c988ff..f49d05a33b 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;
@@ -49,11 +50,9 @@ public class PipelineUpdateCoordinator {
private static final PipelinesStats PIPELINES_STATS = new PipelinesStats();
private final ExtensionServiceRequestManager requestManager;
- private final PipelineMeasurementChangeDetector
measurementFieldChangeDetector;
public PipelineUpdateCoordinator(ExtensionServiceRequestManager
requestManager) {
this.requestManager = requestManager;
- this.measurementFieldChangeDetector = new
PipelineMeasurementChangeDetector();
}
public void updatePipelines(SpDataStream dataStream) {
@@ -103,22 +102,21 @@ public class PipelineUpdateCoordinator {
}
var storedPipeline =
PipelineManager.getPipeline(pipeline.getPipelineId());
- var hasChangedMeasurementField =
measurementFieldChangeDetector.hasCriticalMeasurementFieldChange(
- storedPipeline,
- affectedElementId,
- updatedEventSchema
- );
var updatedPipeline = updatePipeline(storedPipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
try {
var verificationHandler = new
PipelineVerificationHandlerV2(updatedPipeline, requestManager);
var modificationMessage = verificationHandler.verifyPipeline();
var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
- var modifiedPipeline =
verificationHandler.makeModifiedPipeline().pipeline();
- var canAutoMigrate = canAutoMigrate(modificationMessage) &&
!hasChangedMeasurementField;
+ 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(),
@@ -146,16 +144,10 @@ public class PipelineUpdateCoordinator {
var updateInfos = new ArrayList<PipelineUpdateInfo>();
affectedPipelines.forEach(pipeline -> {
- var hasChangedMeasurementField =
measurementFieldChangeDetector.hasCriticalMeasurementFieldChange(
- pipeline,
- affectedElementId,
- updatedEventSchema
- );
var updatedPipeline = updatePipeline(pipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
try {
var modificationMessage = new
PipelineVerificationHandlerV2(updatedPipeline, requestManager).verifyPipeline();
var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
- updateInfo.setCanAutoMigrate(updateInfo.isCanAutoMigrate() &&
!hasChangedMeasurementField);
updateInfos.add(updateInfo);
} catch (Exception e) {
throw new RuntimeException(e);
@@ -202,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 ->
MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED
+ .equals(validationInfo.getMessage()));
+ }
+
private List<String> toNotification(PipelineUpdateInfo updateInfo,
String updateType) {
var notifications = new ArrayList<String>();
