implement migration for adapter and pipeline element configurations

streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java

@@ -22,6 +22,7 @@
 import org.apache.streampipes.model.extensions.configuration.SpServiceConfiguration;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
 import org.apache.streampipes.model.function.FunctionDefinition;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
 
 import java.util.List;
 
@@ -39,5 +40,9 @@ public interface IAdminApi {
 
   void deregisterFunction(String functionId);
 
+  void registerAdapterMigrations(List<ModelMigratorConfig> migrationConfigs, String serviceId);
+
+  void registerPipelineElementMigrations(List<ModelMigratorConfig> migratorConfigs, String serviceId);
+
   MessagingSettings getMessagingSettings();
 }

streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java

@@ -24,6 +24,7 @@
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
 import org.apache.streampipes.model.function.FunctionDefinition;
 import org.apache.streampipes.model.message.SuccessMessage;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
 
 import java.util.List;
 
@@ -66,6 +67,19 @@ public void deregisterFunction(String functionId) {
     delete(getDeleteFunctionPath(functionId), SuccessMessage.class);
   }
 
+  /**
+   * Register migration configs {@link ModelMigratorConfig} at the StreamPipes Core service.
+   * @param migrationConfigs list of migration configs to be registered
+   */
+  @Override
+  public void registerAdapterMigrations(List<ModelMigratorConfig> migrationConfigs, String serviceId) {
+    post(getAdapterMigrationPath().addToPath(serviceId), migrationConfigs);
+  }
+
+  public void registerPipelineElementMigrations(List<ModelMigratorConfig> migratorConfigs, String serviceId) {
+    post(getPipelineElementMigrationPath().addToPath(serviceId), migratorConfigs);
+  }
+
   @Override
   public MessagingSettings getMessagingSettings() {
     return getSingle(getMessagingSettingsPath(), MessagingSettings.class);
@@ -98,4 +112,16 @@ private StreamPipesApiPath getFunctionsPath() {
   private StreamPipesApiPath getDeleteFunctionPath(String functionId) {
     return getFunctionsPath().addToPath(functionId);
   }
+
+  private StreamPipesApiPath getAdapterMigrationPath() {
+    return StreamPipesApiPath
+            .fromBaseApiPath()
+            .addToPath("migrations/adapter");
+  }
+
+  private StreamPipesApiPath getPipelineElementMigrationPath() {
+    return StreamPipesApiPath
+            .fromBaseApiPath()
+            .addToPath("migrations/pipeline-element");
+  }
 }

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

@@ -63,8 +63,7 @@ public static void stopStreamAdapter(String baseUrl,
                                        AdapterDescription adapterStreamDescription) throws AdapterException {
     String url = baseUrl + WorkerPaths.getStreamStopPath();
 
-    var ad =
-        getAdapterDescriptionById(new AdapterInstanceStorageImpl(), adapterStreamDescription.getElementId());
+    var ad = getAdapterDescriptionById(new AdapterInstanceStorageImpl(), adapterStreamDescription.getElementId());
 
     stopAdapter(ad, url);
     updateStreamAdapterStatus(adapterStreamDescription.getElementId(), false);
@@ -74,8 +73,8 @@ public static List<AdapterDescription> getAllRunningAdapterInstanceDescriptions(
     try {
       LOG.info("Requesting all running adapter description instances: " + url);
       var responseString = ExtensionServiceExecutions
-          .extServiceGetRequest(url)
-          .execute().returnContent().asString();
+              .extServiceGetRequest(url)
+              .execute().returnContent().asString();
 
       return JacksonSerializer.getObjectMapper().readValue(responseString, List.class);
     } catch (IOException e) {
@@ -84,15 +83,15 @@ public static List<AdapterDescription> getAllRunningAdapterInstanceDescriptions(
     }
   }
 
-  public static void startAdapter(String url,
-                                  AdapterDescription ad) throws AdapterException {
+  private static void startAdapter(String url,
+                                   AdapterDescription ad) throws AdapterException {
     LOG.info("Trying to start adapter on endpoint {} ", url);
     triggerAdapterStateChange(ad, url, "started");
   }
 
 
-  public static void stopAdapter(AdapterDescription ad,
-                                 String url) throws AdapterException {
+  private static void stopAdapter(AdapterDescription ad,
+                                  String url) throws AdapterException {
 
     LOG.info("Trying to stop adapter on endpoint {} ", url);
     triggerAdapterStateChange(ad, url, "stopped");
@@ -134,14 +133,14 @@ private static HttpResponse triggerPost(String url,
   public static RuntimeOptionsResponse getConfiguration(String workerEndpoint,
                                                         String appId,
                                                         RuntimeOptionsRequest runtimeOptionsRequest)
-      throws AdapterException, SpConfigurationException {
+          throws AdapterException, SpConfigurationException {
     String url = workerEndpoint + WorkerPaths.getRuntimeResolvablePath(appId);
 
     try {
       String payload = JacksonSerializer.getObjectMapper().writeValueAsString(runtimeOptionsRequest);
       var response = ExtensionServiceExecutions.extServicePostRequest(url, payload)
-          .execute()
-          .returnResponse();
+              .execute()
+              .returnResponse();
 
       String responseString = IOUtils.toString(response.getEntity().getContent(), StandardCharsets.UTF_8);
 
@@ -163,9 +162,9 @@ public static String getAssets(String workerPath) throws AdapterException {
 
     try {
       return Request.Get(url)
-          .connectTimeout(1000)
-          .socketTimeout(100000)
-          .execute().returnContent().asString();
+              .connectTimeout(1000)
+              .socketTimeout(100000)
+              .execute().returnContent().asString();
     } catch (IOException e) {
       LOG.error(e.getMessage());
       throw new AdapterException("Could not get assets endpoint: " + url);
@@ -178,9 +177,9 @@ public static byte[] getIconAsset(String baseUrl) throws AdapterException {
 
     try {
       byte[] responseString = Request.Get(url)
-          .connectTimeout(1000)
-          .socketTimeout(100000)
-          .execute().returnContent().asBytes();
+              .connectTimeout(1000)
+              .socketTimeout(100000)
+              .execute().returnContent().asBytes();
       return responseString;
     } catch (IOException e) {
       LOG.error(e.getMessage());
@@ -193,9 +192,9 @@ public static String getDocumentationAsset(String baseUrl) throws AdapterExcepti
 
     try {
       return Request.Get(url)
-          .connectTimeout(1000)
-          .socketTimeout(100000)
-          .execute().returnContent().asString();
+              .connectTimeout(1000)
+              .socketTimeout(100000)
+              .execute().returnContent().asString();
     } catch (IOException e) {
       LOG.error(e.getMessage());
       throw new AdapterException("Could not get documentation endpoint: " + url);

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IDataProcessorParameterExtractor.java

@@ -18,11 +18,9 @@
 
 package org.apache.streampipes.extensions.api.extractor;
 
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-
 import java.util.List;
 
-public interface IDataProcessorParameterExtractor extends IParameterExtractor<DataProcessorInvocation> {
+public interface IDataProcessorParameterExtractor extends IParameterExtractor {
   String outputTopic();
 
   List<String> outputKeySelectors();

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IDataSinkParameterExtractor.java

@@ -18,7 +18,5 @@
 
 package org.apache.streampipes.extensions.api.extractor;
 
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-
-public interface IDataSinkParameterExtractor extends IParameterExtractor<DataSinkInvocation>  {
+public interface IDataSinkParameterExtractor extends IParameterExtractor  {
 }

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IParameterExtractor.java

@@ -19,7 +19,6 @@
 package org.apache.streampipes.extensions.api.extractor;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.schema.PropertyScope;
@@ -30,7 +29,7 @@
 import java.io.InputStream;
 import java.util.List;
 
-public interface IParameterExtractor<T extends InvocableStreamPipesEntity> {
+public interface IParameterExtractor {
   String measurementUnit(String runtimeName, Integer streamIndex);
 
   String inputTopic(Integer streamIndex);

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IStaticPropertyExtractor.java

@@ -18,7 +18,5 @@
 
 package org.apache.streampipes.extensions.api.extractor;
 
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-
-public interface IStaticPropertyExtractor extends IParameterExtractor<DataSinkInvocation> {
+public interface IStaticPropertyExtractor extends IParameterExtractor {
 }

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/AdapterMigrator.java

@@ -0,0 +1,33 @@
+/*
+ * 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.extensions.api.migration;
+
+import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+
+public abstract class AdapterMigrator implements ModelMigrator<AdapterDescription, IStaticPropertyExtractor> {
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof ModelMigrator<?, ?>) {
+      return this.config().equals(((ModelMigrator<?, ?>) obj).config());
+    }
+    return false;
+  }
+}

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/DataProcessorMigrator.java

@@ -0,0 +1,34 @@
+/*
+ * 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.extensions.api.migration;
+
+import org.apache.streampipes.extensions.api.extractor.IDataProcessorParameterExtractor;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+
+public abstract class DataProcessorMigrator
+        implements ModelMigrator<DataProcessorInvocation, IDataProcessorParameterExtractor> {
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof ModelMigrator<?, ?>) {
+      return this.config().equals(((ModelMigrator<?, ?>) obj).config());
+    }
+    return false;
+  }
+}

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/DataSinkMigrator.java

@@ -0,0 +1,33 @@
+/*
+ * 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.extensions.api.migration;
+
+import org.apache.streampipes.extensions.api.extractor.IDataSinkParameterExtractor;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+
+public abstract class DataSinkMigrator implements ModelMigrator<DataSinkInvocation, IDataSinkParameterExtractor> {
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof ModelMigrator<?, ?>) {
+      return this.config().equals(((ModelMigrator<?, ?>) obj).config());
+    }
+    return false;
+  }
+}

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/ModelMigrator.java

@@ -0,0 +1,52 @@
+/*
+ * 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.extensions.api.migration;
+
+import org.apache.streampipes.extensions.api.extractor.IParameterExtractor;
+import org.apache.streampipes.model.base.VersionedNamedStreamPipesEntity;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+
+public interface ModelMigrator<
+        T extends VersionedNamedStreamPipesEntity,
+        ExT extends IParameterExtractor
+        > extends Comparable<Object> {
+
+  ModelMigratorConfig config();
+
+  /**
+   * Defines the migration to be performed.
+   *
+   * @param element   Entity to be transformed.
+   * @param extractor Extractor that allows to handle static properties.
+   * @return Result of the migration that describes both outcomes: successful and failed migrations
+   * @throws RuntimeException in case any unexpected error occurs
+   */
+  MigrationResult<T> migrate(T element, ExT extractor) throws RuntimeException;
+
+  @Override
+  default int compareTo(Object o) {
+    if (!(o instanceof ModelMigrator<?, ?>)) {
+      throw new ClassCastException("Given object is not an instance of `ModelMigrator` - "
+              + "only instances of `ModelMigrator` can be compared.");
+    } else {
+      return config().compareTo(((ModelMigrator<?, ?>) o).config());
+    }
+  }
+}

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/pe/param/IParameterGenerator.java

@@ -23,7 +23,7 @@
 
 public interface IParameterGenerator<
     IvT extends InvocableStreamPipesEntity,
-    PeT extends IParameterExtractor<IvT>,
+    PeT extends IParameterExtractor,
     K extends IPipelineElementParameters<IvT, PeT>> {
 
 

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/pe/param/IPipelineElementParameters.java

@@ -28,7 +28,7 @@
 
 public interface IPipelineElementParameters<
     IvT extends InvocableStreamPipesEntity,
-    ExT extends IParameterExtractor<IvT>> {
+    ExT extends IParameterExtractor> {
 
   IvT getModel();
 

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/runtime/ResolvesContainerProvidedOutputStrategy.java

@@ -23,7 +23,7 @@
 import org.apache.streampipes.model.schema.EventSchema;
 
 public interface ResolvesContainerProvidedOutputStrategy<T extends InvocableStreamPipesEntity, K
-    extends IParameterExtractor<T>> {
+    extends IParameterExtractor> {
 
   EventSchema resolveOutputStrategy(T processingElement, K parameterExtractor) throws SpConfigurationException;
 }