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 f26d7f6343 Add starting error for critical schema change
f26d7f6343 is described below
commit f26d7f634300049e3c57b9052d2643c0a7921555
Author: Sven Oehler <[email protected]>
AuthorDate: Wed May 6 14:42:25 2026 +0200
Add starting error for critical schema change
---
.../streampipes/client/http/HttpRequest.java | 32 ++++++++++++++-
.../dataexplorer/DataExplorerSchemaManagement.java | 24 +++++++++++
.../v2/pipeline/MeasurementChangeDetector.java | 47 +++++++++++++++++++---
.../pipeline/MeasurementChangeValidationStep.java | 35 +++++++++++-----
.../pipeline/update/PipelineUpdateCoordinator.java | 4 +-
.../impl/datalake/DataLakeMeasureResource.java | 17 +++++---
6 files changed, 135 insertions(+), 24 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..6534be2a04 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,33 @@ public abstract class HttpRequest<K, V, T> {
}
}
+ private String makeErrorMessage(StatusLine status,
+ HttpEntity entity) throws IOException {
+ String body = entity != null ? entityAsString(entity) : "";
+ String 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..62eda1e6be 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;
@@ -37,11 +39,13 @@ public class DataExplorerSchemaManagement implements
IDataExplorerSchemaManageme
CRUDStorage<DataLakeMeasure> dataLakeStorage;
private final DataLakePermissionManager permissionManager;
+ private final MeasurementChangeDetector changeDetector;
public DataExplorerSchemaManagement(CRUDStorage<DataLakeMeasure>
dataLakeStorage,
DataLakePermissionManager
permissionManager) {
this.dataLakeStorage = dataLakeStorage;
this.permissionManager = permissionManager;
+ this.changeDetector = new MeasurementChangeDetector();
}
@Override
@@ -86,6 +90,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 +202,23 @@ public class DataExplorerSchemaManagement implements
IDataExplorerSchemaManageme
.values();
return new ArrayList<>(unifiedEventProperties);
}
+
+ private void checkFieldChanges(EventSchema existingSchema, EventSchema
schema) {
+ var criticalFieldChanges =
changeDetector.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-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
index 10847fa8c3..f6d503f9f5 100644
---
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
@@ -29,6 +29,7 @@ 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;
@@ -42,14 +43,20 @@ public class MeasurementChangeDetector {
public boolean hasCriticalMeasurementFieldChange(Pipeline pipeline,
String affectedElementId,
EventSchema
updatedEventSchema) {
+ return !findCriticalMeasurementFieldChanges(pipeline, affectedElementId,
updatedEventSchema).isEmpty();
+ }
+
+ public List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(Pipeline pipeline,
+
String affectedElementId,
+
EventSchema updatedEventSchema) {
if (!hasDatabaseSink(pipeline)) {
- return false;
+ return List.of();
}
return getEventSchema(pipeline, affectedElementId)
.map(existingEventSchema ->
- hasCriticalMeasurementFieldChange(existingEventSchema,
updatedEventSchema))
- .orElse(false);
+ findCriticalMeasurementFieldChanges(existingEventSchema,
updatedEventSchema))
+ .orElse(List.of());
}
private boolean hasDatabaseSink(Pipeline pipeline) {
@@ -74,6 +81,11 @@ public class MeasurementChangeDetector {
public boolean hasCriticalMeasurementFieldChange(EventSchema
existingEventSchema,
EventSchema
updatedEventSchema) {
+ return !findCriticalMeasurementFieldChanges(existingEventSchema,
updatedEventSchema).isEmpty();
+ }
+
+ public List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(EventSchema existingEventSchema,
+
EventSchema updatedEventSchema) {
var existingMeasurementFields = existingEventSchema
.getEventProperties()
.stream()
@@ -88,10 +100,20 @@ public class MeasurementChangeDetector {
.getEventProperties()
.stream()
.filter(this::isMeasurementField)
- .anyMatch(updatedProperty -> {
+ .map(updatedProperty -> {
var existingProperty =
existingMeasurementFields.get(updatedProperty.getRuntimeName());
- return existingProperty != null &&
hasCriticalFieldTypeChange(existingProperty, updatedProperty);
- });
+ 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 boolean isMeasurementField(EventProperty eventProperty) {
@@ -126,6 +148,19 @@ public class MeasurementChangeDetector {
}
}
+ private String toDisplayType(EventProperty eventProperty) {
+ if (eventProperty instanceof EventPropertyPrimitive primitiveProperty) {
+ return primitiveProperty.getRuntimeType();
+ } else {
+ return eventProperty.getClass().getSimpleName();
+ }
+ }
+
+ public record CriticalMeasurementFieldChange(String runtimeName,
+ String existingType,
+ String updatedType) {
+ }
+
private enum StorageType {
INTEGER,
FLOAT,
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
index dce0582d33..0c26312d8e 100644
---
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
@@ -29,6 +29,7 @@ import org.apache.streampipes.model.schema.EventSchema;
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.stream.Collectors;
public class MeasurementChangeValidationStep extends
AbstractPipelineValidationStep {
@@ -50,10 +51,11 @@ public class MeasurementChangeValidationStep extends
AbstractPipelineValidationS
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 instanceof DataSinkInvocation dataSink &&
detector.isDatabaseSink(dataSink)) {
+ var criticalFieldChanges = findCriticalMeasurementFieldChanges(source,
dataSink);
+ if (!criticalFieldChanges.isEmpty()) {
+
validationInfos.add(PipelineElementValidationInfo.error(makeValidationMessage(criticalFieldChanges)));
+ }
}
if (target.getInputStreams() != null && target.getInputStreams().size() >
1) {
@@ -61,14 +63,29 @@ public class MeasurementChangeValidationStep extends
AbstractPipelineValidationS
}
}
- private boolean hasCriticalMeasurementFieldChange(NamedStreamPipesEntity
source,
- DataSinkInvocation
dataSink) {
+ private List<MeasurementChangeDetector.CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(
+ NamedStreamPipesEntity source,
+ DataSinkInvocation dataSink) {
var existingEventSchema = getExistingDataSinkSchema(dataSink);
var updatedEventSchema = getUpdatedSourceSchema(source);
- return existingEventSchema.isPresent()
- && updatedEventSchema.isPresent()
- &&
detector.hasCriticalMeasurementFieldChange(existingEventSchema.get(),
updatedEventSchema.get());
+ if (existingEventSchema.isPresent() && updatedEventSchema.isPresent()) {
+ return
detector.findCriticalMeasurementFieldChanges(existingEventSchema.get(),
updatedEventSchema.get());
+ } else {
+ return List.of();
+ }
+ }
+
+ private String makeValidationMessage(
+ List<MeasurementChangeDetector.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) {
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 f49d05a33b..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
@@ -199,8 +199,8 @@ public class PipelineUpdateCoordinator {
.getPipelineModifications()
.stream()
.flatMap(modification -> modification.getValidationInfos().stream())
- .anyMatch(validationInfo ->
MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED
- .equals(validationInfo.getMessage()));
+ .anyMatch(validationInfo -> validationInfo.getMessage() != null
+ &&
validationInfo.getMessage().startsWith(MeasurementChangeValidationStep.MEASUREMENT_UPDATE_REQUIRED));
}
private List<String> toNotification(PipelineUpdateInfo updateInfo,
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));
+ }
}
/**
