This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch
4430-chart-settings-not-correctly-migrated-after-event-schema-changes
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4430-chart-settings-not-correctly-migrated-after-event-schema-changes
by this push:
new f1ed665e78 Implement measurement update management for pipeline changes
f1ed665e78 is described below
commit f1ed665e7822025b7935c3820fa53357a433d984
Author: Sven Oehler <[email protected]>
AuthorDate: Mon May 18 15:52:50 2026 +0200
Implement measurement update management for pipeline changes
---
.../datalake}/CriticalMeasurementFieldChange.java | 5 +-
.../model/datalake/MeasurementUpdateInfo.java | 63 ++++++++++
.../v2/pipeline/MeasurementChangeDetector.java | 1 +
.../pipeline/MeasurementChangeValidationStep.java | 21 +---
.../update/ChartSchemaUpdateCoordinator.java | 45 ++------
.../update/MeasurementUpdateManagement.java | 111 ++++++++++++++++++
.../pipeline/update/MeasurementUpdateUtils.java | 78 +++++++++++++
.../update/MeasurementUpdateManagementTest.java | 127 +++++++++++++++++++++
.../streampipes/rest/impl/PipelineResource.java | 16 +++
9 files changed, 410 insertions(+), 57 deletions(-)
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/CriticalMeasurementFieldChange.java
similarity index 89%
rename from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
rename to
streampipes-model/src/main/java/org/apache/streampipes/model/datalake/CriticalMeasurementFieldChange.java
index 250b55ab52..b461e38926 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CriticalMeasurementFieldChange.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/CriticalMeasurementFieldChange.java
@@ -16,8 +16,11 @@
*
*/
-package org.apache.streampipes.manager.matching.v2.pipeline;
+package org.apache.streampipes.model.datalake;
+import org.apache.streampipes.model.shared.annotation.TsModel;
+
+@TsModel
public record CriticalMeasurementFieldChange(String runtimeName,
String existingType,
String updatedType) {
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/MeasurementUpdateInfo.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/MeasurementUpdateInfo.java
new file mode 100644
index 0000000000..77eb9edb78
--- /dev/null
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/MeasurementUpdateInfo.java
@@ -0,0 +1,63 @@
+/*
+ * 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.model.datalake;
+
+import org.apache.streampipes.model.connect.adapter.ChartSchemaUpdateInfo;
+import org.apache.streampipes.model.shared.annotation.TsModel;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@TsModel
+public class MeasurementUpdateInfo {
+
+ private String measurementName;
+ private List<ChartSchemaUpdateInfo> chartSchemaUpdateInfos;
+ private List<CriticalMeasurementFieldChange> criticalMeasurementFieldChanges;
+
+ public MeasurementUpdateInfo() {
+ this.chartSchemaUpdateInfos = new ArrayList<>();
+ this.criticalMeasurementFieldChanges = new ArrayList<>();
+ }
+
+ public String getMeasurementName() {
+ return measurementName;
+ }
+
+ public void setMeasurementName(String measurementName) {
+ this.measurementName = measurementName;
+ }
+
+ public List<ChartSchemaUpdateInfo> getChartSchemaUpdateInfos() {
+ return chartSchemaUpdateInfos;
+ }
+
+ public void setChartSchemaUpdateInfos(List<ChartSchemaUpdateInfo>
chartSchemaUpdateInfos) {
+ this.chartSchemaUpdateInfos = chartSchemaUpdateInfos;
+ }
+
+ public List<CriticalMeasurementFieldChange>
getCriticalMeasurementFieldChanges() {
+ return criticalMeasurementFieldChanges;
+ }
+
+ public void setCriticalMeasurementFieldChanges(
+ List<CriticalMeasurementFieldChange> criticalMeasurementFieldChanges) {
+ this.criticalMeasurementFieldChanges = criticalMeasurementFieldChanges;
+ }
+}
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 a7026fb7a6..0c0c51e993 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,6 +18,7 @@
package org.apache.streampipes.manager.matching.v2.pipeline;
+import org.apache.streampipes.model.datalake.CriticalMeasurementFieldChange;
import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
import org.apache.streampipes.model.schema.EventSchema;
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 4d53c0d49c..325b821faa 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,10 +18,11 @@
package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.manager.pipeline.update.MeasurementUpdateUtils;
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.datalake.CriticalMeasurementFieldChange;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
@@ -31,21 +32,18 @@ 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)) {
+ if (target instanceof DataSinkInvocation dataSink &&
MeasurementUpdateUtils.isDataLakeSink(dataSink)) {
var criticalFieldChanges = findCriticalMeasurementFieldChanges(source,
dataSink);
if (!criticalFieldChanges.isEmpty()) {
validationInfos.add(PipelineElementValidationInfo.error(makeValidationMessage(criticalFieldChanges)));
@@ -57,11 +55,6 @@ public class MeasurementChangeValidationStep extends
AbstractPipelineValidationS
}
}
- 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) {
@@ -108,12 +101,4 @@ 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/pipeline/update/ChartSchemaUpdateCoordinator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/ChartSchemaUpdateCoordinator.java
index 38015777f3..b7d2272f50 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/ChartSchemaUpdateCoordinator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/ChartSchemaUpdateCoordinator.java
@@ -22,11 +22,9 @@ import
org.apache.streampipes.model.connect.adapter.ChartSchemaUpdateInfo;
import org.apache.streampipes.model.datalake.DataExplorerWidgetHealthStatus;
import org.apache.streampipes.model.datalake.DataExplorerWidgetModel;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
-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.EventSchema;
-import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.storage.api.explorer.IDataExplorerWidgetStorage;
import org.apache.streampipes.storage.management.StorageDispatcher;
@@ -44,8 +42,6 @@ import java.util.stream.Collectors;
public class ChartSchemaUpdateCoordinator {
- private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
- private static final String DATA_LAKE_MEASUREMENT_FIELD = "db_measurement";
private static final String SOURCE_CONFIGS = "sourceConfigs";
private static final String MEASURE_NAME = "measureName";
private static final String MEASURE = "measure";
@@ -69,9 +65,12 @@ public class ChartSchemaUpdateCoordinator {
this.objectMapper = JacksonSerializer.getObjectMapper();
}
- public List<ChartSchemaUpdateInfo> checkChartMigrations(Pipeline pipeline,
- EventSchema
updatedSchema) {
- var measureNames = extractMeasureNames(pipeline);
+ public List<ChartSchemaUpdateInfo> checkChartMigrations(Pipeline pipeline,
EventSchema updatedSchema) {
+ var measureNames = MeasurementUpdateUtils.extractMeasureNames(pipeline);
+ return checkChartMigrations(measureNames, updatedSchema);
+ }
+
+ public List<ChartSchemaUpdateInfo> checkChartMigrations(Set<String>
measureNames, EventSchema updatedSchema) {
return widgetStorage
.findAll()
.stream()
@@ -82,7 +81,7 @@ public class ChartSchemaUpdateCoordinator {
public void updateCharts(Pipeline pipeline,
EventSchema updatedSchema) {
- var measureNames = extractMeasureNames(pipeline);
+ var measureNames = MeasurementUpdateUtils.extractMeasureNames(pipeline);
updateCharts(measureNames, updatedSchema);
}
@@ -248,33 +247,6 @@ public class ChartSchemaUpdateCoordinator {
}
}
- private Set<String> extractMeasureNames(Pipeline pipeline) {
- if (pipeline.getActions() == null) {
- return Set.of();
- }
-
- return pipeline
- .getActions()
- .stream()
- .filter(ChartSchemaUpdateCoordinator::isDataLakeSink)
- .map(this::extractMeasureName)
- .flatMap(Optional::stream)
- .collect(Collectors.toSet());
- }
-
- private Optional<String> extractMeasureName(DataSinkInvocation sink) {
- return Optional
- .ofNullable(sink.getStaticProperties())
- .stream()
- .flatMap(List::stream)
- .filter(property ->
DATA_LAKE_MEASUREMENT_FIELD.equals(property.getInternalName()))
- .filter(FreeTextStaticProperty.class::isInstance)
- .map(FreeTextStaticProperty.class::cast)
- .map(FreeTextStaticProperty::getValue)
- .filter(Objects::nonNull)
- .findFirst();
- }
-
private String getChartTitle(DataExplorerWidgetModel widget) {
var baseAppearanceConfig = widget.getBaseAppearanceConfig();
if (baseAppearanceConfig != null && baseAppearanceConfig.get(WIDGET_TITLE)
instanceof String widgetTitle) {
@@ -286,7 +258,4 @@ public class ChartSchemaUpdateCoordinator {
}
}
- private static boolean isDataLakeSink(DataSinkInvocation dataSink) {
- return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId());
- }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagement.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagement.java
new file mode 100644
index 0000000000..5d253a6738
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagement.java
@@ -0,0 +1,111 @@
+/*
+ * 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.pipeline.update;
+
+import
org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeDetector;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.connect.adapter.ChartSchemaUpdateInfo;
+import org.apache.streampipes.model.datalake.CriticalMeasurementFieldChange;
+import org.apache.streampipes.model.datalake.MeasurementUpdateInfo;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+public class MeasurementUpdateManagement {
+
+ private final ChartSchemaUpdateCoordinator chartSchemaUpdateCoordinator;
+
+ public MeasurementUpdateManagement() {
+ this(new ChartSchemaUpdateCoordinator());
+ }
+
+ MeasurementUpdateManagement(ChartSchemaUpdateCoordinator
chartSchemaUpdateCoordinator) {
+ this.chartSchemaUpdateCoordinator = chartSchemaUpdateCoordinator;
+ }
+
+ public List<MeasurementUpdateInfo> checkPipelineMigrations(String pipelineId,
+ Pipeline
updatedPipeline) {
+ var storedPipeline = StorageDispatcher.INSTANCE.getNoSqlStore()
+ .getPipelineStorageAPI()
+ .getElementById(pipelineId);
+
+ return checkPipelineMigrations(storedPipeline, updatedPipeline);
+ }
+
+ List<MeasurementUpdateInfo> checkPipelineMigrations(Pipeline storedPipeline,
+ Pipeline
updatedPipeline) {
+ return MeasurementUpdateUtils.getDataLakeSinks(updatedPipeline)
+ .stream()
+ .map(updatedSink -> makeUpdateInfo(storedPipeline, updatedSink))
+ .flatMap(Optional::stream)
+ .toList();
+ }
+
+ private Optional<MeasurementUpdateInfo> makeUpdateInfo(Pipeline
storedPipeline,
+ DataSinkInvocation
updatedSink) {
+ var criticalFieldChanges =
findCriticalMeasurementFieldChanges(storedPipeline, updatedSink);
+ var chartSchemaUpdateInfos = findChartSchemaUpdateInfos(updatedSink);
+
+ if (criticalFieldChanges.isEmpty() && chartSchemaUpdateInfos.isEmpty()) {
+ return Optional.empty();
+ } else {
+ var measurementUpdateInfo = new MeasurementUpdateInfo();
+
measurementUpdateInfo.setMeasurementName(MeasurementUpdateUtils.extractMeasureName(updatedSink).orElse(null));
+
measurementUpdateInfo.setCriticalMeasurementFieldChanges(criticalFieldChanges);
+ measurementUpdateInfo.setChartSchemaUpdateInfos(chartSchemaUpdateInfos);
+ return Optional.of(measurementUpdateInfo);
+ }
+ }
+
+ private List<CriticalMeasurementFieldChange>
findCriticalMeasurementFieldChanges(Pipeline storedPipeline,
+
DataSinkInvocation updatedSink) {
+ return MeasurementUpdateUtils.getDataLakeSinkById(storedPipeline,
updatedSink.getElementId())
+ .flatMap(storedSink -> getFirstInputStreamSchema(storedSink)
+ .flatMap(existingSchema -> getFirstInputStreamSchema(updatedSink)
+ .map(updatedSchema ->
MeasurementChangeDetector.findCriticalMeasurementFieldChanges(
+ existingSchema,
+ updatedSchema
+ ))))
+ .orElse(List.of());
+ }
+
+ private List<ChartSchemaUpdateInfo>
findChartSchemaUpdateInfos(DataSinkInvocation updatedSink) {
+ return getFirstInputStreamSchema(updatedSink)
+ .map(updatedSchema ->
chartSchemaUpdateCoordinator.checkChartMigrations(
+
MeasurementUpdateUtils.extractMeasureName(updatedSink).stream().collect(Collectors.toSet()),
+ updatedSchema
+ ))
+ .orElse(List.of());
+ }
+
+ private Optional<EventSchema> getFirstInputStreamSchema(DataSinkInvocation
sink) {
+ return Optional
+ .ofNullable(sink.getInputStreams())
+ .stream()
+ .flatMap(List::stream)
+ .findFirst()
+ .map(SpDataStream::getEventSchema);
+ }
+
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateUtils.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateUtils.java
new file mode 100644
index 0000000000..5145ae8099
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateUtils.java
@@ -0,0 +1,78 @@
+/*
+ * 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.pipeline.update;
+
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public final class MeasurementUpdateUtils {
+
+ public static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+ public static final String DATA_LAKE_MEASUREMENT_FIELD = "db_measurement";
+
+ private MeasurementUpdateUtils() {
+ }
+
+ public static List<DataSinkInvocation> getDataLakeSinks(Pipeline pipeline) {
+ return pipeline.getActions()
+ .stream()
+ .filter(MeasurementUpdateUtils::isDataLakeSink)
+ .toList();
+ }
+
+ public static Optional<DataSinkInvocation> getDataLakeSinkById(Pipeline
pipeline, String id) {
+ return pipeline.getActions()
+ .stream()
+ .filter(MeasurementUpdateUtils::isDataLakeSink)
+ .filter(storedSink -> Objects.equals(storedSink.getElementId(), id))
+ .findFirst();
+ }
+
+ public static Set<String> extractMeasureNames(Pipeline pipeline) {
+ return getDataLakeSinks(pipeline)
+ .stream()
+ .map(MeasurementUpdateUtils::extractMeasureName)
+ .flatMap(Optional::stream)
+ .collect(Collectors.toSet());
+ }
+
+ public static Optional<String> extractMeasureName(DataSinkInvocation sink) {
+ return Optional
+ .ofNullable(sink.getStaticProperties())
+ .stream()
+ .flatMap(List::stream)
+ .filter(property ->
DATA_LAKE_MEASUREMENT_FIELD.equals(property.getInternalName()))
+ .filter(FreeTextStaticProperty.class::isInstance)
+ .map(FreeTextStaticProperty.class::cast)
+ .map(FreeTextStaticProperty::getValue)
+ .filter(Objects::nonNull)
+ .findFirst();
+ }
+
+ public static boolean isDataLakeSink(DataSinkInvocation dataSink) {
+ return DATA_LAKE_SINK_APP_ID.equals(dataSink.getAppId());
+ }
+}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagementTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagementTest.java
new file mode 100644
index 0000000000..e6b1ae5494
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/MeasurementUpdateManagementTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.pipeline.update;
+
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.connect.adapter.ChartSchemaUpdateInfo;
+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.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class MeasurementUpdateManagementTest {
+
+ private static final String DATA_LAKE_SINK_APP_ID =
"org.apache.streampipes.sinks.internal.jvm.datalake";
+ private static final String DATA_LAKE_MEASUREMENT_FIELD = "db_measurement";
+
+ @Test
+ void checkPipelineMigrations_ShouldReturnEmptyListWhenNoWarningsExist() {
+ var chartSchemaUpdateCoordinator =
mock(ChartSchemaUpdateCoordinator.class);
+ var management = new
MeasurementUpdateManagement(chartSchemaUpdateCoordinator);
+ var storedPipeline =
makePipeline(makeDataLakeSink(makeSchema(property("temperature",
XSD.INTEGER))));
+ var updatedPipeline =
makePipeline(makeDataLakeSink(makeSchema(property("temperature", XSD.LONG))));
+
when(chartSchemaUpdateCoordinator.checkChartMigrations(Collections.singleton(any()),
any())).thenReturn(List.of());
+
+ var result = management.checkPipelineMigrations(storedPipeline,
updatedPipeline);
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void
checkPipelineMigrations_ShouldReturnWarningForCriticalMeasurementFieldChange() {
+ var chartSchemaUpdateCoordinator =
mock(ChartSchemaUpdateCoordinator.class);
+ var management = new
MeasurementUpdateManagement(chartSchemaUpdateCoordinator);
+ var storedPipeline =
makePipeline(makeDataLakeSink(makeSchema(property("temperature",
XSD.INTEGER))));
+ var updatedPipeline =
makePipeline(makeDataLakeSink(makeSchema(property("temperature", XSD.STRING))));
+
when(chartSchemaUpdateCoordinator.checkChartMigrations(Collections.singleton(any()),
any())).thenReturn(List.of());
+
+ var result = management.checkPipelineMigrations(storedPipeline,
updatedPipeline);
+
+ assertEquals(1, result.size());
+ assertEquals("measure", result.get(0).getMeasurementName());
+
assertTrue(result.get(0).getCriticalMeasurementFieldChanges().get(0).runtimeName().contains("temperature"));
+ }
+
+ @Test
+ void checkPipelineMigrations_ShouldReturnWarningForAffectedCharts() {
+ var chartSchemaUpdateCoordinator =
mock(ChartSchemaUpdateCoordinator.class);
+ var management = new
MeasurementUpdateManagement(chartSchemaUpdateCoordinator);
+ var storedPipeline =
makePipeline(makeDataLakeSink(makeSchema(property("temperature",
XSD.INTEGER))));
+ var updatedPipeline = makePipeline(makeDataLakeSink(makeSchema()));
+ var chartSchemaUpdateInfo = new ChartSchemaUpdateInfo();
+ chartSchemaUpdateInfo.setChartId("chart-1");
+ chartSchemaUpdateInfo.setChartTitle("Chart");
+ chartSchemaUpdateInfo.setAffectedFields(List.of("temperature"));
+
when(chartSchemaUpdateCoordinator.checkChartMigrations(Collections.singleton(any()),
any())).thenReturn(List.of(chartSchemaUpdateInfo));
+
+ var result = management.checkPipelineMigrations(storedPipeline,
updatedPipeline);
+
+ assertEquals(1, result.size());
+ assertEquals(List.of(chartSchemaUpdateInfo),
result.get(0).getChartSchemaUpdateInfos());
+ }
+
+ private Pipeline makePipeline(DataSinkInvocation dataSink) {
+ var pipeline = new Pipeline();
+ pipeline.setPipelineId("pipeline-1");
+ pipeline.setName("Pipeline");
+ pipeline.setActions(List.of(dataSink));
+ pipeline.setStreams(dataSink.getInputStreams());
+ return pipeline;
+ }
+
+ private DataSinkInvocation makeDataLakeSink(EventSchema eventSchema) {
+ var stream = new SpDataStream();
+ stream.setElementId("stream-1");
+ stream.setEventSchema(eventSchema);
+
+ var sink = new DataSinkInvocation();
+ sink.setElementId("sink-1");
+ sink.setName("Data Lake");
+ sink.setAppId(DATA_LAKE_SINK_APP_ID);
+ sink.setInputStreams(List.of(stream));
+
sink.setStaticProperties(List.of(FreeTextStaticProperty.of(DATA_LAKE_MEASUREMENT_FIELD,
"measure")));
+ return sink;
+ }
+
+ private EventSchema makeSchema(EventPropertyPrimitive... eventProperties) {
+ return new EventSchema(List.of(eventProperties));
+ }
+
+ private EventPropertyPrimitive property(String runtimeName,
+ URI runtimeType) {
+ var property = new EventPropertyPrimitive();
+ property.setRuntimeName(runtimeName);
+ property.setRuntimeType(runtimeType.toString());
+ return property;
+ }
+}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
index 551cf3ec1b..077f0e7a2c 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
@@ -24,9 +24,11 @@ import
org.apache.streampipes.manager.execution.status.PipelineStatusManager;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import
org.apache.streampipes.manager.pipeline.compact.CompactPipelineManagement;
+import
org.apache.streampipes.manager.pipeline.update.MeasurementUpdateManagement;
import org.apache.streampipes.manager.recommender.ElementRecommender;
import org.apache.streampipes.manager.storage.PipelineStorageService;
import org.apache.streampipes.model.client.user.DefaultPrivilege;
+import org.apache.streampipes.model.datalake.MeasurementUpdateInfo;
import org.apache.streampipes.model.message.ErrorMessage;
import org.apache.streampipes.model.message.Message;
import org.apache.streampipes.model.message.Notification;
@@ -83,6 +85,7 @@ public class PipelineResource extends
AbstractAuthGuardedRestResource {
private final CompactPipelineManagement compactPipelineManagement;
private final ExtensionServiceRequestManager requestManager;
+ private final MeasurementUpdateManagement measurementUpdateManagement;
public PipelineResource(ExtensionServiceRequestManager requestManager) {
this.compactPipelineManagement = new CompactPipelineManagement(
@@ -90,6 +93,7 @@ public class PipelineResource extends
AbstractAuthGuardedRestResource {
requestManager
);
this.requestManager = requestManager;
+ this.measurementUpdateManagement = new MeasurementUpdateManagement();
}
@GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
@@ -189,6 +193,18 @@ public class PipelineResource extends
AbstractAuthGuardedRestResource {
return ok(compactPipelineManagement.convertPipeline(pipeline));
}
+ @PutMapping(
+ path = "/{pipelineId}/pipeline-migration-preflight",
+ consumes = MediaType.APPLICATION_JSON_VALUE,
+ produces = MediaType.APPLICATION_JSON_VALUE)
+ @Operation(summary = "Check migration warnings for an existing pipeline
update", tags = {"Pipeline"})
+ @PreAuthorize("this.hasWriteAuthority() and hasPermission(#pipelineId,
'WRITE')")
+ public ResponseEntity<List<MeasurementUpdateInfo>>
performPipelineMigrationPreflight(
+ @PathVariable("pipelineId") String pipelineId,
+ @RequestBody Pipeline pipeline) {
+ return ok(measurementUpdateManagement.checkPipelineMigrations(pipelineId,
pipeline));
+ }
+
@PostMapping(
path = "/recommend/{recId}",
consumes = MediaType.APPLICATION_JSON_VALUE,
