feat(#1960): Allow modification of adapters with pipelines

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java

@@ -91,15 +91,7 @@ public String addAdapter(AdapterDescription ad,
     return ad.getElementId();
   }
 
-  public void updateAdapter(AdapterDescription ad,
-                            String principalSid)
-      throws AdapterException {
-    // update adapter in database
-    this.adapterResourceManager.encryptAndUpdate(ad);
 
-    // update data source in database
-    this.updateDataSource(ad);
-  }
 
   public AdapterDescription getAdapter(String elementId) throws AdapterException {
     List<AdapterDescription> allAdapters = adapterInstanceStorage.getAllAdapters();
@@ -180,16 +172,6 @@ public void startStreamAdapter(String elementId) throws AdapterException {
     }
   }
 
-  private void updateDataSource(AdapterDescription ad) {
-    // get data source
-    SpDataStream dataStream = this.dataStreamResourceManager.find(ad.getCorrespondingDataStreamElementId());
-
-    SourcesManagement.updateDataStream(ad, dataStream);
-
-    // Update data source in database
-    this.dataStreamResourceManager.update(dataStream);
-  }
-
   private void installDataSource(SpDataStream stream,
                                  String principalSid,
                                  boolean publicElement) throws AdapterException {
@@ -204,5 +186,4 @@ private void installDataSource(SpDataStream stream,
   private IAdapterStorage getAdapterInstanceStorage() {
     return new AdapterInstanceStorageImpl();
   }
-
 }

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java

@@ -0,0 +1,208 @@
+/*
+ * 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.connect.management.management;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.manager.pipeline.PipelineManager;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
+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.resource.management.AdapterResourceManager;
+import org.apache.streampipes.resource.management.DataStreamResourceManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+public class AdapterUpdateManagement {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AdapterUpdateManagement.class);
+
+  private final AdapterMasterManagement adapterMasterManagement;
+  private final AdapterResourceManager adapterResourceManager;
+  private final DataStreamResourceManager dataStreamResourceManager;
+
+  public AdapterUpdateManagement(AdapterMasterManagement adapterMasterManagement) {
+    this.adapterMasterManagement = adapterMasterManagement;
+    this.adapterResourceManager = new SpResourceManager().manageAdapters();
+    this.dataStreamResourceManager = new SpResourceManager().manageDataStreams();
+  }
+
+  public void updateAdapter(AdapterDescription ad)
+      throws AdapterException {
+    // update adapter in database
+    this.adapterResourceManager.encryptAndUpdate(ad);
+    boolean shouldRestart = ad.isRunning();
+
+    if (ad.isRunning()) {
+      this.adapterMasterManagement.stopStreamAdapter(ad.getElementId());
+    }
+
+    // update data source in database
+    this.updateDataSource(ad);
+
+    // update pipelines
+    var affectedPipelines = PipelineManager.getPipelinesContainingElements(ad.getCorrespondingDataStreamElementId());
+
+    affectedPipelines.forEach(p -> {
+      var shouldRestartPipeline = p.isRunning();
+      if (shouldRestartPipeline) {
+        Operations.stopPipeline(p, true);
+      }
+      var storedPipeline = PipelineManager.getPipeline(p.getPipelineId());
+      var pipeline = applyUpdatedDataStream(storedPipeline, ad);
+      try {
+        var modificationMessage = Operations.validatePipeline(pipeline);
+        var updateInfo = makeUpdateInfo(modificationMessage, pipeline);
+        var modifiedPipeline = new PipelineVerificationHandlerV2(pipeline).makeModifiedPipeline();
+        var canAutoMigrate = canAutoMigrate(modificationMessage);
+        if (!canAutoMigrate) {
+          modifiedPipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
+          modifiedPipeline.setPipelineNotifications(toNotification(updateInfo));
+          modifiedPipeline.setValid(false);
+        }
+        StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(modifiedPipeline);
+        if (shouldRestartPipeline && canAutoMigrate) {
+          Operations.startPipeline(PipelineManager.getPipeline(p.getPipelineId()));
+        }
+      } catch (Exception e) {
+        LOG.error("Could not update pipeline {}", pipeline.getName(), e);
+      }
+    });
+
+    if (shouldRestart) {
+      this.adapterMasterManagement.startStreamAdapter(ad.getElementId());
+    }
+  }
+
+  public List<PipelineUpdateInfo> checkPipelineMigrations(AdapterDescription adapterDescription) {
+    var affectedPipelines = PipelineManager
+        .getPipelinesContainingElements(adapterDescription.getCorrespondingDataStreamElementId());
+    var updateInfos = new ArrayList<PipelineUpdateInfo>();
+
+    affectedPipelines.forEach(pipeline -> {
+      var updatedPipeline = applyUpdatedDataStream(pipeline, adapterDescription);
+      try {
+        var modificationMessage = Operations.validatePipeline(updatedPipeline);
+        var updateInfo = makeUpdateInfo(modificationMessage, updatedPipeline);
+        updateInfos.add(updateInfo);
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+
+    return updateInfos;
+  }
+
+  private PipelineUpdateInfo makeUpdateInfo(PipelineModificationMessage modificationMessage,
+                                            Pipeline pipeline) {
+    var updateInfo = new PipelineUpdateInfo();
+    updateInfo.setPipelineId(pipeline.getPipelineId());
+    updateInfo.setPipelineName(pipeline.getName());
+    updateInfo.setCanAutoMigrate(canAutoMigrate(modificationMessage));
+    updateInfo.setValidationInfos(extractModificationWarnings(pipeline, modificationMessage));
+    return updateInfo;
+  }
+
+  private boolean canAutoMigrate(PipelineModificationMessage modificationMessage) {
+    return modificationMessage
+        .getPipelineModifications()
+        .stream()
+        .allMatch(m -> m.isPipelineElementValid() && m.getValidationInfos().isEmpty());
+  }
+
+  private List<String> toNotification(PipelineUpdateInfo updateInfo) {
+    var notifications = new ArrayList<String>();
+    updateInfo.getValidationInfos().keySet().forEach((k) -> {
+      var msg =  updateInfo
+          .getValidationInfos()
+          .get(k)
+          .stream()
+          .map(PipelineElementValidationInfo::getMessage)
+          .toList()
+          .toString();
+      notifications.add(String.format("Adapter modification: %s: %s", k, msg));
+    });
+    return notifications;
+  }
+
+  private Map<String, List<PipelineElementValidationInfo>> extractModificationWarnings(
+      Pipeline pipeline,
+      PipelineModificationMessage modificationMessage) {
+    var infos = new HashMap<String, List<PipelineElementValidationInfo>>();
+    modificationMessage
+        .getPipelineModifications()
+        .stream()
+        .filter(v -> !v.getValidationInfos().isEmpty())
+        .forEach(m -> infos.put(getPipelineElementName(pipeline, m.getElementId()), m.getValidationInfos()));
+
+    return infos;
+  }
+
+  private String getPipelineElementName(Pipeline pipeline,
+                                        String elementId) {
+    return Stream
+        .concat(pipeline.getSepas().stream(), pipeline.getActions().stream())
+        .filter(p -> p.getElementId().equals(elementId))
+        .findFirst()
+        .map(NamedStreamPipesEntity::getName)
+        .orElse(elementId);
+  }
+
+  private Pipeline applyUpdatedDataStream(Pipeline originalPipeline,
+                                          AdapterDescription updatedAdapter) {
+    var updatedStreams = originalPipeline
+        .getStreams()
+        .stream()
+        .peek(s -> {
+          if (s.getElementId().equals(updatedAdapter.getCorrespondingDataStreamElementId())) {
+            s.setEventSchema(updatedAdapter.getEventSchema());
+          }
+        })
+        .toList();
+
+    originalPipeline.setStreams(updatedStreams);
+
+    return originalPipeline;
+  }
+
+  private void updateDataSource(AdapterDescription ad) {
+    // get data source
+    SpDataStream dataStream = this.dataStreamResourceManager.find(ad.getCorrespondingDataStreamElementId());
+
+    SourcesManagement.updateDataStream(ad, dataStream);
+
+    // Update data source in database
+    this.dataStreamResourceManager.update(dataStream);
+  }
+}

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java

@@ -31,6 +31,7 @@
 import org.apache.streampipes.serializers.json.JacksonSerializer;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.fluent.Response;
 import org.apache.http.util.EntityUtils;
@@ -41,21 +42,21 @@
 
 public class GuessManagement {
 
-  private static Logger logger = LoggerFactory.getLogger(GuessManagement.class);
-  private WorkerUrlProvider workerUrlProvider;
+  private static final Logger LOG = LoggerFactory.getLogger(GuessManagement.class);
+  private final WorkerUrlProvider workerUrlProvider;
+  private final ObjectMapper objectMapper;
 
   public GuessManagement() {
     this.workerUrlProvider = new WorkerUrlProvider();
+    this.objectMapper = JacksonSerializer.getObjectMapper();
   }
 
   public GuessSchema guessSchema(AdapterDescription adapterDescription)
       throws ParseException, WorkerAdapterException, NoServiceEndpointsAvailableException, IOException {
-    var workerUrl = workerUrlProvider.getWorkerBaseUrl(adapterDescription.getAppId());
-    workerUrl = workerUrl + WorkerPaths.getGuessSchemaPath();
-
-    var objectMapper = JacksonSerializer.getObjectMapper();
+    var workerUrl = getWorkerUrl(adapterDescription.getAppId());
     var description = objectMapper.writeValueAsString(adapterDescription);
-    logger.info("Guess schema at: " + workerUrl);
+
+    LOG.info("Guess schema at: " + workerUrl);
     Response requestResponse = ExtensionServiceExecutions
         .extServicePostRequest(workerUrl, description)
         .execute();
@@ -71,6 +72,11 @@ public GuessSchema guessSchema(AdapterDescription adapterDescription)
     }
   }
 
+  private String getWorkerUrl(String appId) throws NoServiceEndpointsAvailableException {
+    var baseUrl = workerUrlProvider.getWorkerBaseUrl(appId);
+    return baseUrl + WorkerPaths.getGuessSchemaPath();
+  }
+
   public String performAdapterEventPreview(AdapterEventPreview previewRequest) throws JsonProcessingException {
     return new AdapterEventPreviewPipeline(previewRequest).makePreview();
   }

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ToTransformedSchemaConverter.java

@@ -34,13 +34,14 @@
 import org.apache.streampipes.model.connect.rules.value.UnitTransformRuleDescription;
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventPropertyNested;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.util.Cloner;
 import org.apache.streampipes.sdk.helpers.EpProperties;
-import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.utils.Datatypes;
 
 import java.net.URI;
 import java.util.List;
+import java.util.Objects;
 
 import static org.apache.streampipes.connect.shared.preprocessing.utils.ConversionUtils.findPrimitiveProperty;
 import static org.apache.streampipes.connect.shared.preprocessing.utils.ConversionUtils.findProperty;
@@ -106,7 +107,21 @@ public void visit(AddTimestampRuleDescription rule) {
 
   @Override
   public void visit(AddValueTransformationRuleDescription rule) {
-    this.properties.add(EpProperties.stringEp(Labels.empty(), rule.getRuntimeKey(), ""));
+    var property = new EventPropertyPrimitive();
+    property.setElementId("http://eventProperty.de/staticValue/" + rule.getStaticValue());
+    property.setRuntimeName(rule.getRuntimeKey());
+    property.setRuntimeType(rule.getDatatype());
+    property.setLabel(rule.getLabel());
+    property.setDescription(rule.getDescription());
+    property.setPropertyScope(rule.getPropertyScope().name());
+
+    if (Objects.nonNull(rule.getSemanticType())) {
+      property.setDomainProperties(List.of(URI.create(rule.getSemanticType())));
+    }
+    if (Objects.nonNull(rule.getMeasurementUnit())) {
+      property.setMeasurementUnit(URI.create(rule.getMeasurementUnit()));
+    }
+    this.properties.add(property);
   }
 
   @Override

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/StatelessTransformationRuleGeneratorVisitor.java

@@ -91,7 +91,8 @@ public void visit(AddTimestampRuleDescription ruleDesc) {
   public void visit(AddValueTransformationRuleDescription ruleDesc) {
     rules.add(new AddValueTransformationRule(
         ruleDesc.getRuntimeKey(),
-        ruleDesc.getStaticValue()));
+        ruleDesc.getStaticValue(),
+        ruleDesc.getDatatype()));
   }
 
   @Override

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/schema/AddValueTransformationRule.java

@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.schema;
 
+import org.apache.streampipes.connect.shared.DatatypeUtils;
 import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
 
 import java.util.Map;
@@ -26,15 +27,20 @@ public class AddValueTransformationRule implements TransformationRule {
 
   private final String runtimeKey;
   private final String value;
+  private final String datatype;
 
-  public AddValueTransformationRule(String runtimeKey, String value) {
+  public AddValueTransformationRule(String runtimeKey,
+                                    String value,
+                                    String datatype) {
     this.runtimeKey = runtimeKey;
     this.value = value;
+    this.datatype = datatype;
   }
 
   @Override
   public Map<String, Object> apply(Map<String, Object> event) {
-    event.put(runtimeKey, value);
+    var convertedValue = DatatypeUtils.convertValue(value, datatype);
+    event.put(runtimeKey, convertedValue);
     return event;
   }
 

streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java

@@ -43,7 +43,7 @@ public TimeSeriesStore(Environment environment,
                          DataLakeMeasure measure,
                          boolean enableImageStore) {
 
-    measure = DataExplorerUtils.sanitizeAndRegisterAtDataLake(client, measure);
+    DataExplorerUtils.sanitizeAndRegisterAtDataLake(client, measure);
 
     if (enableImageStore) {
       // TODO check if event properties are replaces correctly

streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagement.java

@@ -60,6 +60,10 @@ public DataLakeMeasure createMeasurement(DataLakeMeasure measure) {
       DataLakeMeasure oldEntry = optional.get();
       if (!compareEventProperties(oldEntry.getEventSchema().getEventProperties(),
           measure.getEventSchema().getEventProperties())) {
+        oldEntry.setEventSchema(measure.getEventSchema());
+        oldEntry.setTimestampField(measure.getTimestampField());
+        oldEntry.setPipelineName(measure.getPipelineName());
+        getDataLakeStorage().updateDataLakeMeasure(oldEntry);
         return oldEntry;
       }
     } else {