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 9f3e5d1162 Clean ups
9f3e5d1162 is described below
commit 9f3e5d1162311a0d578453b49fe947192236bb9c
Author: Sven Oehler <[email protected]>
AuthorDate: Thu May 7 10:44:12 2026 +0200
Clean ups
---
.../streampipes/client/http/HttpRequest.java | 9 +-
.../dataexplorer/DataExplorerSchemaManagement.java | 4 +-
.../pipeline/CriticalMeasurementFieldChange.java | 24 +++++
.../v2/pipeline/MeasurementChangeDetector.java | 94 +++-------------
.../pipeline/MeasurementChangeValidationStep.java | 38 ++++---
.../manager/matching/v2/pipeline/StorageType.java | 26 +++++
.../v2/pipeline/MeasurementChangeDetectorTest.java | 120 ++++-----------------
.../MeasurementChangeValidationStepTest.java | 19 ++++
8 files changed, 133 insertions(+), 201 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 6534be2a04..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
@@ -124,8 +124,13 @@ 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);
+ String message;
+ if (!status.getReasonPhrase().isBlank()) {
+ message = status.getReasonPhrase();
+ } else {
+ String body = entity != null ? entityAsString(entity) : "";
+ message = extractErrorMessage(body);
+ }
return status.getStatusCode() + " - " + message;
}
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 62eda1e6be..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
@@ -39,13 +39,11 @@ 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
@@ -204,7 +202,7 @@ public class DataExplorerSchemaManagement implements
IDataExplorerSchemaManageme
}
private void checkFieldChanges(EventSchema existingSchema, EventSchema
schema) {
- var criticalFieldChanges =
changeDetector.findCriticalMeasurementFieldChanges(
+ var criticalFieldChanges =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(
existingSchema,
schema
);
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
new file mode 100644
index 0000000000..250b55ab52
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+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
index f6d503f9f5..a7026fb7a6 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
@@ -18,10 +18,6 @@
package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.SpDataStream;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
import org.apache.streampipes.model.schema.EventSchema;
@@ -33,63 +29,19 @@ 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 MeasurementChangeDetector {
+public final class MeasurementChangeDetector {
- private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
-
- 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 List.of();
- }
-
- return getEventSchema(pipeline, affectedElementId)
- .map(existingEventSchema ->
- findCriticalMeasurementFieldChanges(existingEventSchema,
updatedEventSchema))
- .orElse(List.of());
- }
-
- private boolean hasDatabaseSink(Pipeline pipeline) {
- return streamOf(pipeline.getActions())
- .anyMatch(this::isDatabaseSink);
- }
-
- public boolean isDatabaseSink(DataSinkInvocation dataSink) {
- return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId())
- ||
streamOf(dataSink.getCategory()).anyMatch(DataSinkType.DATABASE.name()::equals);
- }
-
- private Optional<EventSchema> getEventSchema(Pipeline pipeline,
- String affectedElementId) {
- return pipeline
- .getStreams()
- .stream()
- .filter(stream -> stream.getElementId().equals(affectedElementId))
- .findFirst()
- .map(SpDataStream::getEventSchema);
- }
-
- public boolean hasCriticalMeasurementFieldChange(EventSchema
existingEventSchema,
- EventSchema
updatedEventSchema) {
- return !findCriticalMeasurementFieldChanges(existingEventSchema,
updatedEventSchema).isEmpty();
+ private MeasurementChangeDetector() {
}
- public List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(EventSchema existingEventSchema,
-
EventSchema updatedEventSchema) {
+ public static List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(
+ EventSchema existingEventSchema,
+ EventSchema updatedEventSchema) {
var existingMeasurementFields = existingEventSchema
.getEventProperties()
.stream()
- .filter(this::isMeasurementField)
+ .filter(MeasurementChangeDetector::isMeasurementField)
.collect(Collectors.toMap(
EventProperty::getRuntimeName,
Function.identity(),
@@ -99,7 +51,7 @@ public class MeasurementChangeDetector {
return updatedEventSchema
.getEventProperties()
.stream()
- .filter(this::isMeasurementField)
+ .filter(MeasurementChangeDetector::isMeasurementField)
.map(updatedProperty -> {
var existingProperty =
existingMeasurementFields.get(updatedProperty.getRuntimeName());
if (existingProperty != null &&
hasCriticalFieldTypeChange(existingProperty, updatedProperty)) {
@@ -116,16 +68,16 @@ public class MeasurementChangeDetector {
.toList();
}
- private boolean isMeasurementField(EventProperty eventProperty) {
+ private static boolean isMeasurementField(EventProperty eventProperty) {
return
!PropertyScope.DIMENSION_PROPERTY.name().equals(eventProperty.getPropertyScope());
}
- private boolean hasCriticalFieldTypeChange(EventProperty existingProperty,
- EventProperty updatedProperty) {
+ private static boolean hasCriticalFieldTypeChange(EventProperty
existingProperty,
+ EventProperty
updatedProperty) {
return toStorageType(existingProperty) != toStorageType(updatedProperty);
}
- private StorageType toStorageType(EventProperty eventProperty) {
+ private static StorageType toStorageType(EventProperty eventProperty) {
if (eventProperty instanceof EventPropertyPrimitive primitiveProperty) {
return toPrimitiveStorageType(primitiveProperty.getRuntimeType());
} else {
@@ -133,7 +85,7 @@ public class MeasurementChangeDetector {
}
}
- private StorageType toPrimitiveStorageType(String runtimeType) {
+ private static StorageType toPrimitiveStorageType(String runtimeType) {
if (XSD.INTEGER.toString().equals(runtimeType)
|| XSD.LONG.toString().equals(runtimeType)) {
return StorageType.INTEGER;
@@ -148,31 +100,11 @@ public class MeasurementChangeDetector {
}
}
- private String toDisplayType(EventProperty eventProperty) {
+ private static 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,
- 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
index 0c26312d8e..4d53c0d49c 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
@@ -18,6 +18,7 @@
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;
@@ -30,28 +31,21 @@ 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 final MeasurementChangeDetector detector;
-
- public MeasurementChangeValidationStep() {
- this(new MeasurementChangeDetector());
- }
-
- MeasurementChangeValidationStep(MeasurementChangeDetector detector) {
- this.detector = detector;
- }
+ 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 &&
detector.isDatabaseSink(dataSink)) {
+ if (target instanceof DataSinkInvocation dataSink &&
isDatabaseSink(dataSink)) {
var criticalFieldChanges = findCriticalMeasurementFieldChanges(source,
dataSink);
if (!criticalFieldChanges.isEmpty()) {
validationInfos.add(PipelineElementValidationInfo.error(makeValidationMessage(criticalFieldChanges)));
@@ -63,21 +57,29 @@ public class MeasurementChangeValidationStep extends
AbstractPipelineValidationS
}
}
- private List<MeasurementChangeDetector.CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(
+ 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
detector.findCriticalMeasurementFieldChanges(existingEventSchema.get(),
updatedEventSchema.get());
+ return MeasurementChangeDetector.findCriticalMeasurementFieldChanges(
+ existingEventSchema.get(),
+ updatedEventSchema.get()
+ );
} else {
return List.of();
}
}
private String makeValidationMessage(
- List<MeasurementChangeDetector.CriticalMeasurementFieldChange>
criticalFieldChanges) {
+ List<CriticalMeasurementFieldChange> criticalFieldChanges) {
return MEASUREMENT_UPDATE_REQUIRED + ": " + criticalFieldChanges
.stream()
.map(change -> "%s (%s -> %s)".formatted(
@@ -106,4 +108,12 @@ public class MeasurementChangeValidationStep extends
AbstractPipelineValidationS
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/StorageType.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/StorageType.java
new file mode 100644
index 0000000000..07f10abcbf
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/StorageType.java
@@ -0,0 +1,26 @@
+/*
+ * 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;
+
+enum StorageType {
+ INTEGER,
+ FLOAT,
+ BOOLEAN,
+ 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
index db952620c4..d5b99a2956 100644
---
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
@@ -18,10 +18,6 @@
package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.SpDataStream;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.model.schema.PropertyScope;
@@ -33,52 +29,10 @@ import java.net.URI;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class MeasurementChangeDetectorTest {
- private static final String STREAM_ID = "stream-1";
- private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
-
- private final MeasurementChangeDetector detector = new
MeasurementChangeDetector();
-
- @Test
- void
hasCriticalMeasurementFieldChange_ShouldIgnoreChangesWithoutDatabaseSink() {
- var pipeline =
makePipeline(makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)));
- var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.STRING));
-
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
-
- assertFalse(result);
- }
-
- @Test
- void
hasCriticalMeasurementFieldChange_ShouldDetectCriticalChangeForDataLakeSink() {
- var pipeline = makePipeline(
- makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
- makeDataLakeSink()
- );
- var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.STRING));
-
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
-
- assertTrue(result);
- }
-
- @Test
- void
hasCriticalMeasurementFieldChange_ShouldDetectCriticalChangeForDatabaseSinkCategory()
{
- var pipeline = makePipeline(
- makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
- makeDatabaseSink()
- );
- var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.BOOLEAN));
-
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
-
- assertTrue(result);
- }
-
@Test
void findCriticalMeasurementFieldChanges_ShouldReturnChangedNamesAndTypes() {
var existingSchema = makeSchema(
@@ -90,7 +44,7 @@ class MeasurementChangeDetectorTest {
makeMeasurementProperty("pressure", XSD.BOOLEAN)
);
- var result = detector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
assertEquals(2, result.size());
assertEquals("temperature", result.get(0).runtimeName());
@@ -102,85 +56,49 @@ class MeasurementChangeDetectorTest {
}
@Test
- void hasCriticalMeasurementFieldChange_ShouldIgnoreIntegerToLongChange() {
- var pipeline = makePipeline(
- makeSchema(makeMeasurementProperty("temperature", XSD.INTEGER)),
- makeDatabaseSink()
- );
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreIntegerToLongChange() {
+ var existingSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.INTEGER));
var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.LONG));
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
- assertFalse(result);
+ assertTrue(result.isEmpty());
}
@Test
- void hasCriticalMeasurementFieldChange_ShouldIgnoreFloatToDoubleChange() {
- var pipeline = makePipeline(
- makeSchema(makeMeasurementProperty("temperature", XSD.FLOAT)),
- makeDatabaseSink()
- );
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreFloatToDoubleChange() {
+ var existingSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.FLOAT));
var updatedSchema = makeSchema(makeMeasurementProperty("temperature",
XSD.DOUBLE));
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
- assertFalse(result);
+ assertTrue(result.isEmpty());
}
@Test
- void hasCriticalMeasurementFieldChange_ShouldIgnoreAddedAndRemovedFields() {
- var pipeline = makePipeline(
- makeSchema(
- makeMeasurementProperty("temperature", XSD.INTEGER),
- makeMeasurementProperty("pressure", XSD.DOUBLE)
- ),
- makeDatabaseSink()
+ 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 = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
- assertFalse(result);
+ assertTrue(result.isEmpty());
}
@Test
- void hasCriticalMeasurementFieldChange_ShouldIgnoreDimensionFieldChanges() {
- var pipeline = makePipeline(
- makeSchema(makeDimensionProperty("machineId", XSD.INTEGER)),
- makeDatabaseSink()
- );
+ void findCriticalMeasurementFieldChanges_ShouldIgnoreDimensionFieldChanges()
{
+ var existingSchema = makeSchema(makeDimensionProperty("machineId",
XSD.INTEGER));
var updatedSchema = makeSchema(makeDimensionProperty("machineId",
XSD.STRING));
- var result = detector.hasCriticalMeasurementFieldChange(pipeline,
STREAM_ID, updatedSchema);
-
- assertFalse(result);
- }
-
- private Pipeline makePipeline(EventSchema eventSchema,
- DataSinkInvocation... actions) {
- var dataStream = new SpDataStream();
- dataStream.setElementId(STREAM_ID);
- dataStream.setEventSchema(eventSchema);
-
- var pipeline = new Pipeline();
- pipeline.setStreams(List.of(dataStream));
- pipeline.setActions(List.of(actions));
- return pipeline;
- }
-
- private DataSinkInvocation makeDataLakeSink() {
- var sink = new DataSinkInvocation();
- sink.setAppId(DATA_LAKE_SINK_APP_ID);
- return sink;
- }
+ var result =
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(existingSchema,
updatedSchema);
- private DataSinkInvocation makeDatabaseSink() {
- var sink = new DataSinkInvocation();
- sink.setCategory(List.of(DataSinkType.DATABASE.name()));
- return sink;
+ assertTrue(result.isEmpty());
}
private EventSchema makeSchema(EventPropertyPrimitive... properties) {
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
index 8668bec49e..559e209fdc 100644
---
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
@@ -39,6 +39,8 @@ 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
@@ -57,6 +59,17 @@ class MeasurementChangeValidationStepTest {
);
}
+ @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)));
@@ -100,6 +113,12 @@ class MeasurementChangeValidationStepTest {
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));
}
