Repository: apache/streampipes
Pull request #2105 (2023-10-30 07:04:47)

Improve startup behaviour
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

* Improve the lifecycle of various requests during the initialization of core and extensions
* Introduce states for extension services REGISTERED, MIGRATING, HEALTHY, UNHEALTHY
* Introduce states for core INSTALLING, MIGRATING; READY
* Add logic to improve the order of initialization actions: Start core, register extensions, perform core migrations, perform extenion migrations

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): <yes/no>

PR introduces (a) deprecation(s): <yes/no>



Patch:
diff --git a/VULNERABILITY.md b/VULNERABILITY.md
index 09a7553baf..8c44874110 100644
--- a/VULNERABILITY.md
+++ b/VULNERABILITY.md
@@ -1,11 +1,14 @@
 <!--\n  ~ Licensed to the Apache Software Foundation (ASF) under one or more\n  ~ contributor license agreements.  See the NOTICE file distributed with\n  ~ this work for additional information regarding copyright ownership.\n  ~ The ASF licenses this file to You under the Apache License, Version 2.0\n  ~ (the "License"); you may not use this file except in compliance with\n  ~ the License.  You may obtain a copy of the License at\n  ~\n  ~    http://www.apache.org/licenses/LICENSE-2.0\n  ~\n  ~ Unless required by applicable law or agreed to in writing, software\n  ~ distributed under the License is distributed on an "AS IS" BASIS,\n  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n  ~ See the License for the specific language governing permissions and\n  ~ limitations under the License.\n  ~\n  -->
 | OSV URL | CVSS | Ecosystem | Package | Version | Source |
 | --- | --- | --- | --- | --- | --- |
+| https://osv.dev/GHSA-57m8-f3v5-hm5m | 5.3 | Maven | io.netty:netty-handler | 4.1.72.Final | pom.xml |
 | https://osv.dev/GHSA-6mjq-h674-j845 | 6.5 | Maven | io.netty:netty-handler | 4.1.72.Final | pom.xml |
+| https://osv.dev/GHSA-crg9-44h2-xw35 | 10 | Maven | org.apache.activemq:activemq-client | 5.18.0 | pom.xml |
 | https://osv.dev/GHSA-cgwf-w82q-5jrr | 5.5 | Maven | org.apache.commons:commons-compress | 1.23.0 | pom.xml |
 | https://osv.dev/GHSA-hr8g-6v94-x4m9 | 5.3 | Maven | org.bouncycastle:bcprov-jdk15on | 1.70 | pom.xml |
-| https://osv.dev/PYSEC-2023-175 |  | PyPI | pillow | 10.0.0 | streampipes-client-python/poetry.lock |
+| https://osv.dev/GHSA-2cqf-6xv9-f22w | 7.5 | Maven | org.elasticsearch:elasticsearch | 6.8.17 | pom.xml |
 | https://osv.dev/GHSA-w596-4wvx-j9j6<br/>https://osv.dev/PYSEC-2022-42969 | 7.5 | PyPI | py | 1.11.0 | streampipes-client-python/poetry.lock |
 | https://osv.dev/GHSA-ww39-953v-wcq6 | 7.5 | npm | glob-parent | 3.1.0 | ui/package-lock.json |
+| https://osv.dev/GHSA-7fh5-64p2-3v2j | 5.3 | npm | postcss | 8.4.21 | ui/package-lock.json |
 | https://osv.dev/GHSA-4943-9vgg-gr5r | 6.1 | npm | quill | 1.3.7 | ui/package-lock.json |
 | https://osv.dev/MAL-2022-6662 |  | npm | tsconfig-package | 7.0.0 | ui/package-lock.json |
diff --git a/pom.xml b/pom.xml
index 2f4eed87af..6b1cc39afe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -185,7 +185,7 @@
 	<netty.version>4.1.72.Final</netty.version>
 	<opencsv.version>5.8</opencsv.version>
 	<plc4x.version>0.10.0</plc4x.version>
-	<protobuf.version>3.24.0</protobuf.version>
+	<protobuf.version>3.25.0</protobuf.version>
 	<proton-j.version>0.34.0</proton-j.version>
 
 	<apache-rat-plugin.version>0.13</apache-rat-plugin.version>
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
index 07d70cf7dd..e59187ea69 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
@@ -34,7 +34,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public class AdapterHealthCheck {
+public class AdapterHealthCheck implements Runnable {
 
   private static final Logger LOG = LoggerFactory.getLogger(AdapterHealthCheck.class);
 
@@ -52,6 +52,11 @@ public class AdapterHealthCheck {
     this.adapterMasterManagement = adapterMasterManagement;
   }
 
+  @Override
+  public void run() {
+    this.checkAndRestoreAdapters();
+  }
+
   /**
    * In this method it is checked which adapters are currently running.
    * Then it calls all workers to validate if the adapter instance is
@@ -114,7 +119,7 @@ public class AdapterHealthCheck {
         allRunningInstancesOfOneWorker.forEach(adapterDescription ->
             allRunningInstancesAdapterDescription.remove(adapterDescription.getElementId()));
       } catch (AdapterException e) {
-        e.printStackTrace();
+        LOG.info("Could not recover adapter at endpoint {} due to {}", adapterEndpointUrl, e.getMessage());
       }
     });
 
@@ -130,10 +135,9 @@ public class AdapterHealthCheck {
           this.adapterMasterManagement.startStreamAdapter(adapterDescription.getElementId());
         }
       } catch (AdapterException e) {
-        LOG.warn("Could not start adapter {}", adapterDescription.getName(), e);
+        LOG.warn("Could not start adapter {} ({})", adapterDescription.getName(), e.getMessage());
       }
     }
 
   }
-
 }
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
index a551201e29..c715459070 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
@@ -73,14 +73,12 @@ public class WorkerRestClient {
 
   public static List<AdapterDescription> getAllRunningAdapterInstanceDescriptions(String url) throws AdapterException {
     try {
-      LOG.info("Requesting all running adapter description instances: " + url);
       var responseString = ExtensionServiceExecutions
               .extServiceGetRequest(url)
               .execute().returnContent().asString();
 
       return JacksonSerializer.getObjectMapper().readValue(responseString, List.class);
     } catch (IOException e) {
-      LOG.error("List of running adapters could not be fetched", e);
       throw new AdapterException("List of running adapters could not be fetched from: " + url);
     }
   }
@@ -112,9 +110,6 @@ public class WorkerRestClient {
         var exception = getSerializer().readValue(responseString, AdapterException.class);
         throw new AdapterException(exception.getMessage(), exception.getCause());
       }
-
-      LOG.info("Adapter {} on endpoint: " + url + " with Response: ", ad.getName() + responseString);
-
     } catch (IOException e) {
       LOG.error("Adapter was not {} successfully", action, e);
       throw new AdapterException("Adapter was not " + action + " successfully with url " + url, e);
@@ -153,8 +148,7 @@ public class WorkerRestClient {
         throw new SpConfigurationException(exception.getMessage(), exception.getCause());
       }
     } catch (IOException e) {
-      e.printStackTrace();
-      throw new AdapterException("Could not resolve runtime configurations from " + url);
+      throw new AdapterException("Could not resolve runtime configurations from " + url, e);
     }
   }
 
@@ -178,11 +172,10 @@ public class WorkerRestClient {
     String url = baseUrl + "/assets/icon";
 
     try {
-      byte[] responseString = Request.Get(url)
+      return Request.Get(url)
               .connectTimeout(1000)
               .socketTimeout(100000)
               .execute().returnContent().asBytes();
-      return responseString;
     } catch (IOException e) {
       LOG.error(e.getMessage());
       throw new AdapterException("Could not get icon endpoint: " + url);
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IAdapterMigrator.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IAdapterMigrator.java
index e7a0996dd9..64ead7f632 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IAdapterMigrator.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IAdapterMigrator.java
@@ -21,5 +21,5 @@ package org.apache.streampipes.extensions.api.migration;
 import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 
-public  interface IAdapterMigrator extends IModelMigrator<AdapterDescription, IStaticPropertyExtractor> {
+public interface IAdapterMigrator extends IModelMigrator<AdapterDescription, IStaticPropertyExtractor> {
 }
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/DataSinkMigrator.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IDataSinkMigrator.java
similarity index 90%
rename from streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/DataSinkMigrator.java
rename to streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IDataSinkMigrator.java
index e30ab00c2a..d1ce71f892 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/DataSinkMigrator.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/migration/IDataSinkMigrator.java
@@ -21,5 +21,5 @@ package org.apache.streampipes.extensions.api.migration;
 import org.apache.streampipes.extensions.api.extractor.IDataSinkParameterExtractor;
 import org.apache.streampipes.model.graph.DataSinkInvocation;
 
-public interface DataSinkMigrator extends IModelMigrator<DataSinkInvocation, IDataSinkParameterExtractor> {
+public interface IDataSinkMigrator extends IModelMigrator<DataSinkInvocation, IDataSinkParameterExtractor> {
 }
diff --git a/streampipes-extensions-management/pom.xml b/streampipes-extensions-management/pom.xml
index a63d88a2f1..7ce180de6c 100644
--- a/streampipes-extensions-management/pom.xml
+++ b/streampipes-extensions-management/pom.xml
@@ -63,31 +63,6 @@
             <artifactId>streampipes-measurement-units</artifactId>
             <version>0.93.0-SNAPSHOT</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-kafka</artifactId>
-            <version>0.93.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-jms</artifactId>
-            <version>0.93.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-mqtt</artifactId>
-            <version>0.93.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-nats</artifactId>
-            <version>0.93.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-pulsar</artifactId>
-            <version>0.93.0-SNAPSHOT</version>
-        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-sdk</artifactId>
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
index 3fe2452559..2e20e276c9 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
@@ -19,21 +19,9 @@
 package org.apache.streampipes.extensions.management.connect.adapter;
 
 import org.apache.streampipes.connect.shared.AdapterPipelineGeneratorBase;
-import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver;
 import org.apache.streampipes.extensions.management.connect.adapter.model.pipeline.AdapterPipeline;
 import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToBrokerAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToJmsAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToKafkaAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToMqttAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToNatsAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToPulsarAdapterSink;
-import org.apache.streampipes.model.configuration.MessagingSettings;
-import org.apache.streampipes.model.configuration.SpProtocol;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.JmsTransportProtocol;
-import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.grounding.MqttTransportProtocol;
-import org.apache.streampipes.model.grounding.PulsarTransportProtocol;
 
 public class AdapterPipelineGenerator extends AdapterPipelineGeneratorBase {
 
@@ -51,31 +39,8 @@ public class AdapterPipelineGenerator extends AdapterPipelineGeneratorBase {
     }
   }
 
-  private SendToBrokerAdapterSink<?> getAdapterSink(AdapterDescription adapterDescription) {
-    var prioritizedProtocol =
-        getMessagingSettings().getPrioritizedProtocols().get(0);
-
-    if (isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) {
-      return new SendToJmsAdapterSink(adapterDescription);
-    } else if (isPrioritized(prioritizedProtocol, KafkaTransportProtocol.class)) {
-      return new SendToKafkaAdapterSink(adapterDescription);
-    } else if (isPrioritized(prioritizedProtocol, MqttTransportProtocol.class)) {
-      return new SendToMqttAdapterSink(adapterDescription);
-    } else if (isPrioritized(prioritizedProtocol, PulsarTransportProtocol.class)) {
-      return new SendToPulsarAdapterSink(adapterDescription);
-    } else {
-      return new SendToNatsAdapterSink(adapterDescription);
-    }
-  }
-
-  private boolean isPrioritized(SpProtocol prioritizedProtocol,
-                                Class<?> protocolClass) {
-    return prioritizedProtocol.getProtocolClass().equals(protocolClass.getCanonicalName());
-  }
-
-  private MessagingSettings getMessagingSettings() {
-    var client = new StreamPipesClientResolver().makeStreamPipesClientInstance();
-    return client.adminApi().getMessagingSettings();
+  private SendToBrokerAdapterSink getAdapterSink(AdapterDescription adapterDescription) {
+    return new SendToBrokerAdapterSink(adapterDescription);
   }
 
   private boolean hasValidGrounding(AdapterDescription adapterDescription) {
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
index b9cebe2581..66983df357 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
@@ -19,51 +19,52 @@ package org.apache.streampipes.extensions.management.connect.adapter.preprocessi
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
 import org.apache.streampipes.extensions.api.monitoring.SpMonitoringManager;
 import org.apache.streampipes.extensions.management.connect.adapter.util.TransportFormatSelector;
 import org.apache.streampipes.extensions.management.monitoring.ExtensionsLogger;
 import org.apache.streampipes.messaging.EventProducer;
+import org.apache.streampipes.messaging.SpProtocolManager;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import java.util.Map;
 
-public abstract class SendToBrokerAdapterSink<T extends TransportProtocol> implements IAdapterPipelineElement {
+public class SendToBrokerAdapterSink implements IAdapterPipelineElement {
 
   protected AdapterDescription adapterDescription;
   protected SpDataFormatDefinition dataFormatDefinition;
-  protected T protocol;
+  protected TransportProtocol protocol;
   private final EventProducer producer;
 
-  public SendToBrokerAdapterSink(AdapterDescription adapterDescription,
-                                 Class<T> protocolClass) {
+  public SendToBrokerAdapterSink(AdapterDescription adapterDescription) {
     this.adapterDescription = adapterDescription;
-    this.protocol = protocolClass.cast(adapterDescription
+    this.protocol = adapterDescription
         .getEventGrounding()
-        .getTransportProtocol());
+        .getTransportProtocol();
 
     if (getEnvironment().getSpDebug().getValueOrDefault()) {
       modifyProtocolForDebugging(this.protocol);
     }
 
-    this.producer = makeProducer(this.protocol);
+    var producerOpt = SpProtocolManager.INSTANCE.findDefinition(this.protocol);
+    if (producerOpt.isPresent()) {
+      this.producer = producerOpt.get().getProducer(this.protocol);
 
-    TransportFormat transportFormat = adapterDescription
-        .getEventGrounding()
-        .getTransportFormats()
-        .get(0);
+      TransportFormat transportFormat = adapterDescription
+          .getEventGrounding()
+          .getTransportFormats()
+          .get(0);
 
-    this.dataFormatDefinition =
-        new TransportFormatSelector(transportFormat).getDataFormatDefinition();
+      this.dataFormatDefinition =
+          new TransportFormatSelector(transportFormat).getDataFormatDefinition();
 
-    try {
       producer.connect();
-    } catch (SpRuntimeException e) {
-      e.printStackTrace();
+    } else {
+      throw new RuntimeException("Could not find protocol");
     }
   }
 
@@ -86,9 +87,12 @@ public abstract class SendToBrokerAdapterSink<T extends TransportProtocol> imple
     producer.publish(event);
   }
 
-  protected abstract EventProducer makeProducer(T protocol);
-
-  public abstract void modifyProtocolForDebugging(T protocol);
+  public void modifyProtocolForDebugging(TransportProtocol protocol) {
+    protocol.setBrokerHostname("localhost");
+    if (protocol instanceof KafkaTransportProtocol) {
+      ((KafkaTransportProtocol) protocol).setKafkaPort(9094);
+    }
+  }
 
   private Environment getEnvironment() {
     return Environments.getEnvironment();
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToJmsAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToJmsAdapterSink.java
deleted file mode 100644
index 98af0b42c2..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToJmsAdapterSink.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.messaging.EventProducer;
-import org.apache.streampipes.messaging.jms.ActiveMQPublisher;
-import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.JmsTransportProtocol;
-
-public class SendToJmsAdapterSink extends SendToBrokerAdapterSink<JmsTransportProtocol>
-    implements IAdapterPipelineElement {
-
-  public SendToJmsAdapterSink(AdapterDescription adapterDescription) {
-    super(adapterDescription, JmsTransportProtocol.class);
-  }
-
-  @Override
-  protected EventProducer makeProducer(JmsTransportProtocol protocol) {
-    return new ActiveMQPublisher(protocol);
-  }
-
-  @Override
-  public void modifyProtocolForDebugging(JmsTransportProtocol protocol) {
-    protocol.setBrokerHostname("localhost");
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToKafkaAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToKafkaAdapterSink.java
deleted file mode 100644
index 5db6d32b2b..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToKafkaAdapterSink.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.messaging.EventProducer;
-import org.apache.streampipes.messaging.kafka.SpKafkaProducer;
-import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-
-public class SendToKafkaAdapterSink extends SendToBrokerAdapterSink<KafkaTransportProtocol>
-    implements IAdapterPipelineElement {
-
-  public SendToKafkaAdapterSink(AdapterDescription adapterDescription) {
-    super(adapterDescription, KafkaTransportProtocol.class);
-  }
-
-  @Override
-  protected EventProducer makeProducer(KafkaTransportProtocol protocol) {
-    return new SpKafkaProducer(protocol);
-  }
-
-  @Override
-  public void modifyProtocolForDebugging(KafkaTransportProtocol protocol) {
-    protocol.setBrokerHostname("localhost");
-    protocol.setKafkaPort(9094);
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToMqttAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToMqttAdapterSink.java
deleted file mode 100644
index d2823c1c5b..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToMqttAdapterSink.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.messaging.EventProducer;
-import org.apache.streampipes.messaging.mqtt.MqttPublisher;
-import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.MqttTransportProtocol;
-
-public class SendToMqttAdapterSink extends SendToBrokerAdapterSink<MqttTransportProtocol>
-    implements IAdapterPipelineElement {
-
-  public SendToMqttAdapterSink(AdapterDescription adapterDescription) {
-    super(adapterDescription, MqttTransportProtocol.class);
-  }
-
-  @Override
-  protected EventProducer makeProducer(MqttTransportProtocol protocol) {
-    return new MqttPublisher(protocol);
-  }
-
-  @Override
-  public void modifyProtocolForDebugging(MqttTransportProtocol transportProtocol) {
-    protocol.setBrokerHostname("localhost");
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToNatsAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToNatsAdapterSink.java
deleted file mode 100644
index 5216ee0b04..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToNatsAdapterSink.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.messaging.EventProducer;
-import org.apache.streampipes.messaging.nats.NatsPublisher;
-import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.NatsTransportProtocol;
-
-public class SendToNatsAdapterSink extends SendToBrokerAdapterSink<NatsTransportProtocol>
-    implements IAdapterPipelineElement {
-
-  public SendToNatsAdapterSink(AdapterDescription adapterDescription) {
-    super(adapterDescription, NatsTransportProtocol.class);
-  }
-
-  @Override
-  protected EventProducer makeProducer(NatsTransportProtocol protocol) {
-    return new NatsPublisher(protocol);
-  }
-
-  @Override
-  public void modifyProtocolForDebugging(NatsTransportProtocol protocol) {
-    protocol.setBrokerHostname("localhost");
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToPulsarAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToPulsarAdapterSink.java
deleted file mode 100644
index b9f3ba56d3..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToPulsarAdapterSink.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.messaging.EventProducer;
-import org.apache.streampipes.messaging.pulsar.PulsarProducer;
-import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.grounding.PulsarTransportProtocol;
-
-public class SendToPulsarAdapterSink extends SendToBrokerAdapterSink<PulsarTransportProtocol>
-    implements IAdapterPipelineElement {
-
-  public SendToPulsarAdapterSink(AdapterDescription adapterDescription) {
-    super(adapterDescription, PulsarTransportProtocol.class);
-  }
-
-  @Override
-  protected EventProducer makeProducer(PulsarTransportProtocol protocol) {
-    return new PulsarProducer(protocol);
-  }
-
-  @Override
-  public void modifyProtocolForDebugging(PulsarTransportProtocol protocol) {
-    protocol.setBrokerHostname("pulsar://localhost:6650");
-  }
-}
diff --git a/streampipes-extensions/pom.xml b/streampipes-extensions/pom.xml
index b14c2471b2..0e89e64f6c 100644
--- a/streampipes-extensions/pom.xml
+++ b/streampipes-extensions/pom.xml
@@ -63,6 +63,7 @@
         <module>streampipes-sinks-databases-jvm</module>
         <module>streampipes-sinks-internal-jvm</module>
         <module>streampipes-sinks-notifications-jvm</module>
+        <module>streampipes-connectors-plc</module>
     </modules>
 
     <properties>
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/pom.xml b/streampipes-extensions/streampipes-connect-adapters-iiot/pom.xml
index 39d51cf868..8d5bff6371 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/pom.xml
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/pom.xml
@@ -59,6 +59,11 @@
             <artifactId>streampipes-connectors-opcua</artifactId>
             <version>0.93.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-connectors-plc</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-connectors-rocketmq</artifactId>
@@ -138,30 +143,6 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-text</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.plc4x</groupId>
-            <artifactId>plc4j-api</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.plc4x</groupId>
-            <artifactId>plc4j-connection-pool</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.plc4x</groupId>
-            <artifactId>plc4j-driver-s7</artifactId>
-            <scope>runtime</scope>
-            <exclusions>
-                <exclusion>
-                    <groupId>com.fasterxml.woodstox</groupId>
-                    <artifactId>woodstox-core</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.plc4x</groupId>
-            <artifactId>plc4j-driver-modbus</artifactId>
-            <scope>runtime</scope>
-        </dependency>
         <dependency>
             <groupId>org.bouncycastle</groupId>
             <artifactId>bcprov-jdk15on</artifactId>
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java
index fa651ea92d..dd48df49be 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java
@@ -19,8 +19,6 @@
 package org.apache.streampipes.connect.iiot;
 
 import org.apache.streampipes.connect.iiot.adapters.iolink.IfmAlMqttAdapter;
-import org.apache.streampipes.connect.iiot.adapters.plc4x.modbus.Plc4xModbusAdapter;
-import org.apache.streampipes.connect.iiot.adapters.plc4x.s7.Plc4xS7Adapter;
 import org.apache.streampipes.connect.iiot.adapters.ros.RosBridgeAdapter;
 import org.apache.streampipes.connect.iiot.adapters.simulator.machine.MachineDataSimulatorAdapter;
 import org.apache.streampipes.connect.iiot.protocol.stream.FileReplayAdapter;
@@ -31,6 +29,9 @@ import org.apache.streampipes.extensions.connectors.mqtt.adapter.MqttProtocol;
 import org.apache.streampipes.extensions.connectors.nats.adapter.NatsProtocol;
 import org.apache.streampipes.extensions.connectors.opcua.adapter.OpcUaAdapter;
 import org.apache.streampipes.extensions.connectors.opcua.migration.OpcUaAdapterMigrationV1;
+import org.apache.streampipes.extensions.connectors.plc.adapter.migration.Plc4xS7AdapterMigrationV1;
+import org.apache.streampipes.extensions.connectors.plc.adapter.modbus.Plc4xModbusAdapter;
+import org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter;
 import org.apache.streampipes.extensions.connectors.pulsar.adapter.PulsarProtocol;
 import org.apache.streampipes.extensions.connectors.rocketmq.adapter.RocketMQProtocol;
 import org.apache.streampipes.extensions.connectors.tubemq.adapter.TubeMQProtocol;
@@ -65,7 +66,10 @@ public class ConnectAdapterIiotInit extends ExtensionsModelSubmitter {
         .registerAdapter(new HttpServerProtocol())
         .registerAdapter(new TubeMQProtocol())
 
-        .registerMigrators(new OpcUaAdapterMigrationV1())
+        .registerMigrators(
+            new OpcUaAdapterMigrationV1(),
+            new Plc4xS7AdapterMigrationV1()
+        )
         .build();
   }
 }
diff --git a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
index 2c3e3b1383..612b3e4905 100644
--- a/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
+++ b/streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/config/SpOpcUaConfigExtractor.java
@@ -24,7 +24,6 @@ import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaUtil;
 
 import java.util.List;
 
-import static org.apache.kafka.common.config.ConfigDef.Type.PASSWORD;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ACCESS_MODE;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.ADAPTER_TYPE;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.AVAILABLE_NODES;
@@ -33,6 +32,7 @@ import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabe
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_PORT;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_SERVER_URL;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.OPC_URL;
+import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PASSWORD;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULLING_INTERVAL;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.PULL_MODE;
 import static org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaLabels.UNAUTHENTICATED;
diff --git a/streampipes-extensions/streampipes-connectors-plc/pom.xml b/streampipes-extensions/streampipes-connectors-plc/pom.xml
new file mode 100644
index 0000000000..e41573b065
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-plc/pom.xml
@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-extensions</artifactId>
+        <version>0.93.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>streampipes-connectors-plc</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-extensions-api</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-extensions-management</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-sdk</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.plc4x</groupId>
+            <artifactId>plc4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.plc4x</groupId>
+            <artifactId>plc4j-connection-pool</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.plc4x</groupId>
+            <artifactId>plc4j-driver-s7</artifactId>
+            <scope>runtime</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.fasterxml.woodstox</groupId>
+                    <artifactId>woodstox-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.plc4x</groupId>
+            <artifactId>plc4j-driver-modbus</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>
diff --git a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/migration/Plc4xS7AdapterMigrationV1.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/migration/Plc4xS7AdapterMigrationV1.java
new file mode 100644
index 0000000000..c88d3f9a5b
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/migration/Plc4xS7AdapterMigrationV1.java
@@ -0,0 +1,87 @@
+/*
+ * 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.connectors.plc.adapter.migration;
+
+import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
+import org.apache.streampipes.extensions.api.migration.IAdapterMigrator;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
+import org.apache.streampipes.sdk.StaticProperties;
+import org.apache.streampipes.sdk.helpers.Alternatives;
+import org.apache.streampipes.sdk.helpers.CodeLanguage;
+import org.apache.streampipes.sdk.helpers.Labels;
+
+import java.util.List;
+
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.CODE_TEMPLATE;
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.PLC_CODE_BLOCK;
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.PLC_NODES;
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.PLC_NODE_INPUT_ALTERNATIVES;
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.PLC_NODE_INPUT_CODE_BLOCK_ALTIVE;
+import static org.apache.streampipes.extensions.connectors.plc.adapter.s7.Plc4xS7Adapter.PLC_NODE_INPUT_COLLECTION_ALTERNATIVE;
+
+public class Plc4xS7AdapterMigrationV1 implements IAdapterMigrator {
+  @Override
+  public ModelMigratorConfig config() {
+    return new ModelMigratorConfig(
+        "org.apache.streampipes.connect.iiot.adapters.plc4x.s7",
+        SpServiceTagPrefix.ADAPTER,
+        0,
+        1
+    );
+  }
+
+  @Override
+  public MigrationResult<AdapterDescription> migrate(AdapterDescription element,
+                                                     IStaticPropertyExtractor extractor) throws RuntimeException {
+    var newConfigs = new java.util.ArrayList<>(element.getConfig().stream().map(config -> {
+      if (isCollectionConfig(config)) {
+        return modifyCollection((CollectionStaticProperty) config);
+      } else {
+        return config;
+      }
+    }).toList());
+
+    newConfigs.removeIf(c -> c.getInternalName().equals(PLC_NODES));
+    element.setConfig(newConfigs);
+
+    return MigrationResult.success(element);
+  }
+
+  private StaticProperty modifyCollection(CollectionStaticProperty collectionConfig) {
+
+    var alternatives = List.of(
+        Alternatives.from(Labels.withId(PLC_NODE_INPUT_COLLECTION_ALTERNATIVE),
+            collectionConfig,
+            true),
+        Alternatives.from(Labels.withId(PLC_NODE_INPUT_CODE_BLOCK_ALTIVE),
+            StaticProperties.codeStaticProperty(Labels.withId(PLC_CODE_BLOCK), CodeLanguage.None, CODE_TEMPLATE))
+    );
+
+    return StaticProperties.alternatives(Labels.withId(PLC_NODE_INPUT_ALTERNATIVES), alternatives);
+  }
+
+  private boolean isCollectionConfig(StaticProperty config) {
+    return config.getInternalName().equals(PLC_NODES);
+  }
+}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/ModbusConfigFile.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/ModbusConfigFile.java
similarity index 95%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/ModbusConfigFile.java
rename to streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/ModbusConfigFile.java
index fbda391ecc..f6e3961126 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/ModbusConfigFile.java
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/ModbusConfigFile.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.iiot.adapters.plc4x.modbus;
+package org.apache.streampipes.extensions.connectors.plc.adapter.modbus;
 
 public class ModbusConfigFile {
 
@@ -58,4 +58,4 @@ public class ModbusConfigFile {
   public void setLogicalAddress(String logicalAddress) {
     this.logicalAddress = logicalAddress;
   }
-}
\ No newline at end of file
+}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/Plc4xModbusAdapter.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/Plc4xModbusAdapter.java
similarity index 99%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/Plc4xModbusAdapter.java
rename to streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/Plc4xModbusAdapter.java
index c755901d58..6938ead546 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/modbus/Plc4xModbusAdapter.java
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/modbus/Plc4xModbusAdapter.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.iiot.adapters.plc4x.modbus;
+package org.apache.streampipes.extensions.connectors.plc.adapter.modbus;
 
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/Plc4xS7Adapter.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
similarity index 64%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/Plc4xS7Adapter.java
rename to streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
index 5913480d80..103e428cb1 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/Plc4xS7Adapter.java
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/Plc4xS7Adapter.java
@@ -16,10 +16,11 @@
  *
  */
 
-package org.apache.streampipes.connect.iiot.adapters.plc4x.s7;
+package org.apache.streampipes.extensions.connectors.plc.adapter.s7;
 
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.extensions.connectors.plc.adapter.s7.config.ConfigurationParser;
 import org.apache.streampipes.extensions.api.connect.IAdapterConfiguration;
 import org.apache.streampipes.extensions.api.connect.IEventCollector;
 import org.apache.streampipes.extensions.api.connect.IPullAdapter;
@@ -33,6 +34,7 @@ import org.apache.streampipes.extensions.management.connect.adapter.util.Polling
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyList;
 import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
@@ -41,11 +43,12 @@ import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
 import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
 import org.apache.streampipes.sdk.builder.adapter.GuessSchemaBuilder;
 import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
+import org.apache.streampipes.sdk.helpers.Alternatives;
+import org.apache.streampipes.sdk.helpers.CodeLanguage;
 import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
 import org.apache.streampipes.sdk.helpers.Options;
 import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.sdk.utils.Datatypes;
 
 import org.apache.plc4x.java.PlcDriverManager;
 import org.apache.plc4x.java.api.PlcConnection;
@@ -53,6 +56,7 @@ import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
 import org.apache.plc4x.java.api.messages.PlcReadRequest;
 import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.apache.plc4x.java.api.value.PlcValue;
 import org.apache.plc4x.java.utils.connectionpool.PooledPlcDriverManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,17 +82,33 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
    */
   private static final String PLC_IP = "plc_ip";
   private static final String PLC_POLLING_INTERVAL = "plc_polling_interval";
-  private static final String PLC_NODES = "plc_nodes";
+  public static final String PLC_NODES = "plc_nodes";
   private static final String PLC_NODE_NAME = "plc_node_name";
   private static final String PLC_NODE_RUNTIME_NAME = "plc_node_runtime_name";
   private static final String PLC_NODE_TYPE = "plc_node_type";
 
+  public static final String PLC_NODE_INPUT_CODE_BLOCK_ALTIVE = "plc_node_input_code_block_altive";
+  public static final String PLC_CODE_BLOCK = "plc_code_block";
+  public static final String PLC_NODE_INPUT_ALTERNATIVES = "plc_node_input_alternatives";
+  public static final String PLC_NODE_INPUT_COLLECTION_ALTERNATIVE = "plc_node_input_collection_alternative";
+
+  public static final String CODE_TEMPLATE = """
+      // This code block can be used to manually specify the addresses of the PLC registers.
+      // The syntax is based on the PLC4X syntax, see [1].
+      // Address Pattern:
+      // propertyName=%{Memory-Area}{start-address}:{Data-Type}[{array-size}]
+
+      temperature=%I0.0:INT
+
+      // [1] https://plc4x.apache.org/users/protocols/s7.html
+      """;
+
   /**
    * Values of user configuration parameters
    */
   private String ip;
   private int pollingInterval;
-  private List<Map<String, String>> nodes;
+  private Map<String, String> nodes;
 
   private PlcDriverManager driverManager;
 
@@ -101,10 +121,8 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
 
   /**
    * This method is executed when the adapter is started. A connection to the PLC is initialized
-   *
-   * @throws AdapterException
    */
-  private void before(IStaticPropertyExtractor extractor) throws AdapterException {
+  private void before(IStaticPropertyExtractor extractor) {
     // Extract user input
     getConfigurations(extractor);
 
@@ -134,17 +152,18 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
     }
   }
 
-  private PlcReadRequest makeReadRequest(PlcConnection plcConnection) throws PlcConnectionException {
+  private PlcReadRequest makeReadRequest(PlcConnection plcConnection) {
     PlcReadRequest.Builder builder = plcConnection.readRequestBuilder();
-    for (Map<String, String> node : this.nodes) {
-      builder.addItem(node.get(PLC_NODE_NAME),
-          node.get(PLC_NODE_NAME) + ":" + node.get(PLC_NODE_TYPE).toUpperCase().replaceAll(" ", "_"));
+
+    for (Map.Entry<String, String> entry : this.nodes.entrySet()) {
+      builder.addItem(entry.getKey(), entry.getValue());
     }
+
     return builder.build();
   }
 
   private void readPlcData(
-      PlcConnection plcConnection, PlcReadResponseHandler handler) throws PlcConnectionException {
+      PlcConnection plcConnection, PlcReadResponseHandler handler) {
     var readRequest = makeReadRequest(plcConnection);
     // Execute the request
     CompletableFuture<? extends PlcReadResponse> asyncResponse = readRequest.execute();
@@ -175,44 +194,28 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
    *
    * @param extractor StaticPropertyExtractor
    */
-  private void getConfigurations(IStaticPropertyExtractor extractor) throws AdapterException {
+  private void getConfigurations(IStaticPropertyExtractor extractor) {
 
     this.ip = extractor.singleValueParameter(PLC_IP, String.class);
     this.pollingInterval = extractor.singleValueParameter(PLC_POLLING_INTERVAL, Integer.class);
 
-    this.nodes = new ArrayList<>();
-    CollectionStaticProperty sp = (CollectionStaticProperty) extractor.getStaticPropertyByName(PLC_NODES);
-
-    for (StaticProperty member : sp.getMembers()) {
-      StaticPropertyExtractor memberExtractor =
-          StaticPropertyExtractor.from(((StaticPropertyGroup) member).getStaticProperties(), new ArrayList<>());
-      Map<String, String> map = new HashMap<>();
-      map.put(PLC_NODE_RUNTIME_NAME, memberExtractor.textParameter(PLC_NODE_RUNTIME_NAME));
-      map.put(PLC_NODE_NAME, memberExtractor.textParameter(PLC_NODE_NAME));
-      map.put(PLC_NODE_TYPE, memberExtractor.selectedSingleValue(PLC_NODE_TYPE, String.class));
-      this.nodes.add(map);
-    }
-  }
+    this.nodes = new HashMap<>();
 
-  /**
-   * Transforms PLC4X data types to datatypes supported in StreamPipes
-   *
-   * @param plcType String
-   * @return Datatypes
-   */
-  private Datatypes getStreamPipesDataType(String plcType) throws AdapterException {
+    var selectedAlternative = extractor.selectedAlternativeInternalId(PLC_NODE_INPUT_ALTERNATIVES);
 
-    String type = plcType.substring(plcType.lastIndexOf(":") + 1);
+    if (selectedAlternative.equals(PLC_NODE_INPUT_COLLECTION_ALTERNATIVE)) {
+      // Alternative Simple
+      var csp = (CollectionStaticProperty) extractor.getStaticPropertyByName(PLC_NODES);
+      this.nodes = getNodeInformationFromCollectionStaticProperty(csp);
 
-    return switch (type) {
-      case "BOOL" -> Datatypes.Boolean;
-      case "BYTE", "REAL" -> Datatypes.Float;
-      case "INT" -> Datatypes.Integer;
-      case "WORD", "TIME_OF_DAY", "DATE", "DATE_AND_TIME", "STRING", "CHAR" -> Datatypes.String;
-      default -> throw new AdapterException("Datatype " + plcType + " is not supported");
-    };
+    } else {
+      // Alternative Advanced
+      var codePropertyInput = extractor.codeblockValue(PLC_CODE_BLOCK);
+      this.nodes = new ConfigurationParser().getNodeInformationFromCodeProperty(codePropertyInput);
+    }
   }
 
+
   @Override
   public void onReadResult(PlcReadResponse response, Throwable throwable) {
     if (throwable != null) {
@@ -226,13 +229,26 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
   }
 
   private Map<String, Object> makeEvent(PlcReadResponse response) {
-    Map<String, Object> event = new HashMap<>();
-    for (Map<String, String> node : this.nodes) {
-      if (response.getResponseCode(node.get(PLC_NODE_NAME)) == PlcResponseCode.OK) {
-        event.put(node.get(PLC_NODE_RUNTIME_NAME), response.getObject(node.get(PLC_NODE_NAME)));
+    var event = new HashMap<String, Object>();
+
+    for (String key : this.nodes.keySet()) {
+      if (response.getResponseCode(key) == PlcResponseCode.OK) {
+
+        // if the response is a list, add each element to the result
+        if (response.getObject(key) instanceof List) {
+          event.put(key,
+              response.getAsPlcValue()
+                  .getValue(key)
+                  .getList().stream()
+                  .map(PlcValue::getObject)
+                  .toList()
+                  .toArray());
+        } else {
+          event.put(key, response.getObject(key));
+        }
       } else {
-        LOG.error("Error[" + node.get(PLC_NODE_NAME) + "]: "
-            + response.getResponseCode(node.get(PLC_NODE_NAME)).name());
+        LOG.error("Error[" + key + "]: "
+                  + response.getResponseCode(key).name());
       }
     }
     return event;
@@ -246,18 +262,24 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
    */
   @Override
   public IAdapterConfiguration declareConfig() {
-    return AdapterConfigurationBuilder.create(ID, Plc4xS7Adapter::new)
+    return AdapterConfigurationBuilder.create(ID, 1, Plc4xS7Adapter::new)
         .withLocales(Locales.EN)
         .withAssets(Assets.DOCUMENTATION, Assets.ICON)
         .withCategory(AdapterType.Manufacturing)
         .requiredTextParameter(Labels.withId(PLC_IP))
         .requiredIntegerParameter(Labels.withId(PLC_POLLING_INTERVAL), 1000)
-        .requiredCollection(Labels.withId(PLC_NODES),
-            StaticProperties.stringFreeTextProperty(Labels.withId(PLC_NODE_RUNTIME_NAME)),
-            StaticProperties.stringFreeTextProperty(Labels.withId(PLC_NODE_NAME)),
-            StaticProperties.singleValueSelection(Labels.withId(PLC_NODE_TYPE),
-                Options.from("Bool", "Byte", "Int", "Word", "Real", "Char", "String", "Date", "Time of day",
-                    "Date and Time")))
+        .requiredAlternatives(
+            Labels.withId(PLC_NODE_INPUT_ALTERNATIVES),
+            Alternatives.from(Labels.withId(PLC_NODE_INPUT_COLLECTION_ALTERNATIVE),
+                StaticProperties.collection(Labels.withId(PLC_NODES),
+                    StaticProperties.stringFreeTextProperty(Labels.withId(PLC_NODE_RUNTIME_NAME)),
+                    StaticProperties.stringFreeTextProperty(Labels.withId(PLC_NODE_NAME)),
+                    StaticProperties.singleValueSelection(Labels.withId(PLC_NODE_TYPE),
+                        Options.from("Bool", "Byte", "Int", "Word", "Real", "Char", "String", "Date", "Time of day",
+                            "Date and Time"))),
+                true),
+            Alternatives.from(Labels.withId(PLC_NODE_INPUT_CODE_BLOCK_ALTIVE),
+                StaticProperties.codeStaticProperty(Labels.withId(PLC_CODE_BLOCK), CodeLanguage.None, CODE_TEMPLATE)))
         .buildConfiguration();
   }
 
@@ -291,17 +313,27 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
       GuessSchemaBuilder builder = GuessSchemaBuilder.create();
       List<EventProperty> allProperties = new ArrayList<>();
 
-      for (Map<String, String> node : this.nodes) {
-        Datatypes datatype = getStreamPipesDataType(node.get(PLC_NODE_TYPE)
-            .toUpperCase()
-            .replaceAll(" ", "_"));
-
-        allProperties.add(
-            PrimitivePropertyBuilder
-                .create(datatype, node.get(PLC_NODE_RUNTIME_NAME))
-                .label(node.get(PLC_NODE_RUNTIME_NAME))
-                .description("")
-                .build());
+      for (Map.Entry<String, String> entry : this.nodes.entrySet()) {
+        var datatype = new ConfigurationParser().getStreamPipesDataType(entry.getValue());
+
+        var primitiveProperty = PrimitivePropertyBuilder
+            .create(datatype, entry.getKey())
+            .label(entry.getKey())
+            .description("")
+            .build();
+
+        // Check if the address configuration is an array
+        var isArray = new ConfigurationParser().isPLCArray(entry.getValue());
+
+        if (isArray) {
+          var propertyList = new EventPropertyList();
+          propertyList.setRuntimeName(entry.getKey());
+          propertyList.setLabel(entry.getKey());
+          propertyList.setEventProperty(primitiveProperty);
+          allProperties.add(propertyList);
+        } else {
+          allProperties.add(primitiveProperty);
+        }
       }
 
       this.before(extractor.getStaticPropertyExtractor());
@@ -315,4 +347,36 @@ public class Plc4xS7Adapter implements StreamPipesAdapter, IPullAdapter, PlcRead
       throw new AdapterException(e.getMessage(), e);
     }
   }
+
+
+  private Map<String, String> getNodeInformationFromCollectionStaticProperty(CollectionStaticProperty csp) {
+    var result = new HashMap<String, String>();
+
+    for (StaticProperty member : csp.getMembers()) {
+      StaticPropertyExtractor memberExtractor =
+          StaticPropertyExtractor.from(((StaticPropertyGroup) member).getStaticProperties(), new ArrayList<>());
+
+      result.put(
+          memberExtractor.textParameter(PLC_NODE_RUNTIME_NAME),
+          getNodeAddress(memberExtractor));
+
+    }
+
+    return result;
+  }
+
+  /**
+   * Takes the members of the static property collection from the UI and creates the PLC4X node address
+   *
+   * @param memberExtractor member of the static property node collection
+   * @return string in the format of PLC4X node address
+   */
+  private String getNodeAddress(StaticPropertyExtractor memberExtractor) {
+    return "%s:%s".formatted(
+        memberExtractor.textParameter(PLC_NODE_NAME),
+        memberExtractor.selectedSingleValue(PLC_NODE_TYPE, String.class)
+            .toUpperCase()
+            .replaceAll(" ", "_"));
+  }
+
 }
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/PlcReadResponseHandler.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/PlcReadResponseHandler.java
similarity index 93%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/PlcReadResponseHandler.java
rename to streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/PlcReadResponseHandler.java
index b9b8f1bfc2..2a4a44c762 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/plc4x/s7/PlcReadResponseHandler.java
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/PlcReadResponseHandler.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.iiot.adapters.plc4x.s7;
+package org.apache.streampipes.extensions.connectors.plc.adapter.s7;
 
 import org.apache.plc4x.java.api.messages.PlcReadResponse;
 
diff --git a/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParser.java b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParser.java
new file mode 100644
index 0000000000..f45b4dbc40
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParser.java
@@ -0,0 +1,101 @@
+/*
+ * 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.connectors.plc.adapter.s7.config;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.sdk.utils.Datatypes;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/**
+ * This is a helper class to parse the user input for the PLC4X S7 adapter
+ */
+public class ConfigurationParser {
+
+  /**
+   * This method takes a string with the PLC configuration and parses the configuration accorting to this pattern:
+   * variableName=value
+   *
+   * @param codePropertyValue code block with the PLC configuration
+   * @return returns a list of Maps, with the variable name as the Key and value as value
+   */
+  public Map<String, String> getNodeInformationFromCodeProperty(String codePropertyValue) {
+    var result = new HashMap<String, String>();
+
+    var lines = codePropertyValue.split("\n");
+
+    // pattern to match "variableName=value"
+    var pattern = Pattern.compile("(\\w+)=(.+)");
+
+    for (String line : lines) {
+      // Remove leading and trailing whitespace
+      line = line.trim();
+
+      // Skip comments
+      if (line.startsWith("//")) {
+        continue;
+      }
+
+      // Check if macher matches and add to results list
+      var matcher = pattern.matcher(line);
+      if (matcher.find()) {
+        result.put(matcher.group(1), matcher.group(2));
+      }
+    }
+
+    return result;
+  }
+
+
+  /**
+   * Transforms PLC4X data types to datatypes supported in StreamPipes
+   *
+   * @param plcType String
+   * @return Datatypes
+   */
+  public Datatypes getStreamPipesDataType(String plcType) throws AdapterException {
+
+    String type = plcType.substring(plcType.lastIndexOf(":") + 1);
+
+    // replace array information from type
+    type = type.replaceAll("\\[.*?\\]", "");
+
+    return switch (type) {
+      case "BOOL" -> Datatypes.Boolean;
+      case "BYTE", "REAL" -> Datatypes.Float;
+      case "INT" -> Datatypes.Integer;
+      case "WORD", "TIME_OF_DAY", "DATE", "DATE_AND_TIME", "STRING", "CHAR" -> Datatypes.String;
+      default -> throw new AdapterException("Datatype " + plcType + " is not supported");
+    };
+  }
+
+  /**
+   * Takes the PLC4X address description and validates if it describes an array
+   *
+   * @param plcType address of the register that should be read
+   * @return whether the address describes an array or not
+   */
+  public boolean isPLCArray(String plcType) {
+    return plcType.matches(".*\\[.*\\].*");
+  }
+
+
+}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/documentation.md b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/documentation.md
similarity index 100%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/documentation.md
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/documentation.md
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/icon.png b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/icon.png
similarity index 100%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/icon.png
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/icon.png
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/strings.en b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/strings.en
similarity index 100%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/strings.en
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.modbus/strings.en
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/documentation.md b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/documentation.md
similarity index 100%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/documentation.md
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/documentation.md
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/icon.png b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/icon.png
similarity index 100%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/icon.png
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/icon.png
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en
similarity index 74%
rename from streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en
rename to streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en
index 705c36e030..5aebb1485d 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en
+++ b/streampipes-extensions/streampipes-connectors-plc/src/main/resources/org.apache.streampipes.connect.iiot.adapters.plc4x.s7/strings.en
@@ -41,4 +41,16 @@ plc_node_name.title=Node Name
 plc_node_name.description=example: %Q0.4
 
 plc_node_type.title=Data Type
-plc_node_type.description=example: bool
\ No newline at end of file
+plc_node_type.description=example: bool
+
+plc_node_input_code_block_altive.title=Advanced
+plc_node_input_code_block_altive.description=
+
+plc_code_block.title=Advanced
+plc_code_block.description=Enter the nodes in the code block below, acording to the described format
+
+plc_node_input_alternatives.title=Register Input
+plc_node_input_alternatives.description=The information about the registers can be entered in simple or advanced mode
+
+plc_node_input_collection_alternative.title=Simple
+plc_node_input_collection_alternative.description=
diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
index e47a0ab0ae..b1caa7a5bf 100644
--- a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
@@ -61,6 +61,11 @@
             <artifactId>streampipes-connectors-influx</artifactId>
             <version>0.93.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-messaging-pulsar</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-processors-filters-siddhi</artifactId>
diff --git a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
index 3b03c5940b..bfaa6dc034 100644
--- a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
@@ -64,6 +64,11 @@
             <classifier>embed</classifier>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-messaging-pulsar</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-messaging-nats</artifactId>
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java
index 3e3bd251d0..6aad62a172 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java
@@ -34,6 +34,7 @@ public class SpCoreConfiguration {
   private GeneralConfig generalConfig;
 
   private boolean isConfigured;
+  private SpCoreConfigurationStatus serviceStatus;
 
   private String assetDir;
   private String filesDir;
@@ -120,4 +121,12 @@ public class SpCoreConfiguration {
   public void setEmailTemplateConfig(EmailTemplateConfig emailTemplateConfig) {
     this.emailTemplateConfig = emailTemplateConfig;
   }
+
+  public SpCoreConfigurationStatus getServiceStatus() {
+    return this.serviceStatus;
+  }
+
+  public void setServiceStatus(SpCoreConfigurationStatus serviceStatus) {
+    this.serviceStatus = serviceStatus;
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfigurationStatus.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfigurationStatus.java
new file mode 100644
index 0000000000..394b9a8aba
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfigurationStatus.java
@@ -0,0 +1,25 @@
+/*
+ * 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.configuration;
+
+public enum SpCoreConfigurationStatus {
+  INSTALLING,
+  MIGRATING,
+  READY
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/migration/MigrationHelpers.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/migration/MigrationHelpers.java
index d518086e9e..2343d41bb0 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/migration/MigrationHelpers.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/migration/MigrationHelpers.java
@@ -40,6 +40,10 @@ public class MigrationHelpers {
     return adapter.get(REV).getAsString();
   }
 
+  public String getAppId(JsonObject adapter) {
+    return adapter.get("properties").getAsJsonObject().get(APP_ID).getAsString();
+  }
+
   public void updateType(JsonObject adapter,
                          String typeFieldName) {
     adapter.add(typeFieldName, new JsonPrimitive(AdapterModels.NEW_MODEL));
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java
index b1490adfaa..8edd9932fa 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java
@@ -36,8 +36,8 @@ public class SpServiceRegistration {
   private int port;
   private List<SpServiceTag> tags;
   private String healthCheckPath;
-  private boolean healthy = true;
   private long firstTimeSeenUnhealthy = 0;
+  private SpServiceStatus status = SpServiceStatus.REGISTERED;
 
   public SpServiceRegistration() {
   }
@@ -133,14 +133,6 @@ public class SpServiceRegistration {
     this.rev = rev;
   }
 
-  public boolean isHealthy() {
-    return healthy;
-  }
-
-  public void setHealthy(boolean healthy) {
-    this.healthy = healthy;
-  }
-
   public String getScheme() {
     return scheme;
   }
@@ -168,4 +160,12 @@ public class SpServiceRegistration {
   public void setSvcType(String svcType) {
     this.svcType = svcType;
   }
+
+  public SpServiceStatus getStatus() {
+    return status;
+  }
+
+  public void setStatus(SpServiceStatus status) {
+    this.status = status;
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceStatus.java b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceStatus.java
new file mode 100644
index 0000000000..56b388d166
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceStatus.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.model.extensions.svcdiscovery;
+
+public enum SpServiceStatus {
+  REGISTERED,
+  MIGRATING,
+  HEALTHY,
+  UNHEALTHY
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataProcessorInvocation.java b/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataProcessorInvocation.java
index 642134bf1e..64aed3b49b 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataProcessorInvocation.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataProcessorInvocation.java
@@ -75,6 +75,7 @@ public class DataProcessorInvocation extends InvocableStreamPipesEntity implemen
   public DataProcessorInvocation(DataProcessorDescription sepa, String domId) {
     this(sepa);
     this.dom = domId;
+    this.serviceTagPrefix = SpServiceTagPrefix.DATA_PROCESSOR;
   }
 
   public DataProcessorInvocation() {
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataSinkInvocation.java b/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataSinkInvocation.java
index 235ca33b44..4476c6651b 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataSinkInvocation.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/graph/DataSinkInvocation.java
@@ -59,6 +59,7 @@ public class DataSinkInvocation extends InvocableStreamPipesEntity {
   public DataSinkInvocation(DataSinkDescription sec, String domId) {
     this(sec);
     this.setDom(domId);
+    this.serviceTagPrefix = SpServiceTagPrefix.DATA_SINK;
   }
 
   public DataSinkInvocation() {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
index 970d81a0d3..cb2d28408a 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
@@ -72,7 +72,7 @@ public class ExtensionsServiceEndpointGenerator {
 
   private String selectService() throws NoServiceEndpointsAvailableException {
     List<String> serviceEndpoints = getServiceEndpoints();
-    if (serviceEndpoints.size() > 0) {
+    if (!serviceEndpoints.isEmpty()) {
       return getServiceEndpoints().get(0);
     } else {
       LOG.error("Could not find any service endpoints for appId {}, serviceTag {}", appId,
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreServiceStatusManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreServiceStatusManager.java
new file mode 100644
index 0000000000..3d2c407ded
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreServiceStatusManager.java
@@ -0,0 +1,59 @@
+/*
+ * 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.health;
+
+import org.apache.streampipes.model.configuration.SpCoreConfiguration;
+import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
+import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CoreServiceStatusManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CoreServiceStatusManager.class);
+
+  private final ISpCoreConfigurationStorage storage;
+
+  public CoreServiceStatusManager(ISpCoreConfigurationStorage storage) {
+    this.storage = storage;
+  }
+
+  public boolean existsConfig() {
+    return storage.exists();
+  }
+
+  public boolean isCoreReady() {
+    return existsConfig() && storage.get().getServiceStatus() == SpCoreConfigurationStatus.READY;
+  }
+
+  public void updateCoreStatus(SpCoreConfigurationStatus status) {
+    var config = storage.get();
+    config.setServiceStatus(status);
+    storage.updateElement(config);
+    logService(config);
+  }
+
+  private void logService(SpCoreConfiguration coreConfig) {
+    LOG.info(
+        "Core is now in {} state",
+        coreConfig.getServiceStatus()
+    );
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
index d8064055a2..66a448be02 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
@@ -45,6 +45,8 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
+import static org.apache.streampipes.manager.pipeline.PipelineManager.getPipeline;
+
 public class PipelineHealthCheck implements Runnable {
 
   private static final Logger LOG = LoggerFactory.getLogger(PipelineHealthCheck.class);
@@ -68,7 +70,7 @@ public class PipelineHealthCheck implements Runnable {
     pipelinesStats.setRunningPipelines(runningPipelines.size());
     pipelinesStats.setStoppedPipelines(pipelinesStats.getAllPipelines() - pipelinesStats.getRunningPipelines());
 
-    if (runningPipelines.size() > 0) {
+    if (!runningPipelines.isEmpty()) {
       Map<String, List<InvocableStreamPipesEntity>> endpointMap = generateEndpointMap();
       List<String> allRunningInstances = findRunningInstances(endpointMap.keySet());
 
@@ -115,15 +117,16 @@ public class PipelineHealthCheck implements Runnable {
           }
         });
         if (shouldUpdatePipeline.get()) {
-          if (failedInstances.size() > 0) {
-            pipeline.setHealthStatus(PipelineHealthStatus.FAILURE);
+          var currentPipeline = getPipeline(pipeline.getPipelineId());
+          if (!failedInstances.isEmpty()) {
+            currentPipeline.setHealthStatus(PipelineHealthStatus.FAILURE);
             pipelinesStats.failedIncrease();
-          } else if (recoveredInstances.size() > 0) {
-            pipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
+          } else if (!recoveredInstances.isEmpty()) {
+            currentPipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
             pipelinesStats.attentionRequiredIncrease();
           }
-          pipeline.setPipelineNotifications(pipelineNotifications);
-          StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(pipeline);
+          currentPipeline.setPipelineNotifications(pipelineNotifications);
+          StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(currentPipeline);
         }
       });
       int healthNum = pipelinesStats.getRunningPipelines() - pipelinesStats.getFailedPipelines()
@@ -233,13 +236,11 @@ public class PipelineHealthCheck implements Runnable {
   }
 
   private List<Pipeline> getAllPipelines() {
-    List<Pipeline> allPipelines = StorageDispatcher
+    return StorageDispatcher
             .INSTANCE
             .getNoSqlStore()
             .getPipelineStorageAPI()
             .getAllPipelines();
-
-    return allPipelines;
   }
 
   private int getElementsCount(List<Pipeline> allPipelines){
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
index 71842d5925..46d2792eaa 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
@@ -21,7 +21,7 @@ package org.apache.streampipes.manager.health;
 
 import org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
-import org.apache.streampipes.storage.api.CRUDStorage;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.apache.http.HttpStatus;
@@ -35,12 +35,13 @@ public class ServiceHealthCheck implements Runnable {
 
   private static final Logger LOG = LoggerFactory.getLogger(ServiceHealthCheck.class);
 
-  private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 60000;
+  private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 20000;
 
-  private final CRUDStorage<String, SpServiceRegistration> storage;
+  private final ServiceRegistrationManager serviceRegistrationManager;
 
   public ServiceHealthCheck() {
-    this.storage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage();
+    var storage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage();
+    this.serviceRegistrationManager = new ServiceRegistrationManager(storage);
   }
 
   @Override
@@ -58,9 +59,8 @@ public class ServiceHealthCheck implements Runnable {
       if (response.returnResponse().getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
         processUnhealthyService(service);
       } else {
-        if (!service.isHealthy()) {
-          service.setHealthy(true);
-          updateService(service);
+        if (service.getStatus() == SpServiceStatus.UNHEALTHY) {
+          serviceRegistrationManager.applyServiceStatus(service.getSvcId(), SpServiceStatus.HEALTHY);
         }
       }
     } catch (IOException e) {
@@ -69,15 +69,16 @@ public class ServiceHealthCheck implements Runnable {
   }
 
   private void processUnhealthyService(SpServiceRegistration service) {
-    if (service.isHealthy()) {
-      service.setHealthy(false);
-      service.setFirstTimeSeenUnhealthy(System.currentTimeMillis());
-      updateService(service);
+    if (service.getStatus() == SpServiceStatus.HEALTHY) {
+      serviceRegistrationManager.applyServiceStatus(
+          service.getSvcId(),
+          SpServiceStatus.UNHEALTHY,
+          System.currentTimeMillis());
     }
     if (shouldDeleteService(service)) {
       LOG.info("Removing service {} which has been unhealthy for more than {} seconds.",
           service.getSvcId(), MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS / 1000);
-      storage.deleteElement(service);
+      serviceRegistrationManager.removeService(service.getSvcId());
     }
   }
 
@@ -86,15 +87,11 @@ public class ServiceHealthCheck implements Runnable {
     return (currentTimeMillis - service.getFirstTimeSeenUnhealthy() > MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS);
   }
 
-  private void updateService(SpServiceRegistration service) {
-    storage.updateElement(service);
-  }
-
   private String makeHealthCheckUrl(SpServiceRegistration service) {
     return service.getServiceUrl() + service.getHealthCheckPath();
   }
 
   private List<SpServiceRegistration> getRegisteredServices() {
-    return storage.getAll();
+    return serviceRegistrationManager.getAllServices();
   }
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceRegistrationManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceRegistrationManager.java
new file mode 100644
index 0000000000..f29b39d11e
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceRegistrationManager.java
@@ -0,0 +1,104 @@
+/*
+ * 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.health;
+
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
+import org.apache.streampipes.storage.api.CRUDStorage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class ServiceRegistrationManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ServiceRegistrationManager.class);
+
+  private final CRUDStorage<String, SpServiceRegistration> storage;
+
+  public ServiceRegistrationManager(CRUDStorage<String, SpServiceRegistration> storage) {
+    this.storage = storage;
+  }
+
+  public void applyServiceStatus(String serviceId,
+                                 SpServiceStatus status,
+                                 long firstTimeSeenUnhealthy) {
+    var serviceRegistration = storage.getElementById(serviceId);
+    serviceRegistration.setFirstTimeSeenUnhealthy(firstTimeSeenUnhealthy);
+    applyServiceStatus(status, serviceRegistration);
+  }
+
+  public void applyServiceStatus(String serviceId,
+                                 SpServiceStatus status) {
+    var serviceRegistration = storage.getElementById(serviceId);
+    applyServiceStatus(status, serviceRegistration);
+  }
+
+  private void applyServiceStatus(SpServiceStatus status,
+                                  SpServiceRegistration serviceRegistration) {
+    serviceRegistration.setStatus(status);
+    storage.updateElement(serviceRegistration);
+    logService(serviceRegistration);
+  }
+
+  public void addService(SpServiceRegistration serviceRegistration,
+                         SpServiceStatus status) {
+    serviceRegistration.setStatus(status);
+    storage.createElement(serviceRegistration);
+    logService(serviceRegistration);
+  }
+
+  public List<SpServiceRegistration> getAllServices() {
+    return storage.getAll();
+  }
+
+  public SpServiceRegistration getService(String serviceId) {
+    return storage.getElementById(serviceId);
+  }
+
+  public boolean isAnyServiceMigrating() {
+    return storage.getAll()
+        .stream()
+        .anyMatch(service -> service.getStatus() == SpServiceStatus.MIGRATING);
+  }
+
+  public void removeService(String serviceId) {
+    var serviceRegistration = storage.getElementById(serviceId);
+    storage.deleteElement(serviceRegistration);
+    LOG.info(
+        "Service {} (id={}) has been removed",
+        serviceRegistration.getSvcGroup(),
+        serviceRegistration.getSvcId())
+    ;
+  }
+
+  public SpServiceStatus getServiceStatus(String serviceId) {
+    return storage.getElementById(serviceId).getStatus();
+  }
+
+  private void logService(SpServiceRegistration serviceRegistration) {
+    LOG.info(
+        "Service {} (id={}) is now in {} state",
+        serviceRegistration.getSvcGroup(),
+        serviceRegistration.getSvcId(),
+        serviceRegistration.getStatus()
+    );
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093.java
new file mode 100644
index 0000000000..d10d279993
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093.java
@@ -0,0 +1,73 @@
+/*
+ * 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.migration;
+
+import org.apache.streampipes.commons.exceptions.SepaParseException;
+import org.apache.streampipes.manager.endpoint.HttpJsonParser;
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.manager.util.AuthTokenUtils;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import org.apache.streampipes.storage.api.IAdapterStorage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+
+import static org.apache.streampipes.manager.migration.MigrationUtils.getRequestUrl;
+
+public class AdapterDescriptionMigration093 extends AbstractMigrationManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AdapterDescriptionMigration093.class);
+
+  private final IAdapterStorage adapterDescriptionStorage;
+
+  public AdapterDescriptionMigration093(IAdapterStorage adapterDescriptionStorage) {
+    this.adapterDescriptionStorage = adapterDescriptionStorage;
+  }
+
+  public void reinstallAdapters(SpServiceRegistration extensionsServiceConfig) {
+    var migrationProvider = AdapterDescriptionMigration093Provider.INSTANCE;
+    if (migrationProvider.hasAppIdsToReinstall()) {
+      var appIdsToReinstall = migrationProvider.getAppIdsToReinstall();
+      var serviceUrl = extensionsServiceConfig.getServiceUrl();
+      extensionsServiceConfig.getTags()
+          .stream()
+          .filter(tag -> tag.getPrefix() == SpServiceTagPrefix.ADAPTER)
+          .filter(tag -> appIdsToReinstall.contains(tag.getValue()))
+          .forEach(tag -> {
+            var appId = tag.getValue();
+            try {
+              if (adapterDescriptionStorage.getAdaptersByAppId(appId).isEmpty()) {
+                var requestUrl = getRequestUrl(SpServiceTagPrefix.ADAPTER, appId, serviceUrl);
+                var entityPayload = HttpJsonParser.getContentFromUrl(URI.create(requestUrl));
+                Operations.verifyAndAddElement(
+                    entityPayload,
+                    AuthTokenUtils.getAuthTokenForCurrentUser(),
+                    true);
+              }
+            } catch (IOException | SepaParseException e) {
+              LOG.warn("Could not reinstall adapter description {}", appId);
+            }
+          });
+    }
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093Provider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093Provider.java
new file mode 100644
index 0000000000..8ad025d5c4
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AdapterDescriptionMigration093Provider.java
@@ -0,0 +1,45 @@
+/*
+ * 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.migration;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public enum AdapterDescriptionMigration093Provider {
+
+  INSTANCE;
+
+  private final List<String> appIdsToReinstall;
+
+  AdapterDescriptionMigration093Provider() {
+    this.appIdsToReinstall = new ArrayList<>();
+  }
+
+  public void addAppId(String appId) {
+    this.appIdsToReinstall.add(appId);
+  }
+
+  public List<String> getAppIdsToReinstall() {
+    return appIdsToReinstall;
+  }
+
+  public boolean hasAppIdsToReinstall() {
+    return !appIdsToReinstall.isEmpty();
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
index 28338e1ac0..166d24c8e3 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
@@ -38,6 +38,7 @@ import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Stream;
 
 import static org.apache.streampipes.manager.migration.MigrationUtils.getApplicableMigration;
 
@@ -60,77 +61,89 @@ public class PipelineElementMigrationManager extends AbstractMigrationManager im
   @Override
   public void handleMigrations(SpServiceRegistration extensionsServiceConfig,
                                List<ModelMigratorConfig> migrationConfigs) {
+    if (!migrationConfigs.isEmpty()) {
+      LOG.info("Updating pipeline element descriptions by replacement...");
+      updateDescriptions(migrationConfigs, extensionsServiceConfig.getServiceUrl());
+      LOG.info("Pipeline element descriptions are up to date.");
 
-    LOG.info("Updating pipeline element descriptions by replacement...");
-    updateDescriptions(migrationConfigs, extensionsServiceConfig.getServiceUrl());
-    LOG.info("Pipeline element descriptions are up to date.");
-
-    LOG.info("Received {} pipeline element migrations from extension service {}.",
-        migrationConfigs.size(),
-        extensionsServiceConfig.getServiceUrl());
-    var availablePipelines = pipelineStorage.getAllPipelines();
-    if (!availablePipelines.isEmpty()) {
-      LOG.info("Found {} available pipelines. Checking pipelines for applicable migrations...",
-          availablePipelines.size()
-      );
-    }
+      LOG.info("Received {} pipeline element migrations from extension service {}.",
+          migrationConfigs.size(),
+          extensionsServiceConfig.getServiceUrl());
+      var availablePipelines = pipelineStorage.getAllPipelines();
+      if (!availablePipelines.isEmpty()) {
+        LOG.info("Found {} available pipelines. Checking pipelines for applicable migrations...",
+            availablePipelines.size()
+        );
+      }
 
-    for (var pipeline : availablePipelines) {
-      List<MigrationResult<?>> failedMigrations = new ArrayList<>();
-
-      var migratedDataProcessors = pipeline.getSepas()
-          .stream()
-          .map(processor -> {
-            if (getApplicableMigration(processor, migrationConfigs).isPresent()) {
-              return migratePipelineElement(
-                  processor,
-                  migrationConfigs,
-                  String.format("%s/%s/processor",
-                      extensionsServiceConfig.getServiceUrl(),
-                      MIGRATION_ENDPOINT
-                  ),
-                  failedMigrations
-              );
-            } else {
-              LOG.info("No migration applicable for data processor '{}'.", processor.getElementId());
-              return processor;
-            }
-          })
-          .toList();
-      pipeline.setSepas(migratedDataProcessors);
-
-      var migratedDataSinks = pipeline.getActions()
-          .stream()
-          .map(sink -> {
-            if (getApplicableMigration(sink, migrationConfigs).isPresent()) {
-              return migratePipelineElement(
-                  sink,
-                  migrationConfigs,
-                  String.format("%s/%s/sink",
-                      extensionsServiceConfig.getServiceUrl(),
-                      MIGRATION_ENDPOINT
-                  ),
-                  failedMigrations
-              );
-            } else {
-              LOG.info("No migration applicable for data sink '{}'.", sink.getElementId());
-              return sink;
-            }
-          })
-          .toList();
-      pipeline.setActions(migratedDataSinks);
-
-      pipelineStorage.updatePipeline(pipeline);
-
-      if (failedMigrations.isEmpty()) {
-        LOG.info("Migration for pipeline successfully completed.");
-      } else {
-        // pass most recent version of pipeline
-        handleFailedMigrations(pipelineStorage.getPipeline(pipeline.getPipelineId()), failedMigrations);
+      for (var pipeline : availablePipelines) {
+        if (shouldMigratePipeline(pipeline, migrationConfigs)) {
+          List<MigrationResult<?>> failedMigrations = new ArrayList<>();
+
+          var migratedDataProcessors = pipeline.getSepas()
+              .stream()
+              .map(processor -> {
+                if (getApplicableMigration(processor, migrationConfigs).isPresent()) {
+                  return migratePipelineElement(
+                      processor,
+                      migrationConfigs,
+                      String.format("%s/%s/processor",
+                          extensionsServiceConfig.getServiceUrl(),
+                          MIGRATION_ENDPOINT
+                      ),
+                      failedMigrations
+                  );
+                } else {
+                  LOG.info("No migration applicable for data processor '{}'.", processor.getElementId());
+                  return processor;
+                }
+              })
+              .toList();
+          pipeline.setSepas(migratedDataProcessors);
+
+          var migratedDataSinks = pipeline.getActions()
+              .stream()
+              .map(sink -> {
+                if (getApplicableMigration(sink, migrationConfigs).isPresent()) {
+                  return migratePipelineElement(
+                      sink,
+                      migrationConfigs,
+                      String.format("%s/%s/sink",
+                          extensionsServiceConfig.getServiceUrl(),
+                          MIGRATION_ENDPOINT
+                      ),
+                      failedMigrations
+                  );
+                } else {
+                  LOG.info("No migration applicable for data sink '{}'.", sink.getElementId());
+                  return sink;
+                }
+              })
+              .toList();
+          pipeline.setActions(migratedDataSinks);
+
+          pipelineStorage.updatePipeline(pipeline);
+
+          if (failedMigrations.isEmpty()) {
+            LOG.info("Migration for pipeline successfully completed.");
+          } else {
+            // pass most recent version of pipeline
+            handleFailedMigrations(pipelineStorage.getPipeline(pipeline.getPipelineId()), failedMigrations);
+          }
+        }
       }
+    } else {
+      LOG.info("No pipeline element migrations to perform");
     }
   }
 
+  private boolean shouldMigratePipeline(Pipeline pipeline,
+                                        List<ModelMigratorConfig> migrationConfigs) {
+    return Stream
+        .concat(pipeline.getSepas().stream(), pipeline.getActions().stream())
+        .anyMatch(element -> getApplicableMigration(element, migrationConfigs).isPresent());
+  }
+
   /**
    * Takes care about the failed migrations of pipeline elements.
    * This includes the following steps:
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/SpCoreConfigurationStep.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/SpCoreConfigurationStep.java
index 187d677efc..b5e0a8bd3b 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/SpCoreConfigurationStep.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/SpCoreConfigurationStep.java
@@ -19,14 +19,22 @@
 package org.apache.streampipes.manager.setup;
 
 import org.apache.streampipes.model.configuration.DefaultSpCoreConfiguration;
+import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class SpCoreConfigurationStep extends InstallationStep {
+
+  private static final Logger LOG = LoggerFactory.getLogger(SpCoreConfigurationStep.class);
+
   @Override
   public void install() {
     var coreCfg = new DefaultSpCoreConfiguration().make();
-
+    coreCfg.setServiceStatus(SpCoreConfigurationStatus.INSTALLING);
     StorageDispatcher.INSTANCE.getNoSqlStore().getSpCoreConfigurationStorage().createElement(coreCfg);
+    LOG.info("Core is now in {} state", coreCfg.getServiceStatus());
     new StreamPipesEnvChecker().updateEnvironmentVariables();
   }
 
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/StreamPipesEnvChecker.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/StreamPipesEnvChecker.java
index c4fc1bba08..70e5a8aa59 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/StreamPipesEnvChecker.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/StreamPipesEnvChecker.java
@@ -52,7 +52,7 @@ public class StreamPipesEnvChecker {
         .getNoSqlStore()
         .getSpCoreConfigurationStorage();
 
-    if (configStorage.getAll().size() > 0) {
+    if (configStorage.exists()) {
       this.coreConfig = configStorage.get();
 
       LOG.info("Checking and updating environment variables...");
diff --git a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/migration/DataSinkMigrationResource.java b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/migration/DataSinkMigrationResource.java
index 98569ffae7..6d1844743a 100644
--- a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/migration/DataSinkMigrationResource.java
+++ b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/migration/DataSinkMigrationResource.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.rest.extensions.migration;
 
 import org.apache.streampipes.extensions.api.extractor.IDataSinkParameterExtractor;
-import org.apache.streampipes.extensions.api.migration.DataSinkMigrator;
+import org.apache.streampipes.extensions.api.migration.IDataSinkMigrator;
 import org.apache.streampipes.model.extensions.migration.MigrationRequest;
 import org.apache.streampipes.model.graph.DataSinkInvocation;
 import org.apache.streampipes.rest.security.AuthConstants;
@@ -42,7 +42,7 @@ import jakarta.ws.rs.core.Response;
 public class DataSinkMigrationResource extends MigrateExtensionsResource<
         DataSinkInvocation,
         IDataSinkParameterExtractor,
-        DataSinkMigrator
+    IDataSinkMigrator
         > {
   @POST
   @Consumes(MediaType.APPLICATION_JSON)
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java
index 13c87875fd..6b28afe0a3 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java
@@ -18,9 +18,14 @@
 
 package org.apache.streampipes.rest.impl.admin;
 
+import org.apache.streampipes.config.backend.BackendConfig;
 import org.apache.streampipes.connect.management.management.AdapterMigrationManager;
+import org.apache.streampipes.manager.health.CoreServiceStatusManager;
+import org.apache.streampipes.manager.health.ServiceRegistrationManager;
+import org.apache.streampipes.manager.migration.AdapterDescriptionMigration093;
 import org.apache.streampipes.manager.migration.PipelineElementMigrationManager;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
 import org.apache.streampipes.model.migration.ModelMigratorConfig;
 import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
@@ -36,6 +41,8 @@ import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.enums.ParameterIn;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import org.apache.http.HttpStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.stereotype.Component;
 
@@ -53,9 +60,12 @@ import java.util.List;
 @PreAuthorize(AuthConstants.IS_ADMIN_ROLE)
 public class MigrationResource extends AbstractAuthGuardedRestResource {
 
+  private static final Logger LOG = LoggerFactory.getLogger(MigrationResource.class);
+
   private final CRUDStorage<String, SpServiceRegistration> extensionsServiceStorage =
       getNoSqlStorage().getExtensionsServiceStorage();
 
+  private final IAdapterStorage adapterDescriptionStorage = getNoSqlStorage().getAdapterDescriptionStorage();
   private final IAdapterStorage adapterStorage = getNoSqlStorage().getAdapterInstanceStorage();
 
   private final IDataProcessorStorage dataProcessorStorage = getNoSqlStorage().getDataProcessorStorage();
@@ -63,6 +73,10 @@ public class MigrationResource extends AbstractAuthGuardedRestResource {
   private final IDataSinkStorage dataSinkStorage = getNoSqlStorage().getDataSinkStorage();
   private final IPipelineStorage pipelineStorage = getNoSqlStorage().getPipelineStorageAPI();
 
+  private final CoreServiceStatusManager coreServiceStatusManager = new CoreServiceStatusManager(
+      getNoSqlStorage().getSpCoreConfigurationStorage()
+  );
+
   @POST
   @Path("{serviceId}")
   @Consumes(MediaType.APPLICATION_JSON)
@@ -88,24 +102,42 @@ public class MigrationResource extends AbstractAuthGuardedRestResource {
       )
       List<ModelMigratorConfig> migrationConfigs) {
 
-    var extensionsServiceConfig = extensionsServiceStorage.getElementById(serviceId);
-    var adapterMigrations = filterConfigs(migrationConfigs, List.of(SpServiceTagPrefix.ADAPTER));
-    var pipelineElementMigrations = filterConfigs(
-        migrationConfigs,
-        List.of(SpServiceTagPrefix.DATA_PROCESSOR, SpServiceTagPrefix.DATA_SINK)
-    );
-
-    new AdapterMigrationManager(adapterStorage).handleMigrations(extensionsServiceConfig, adapterMigrations);
-    new PipelineElementMigrationManager(
-        pipelineStorage,
-        dataProcessorStorage,
-        dataSinkStorage)
-        .handleMigrations(extensionsServiceConfig, pipelineElementMigrations);
+    var serviceManager = new ServiceRegistrationManager(extensionsServiceStorage);
+    var extensionsServiceConfig = serviceManager.getService(serviceId);
+    if (BackendConfig.INSTANCE.isConfigured()) {
+      new AdapterDescriptionMigration093(adapterDescriptionStorage).reinstallAdapters(extensionsServiceConfig);
+      if (!migrationConfigs.isEmpty()) {
+        if (serviceManager.isAnyServiceMigrating() || !isCoreReady()) {
+          LOG.info("Refusing migration request since precondition is not met.");
+          return Response.status(HttpStatus.SC_CONFLICT).build();
+        } else {
+          serviceManager.applyServiceStatus(serviceId, SpServiceStatus.MIGRATING);
+          var adapterMigrations = filterConfigs(migrationConfigs, List.of(SpServiceTagPrefix.ADAPTER));
+          var pipelineElementMigrations = filterConfigs(
+              migrationConfigs,
+              List.of(SpServiceTagPrefix.DATA_PROCESSOR, SpServiceTagPrefix.DATA_SINK)
+          );
+
+          new AdapterMigrationManager(adapterStorage).handleMigrations(extensionsServiceConfig, adapterMigrations);
+          new PipelineElementMigrationManager(
+              pipelineStorage,
+              dataProcessorStorage,
+              dataSinkStorage)
+              .handleMigrations(extensionsServiceConfig, pipelineElementMigrations);
+        }
+      }
+    }
+    new ServiceRegistrationManager(extensionsServiceStorage)
+        .applyServiceStatus(extensionsServiceConfig.getSvcId(), SpServiceStatus.HEALTHY);
     return ok();
   }
 
+  private boolean isCoreReady() {
+    return coreServiceStatusManager.isCoreReady();
+  }
+
   private List<ModelMigratorConfig> filterConfigs(List<ModelMigratorConfig> migrationConfigs,
-                                                       List<SpServiceTagPrefix> modelTypes) {
+                                                  List<SpServiceTagPrefix> modelTypes) {
     return migrationConfigs
         .stream()
         .filter(config -> modelTypes.stream().anyMatch(modelType -> modelType == config.modelType()))
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
index f89cf34ace..d97653d866 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
@@ -18,11 +18,15 @@
 
 package org.apache.streampipes.rest.impl.admin;
 
+import org.apache.streampipes.manager.health.ServiceRegistrationManager;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
 import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
 import org.apache.streampipes.rest.security.AuthConstants;
 import org.apache.streampipes.storage.api.CRUDStorage;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.stereotype.Component;
 
@@ -40,6 +44,8 @@ import jakarta.ws.rs.core.Response;
 @PreAuthorize(AuthConstants.IS_ADMIN_ROLE)
 public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ServiceRegistrationResource.class);
+
   private final CRUDStorage<String, SpServiceRegistration> extensionsServiceStorage =
       getNoSqlStorage().getExtensionsServiceStorage();
 
@@ -52,7 +58,8 @@ public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource
   @POST
   @Consumes(MediaType.APPLICATION_JSON)
   public Response registerService(SpServiceRegistration serviceRegistration) {
-    extensionsServiceStorage.createElement(serviceRegistration);
+    new ServiceRegistrationManager(extensionsServiceStorage)
+        .addService(serviceRegistration, SpServiceStatus.REGISTERED);
     return ok();
   }
 
@@ -60,8 +67,7 @@ public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource
   @Path("/{serviceId}")
   public Response unregisterService(@PathParam("serviceId") String serviceId) {
     try {
-      var serviceRegistration = extensionsServiceStorage.getElementById(serviceId);
-      extensionsServiceStorage.deleteElement(serviceRegistration);
+      new ServiceRegistrationManager(extensionsServiceStorage).removeService(serviceId);
       return ok();
     } catch (IllegalArgumentException e) {
       return badRequest("Could not find registered service with id " + serviceId);
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
index 6707617e84..bd9c595cf4 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/StaticProperties.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.sdk;
 
 import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.CodeInputStaticProperty;
 import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
 import org.apache.streampipes.model.staticproperty.FileStaticProperty;
 import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
@@ -32,9 +33,11 @@ import org.apache.streampipes.model.staticproperty.RuntimeResolvableTreeInputSta
 import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
 import org.apache.streampipes.model.staticproperty.SelectionStaticProperty;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticPropertyAlternative;
 import org.apache.streampipes.model.staticproperty.StaticPropertyAlternatives;
 import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
 import org.apache.streampipes.model.staticproperty.SupportedProperty;
+import org.apache.streampipes.sdk.helpers.CodeLanguage;
 import org.apache.streampipes.sdk.helpers.Filetypes;
 import org.apache.streampipes.sdk.helpers.Label;
 import org.apache.streampipes.sdk.helpers.RequirementsSelector;
@@ -47,6 +50,23 @@ import java.util.List;
 
 public class StaticProperties {
 
+  public static StaticPropertyAlternatives alternatives(Label label, StaticPropertyAlternative... alternatives) {
+    return alternatives(label, Arrays.asList(alternatives));
+  }
+
+  public static StaticPropertyAlternatives alternatives(Label label, List<StaticPropertyAlternative> alternatives) {
+    StaticPropertyAlternatives alternativesContainer =
+        new StaticPropertyAlternatives(label.getInternalId(), label.getLabel(), label.getDescription());
+
+    for (int i = 0; i < alternatives.size(); i++) {
+      alternatives.get(i).setIndex(i);
+    }
+
+    alternativesContainer.setAlternatives(alternatives);
+
+    return alternativesContainer;
+  }
+
   public static MappingPropertyUnary mappingPropertyUnary(Label label, RequirementsSelector requirementsSelector,
                                                           PropertyScope propertyScope) {
     MappingPropertyUnary mp = new MappingPropertyUnary(label.getInternalId(), label
@@ -134,7 +154,6 @@ public class StaticProperties {
   }
 
 
-
   public static RuntimeResolvableTreeInputStaticProperty runtimeResolvableTreeInput(Label label,
                                                                                     List<String> dependsOn,
                                                                                     boolean resolveDynamically,
@@ -209,6 +228,16 @@ public class StaticProperties {
     }
   }
 
+  public static CodeInputStaticProperty codeStaticProperty(Label label,
+                                                           CodeLanguage codeLanguage,
+                                                           String defaultSkeleton) {
+    var codeInputStaticProperty = new CodeInputStaticProperty(label.getInternalId(),
+        label.getLabel(), label.getDescription());
+    codeInputStaticProperty.setLanguage(codeLanguage.name());
+    codeInputStaticProperty.setCodeTemplate(defaultSkeleton);
+    return codeInputStaticProperty;
+  }
+
   private static StaticProperty setHorizontalRendering(StaticProperty sp) {
     if (sp instanceof StaticPropertyGroup) {
       ((StaticPropertyGroup) sp).setHorizontalRendering(true);
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
index 64d7a05ae4..f0979282f5 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractConfigurablePipelineElementBuilder.java
@@ -977,13 +977,7 @@ public abstract class AbstractConfigurablePipelineElementBuilder<K extends
 
   public K requiredAlternatives(Label label, StaticPropertyAlternative... alternatives) {
     StaticPropertyAlternatives alternativesContainer =
-        new StaticPropertyAlternatives(label.getInternalId(), label.getLabel(), label.getDescription());
-
-    for (int i = 0; i < alternatives.length; i++) {
-      alternatives[i].setIndex(i);
-    }
-
-    alternativesContainer.setAlternatives(Arrays.asList(alternatives));
+        StaticProperties.alternatives(label, alternatives);
     this.staticProperties.add(alternativesContainer);
 
     return me();
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
index 9355c6b895..2442816746 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
@@ -44,13 +44,13 @@ public class PostStartupTask implements Runnable {
   private static final int MAX_PIPELINE_START_RETRIES = 3;
   private static final int WAIT_TIME_AFTER_FAILURE_IN_SECONDS = 10;
 
-  private final List<Pipeline> allPipelines;
+  private final IPipelineStorage pipelineStorage;
   private final Map<String, Integer> failedPipelines = new HashMap<>();
   private final ScheduledExecutorService executorService;
   private final WorkerAdministrationManagement workerAdministrationManagement;
 
-  public PostStartupTask(List<Pipeline> allPipelines) {
-    this.allPipelines = allPipelines;
+  public PostStartupTask(IPipelineStorage pipelineStorage) {
+    this.pipelineStorage = pipelineStorage;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
     this.workerAdministrationManagement = new WorkerAdministrationManagement();
   }
@@ -76,6 +76,7 @@ public class PostStartupTask implements Runnable {
   }
 
   private void startAllPreviouslyStoppedPipelines() {
+    var allPipelines = pipelineStorage.getAllPipelines();
     LOG.info("Checking for orphaned pipelines...");
     List<Pipeline> orphanedPipelines = allPipelines
         .stream()
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
index e03d3291d1..49b218f738 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
@@ -18,6 +18,8 @@
 package org.apache.streampipes.service.core;
 
 import org.apache.streampipes.config.backend.BackendConfig;
+import org.apache.streampipes.connect.management.health.AdapterHealthCheck;
+import org.apache.streampipes.manager.health.CoreServiceStatusManager;
 import org.apache.streampipes.manager.health.PipelineHealthCheck;
 import org.apache.streampipes.manager.health.ServiceHealthCheck;
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsServiceLogExecutor;
@@ -30,6 +32,7 @@ import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory;
 import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
+import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
 import org.apache.streampipes.rest.security.SpPermissionEvaluator;
@@ -37,6 +40,7 @@ import org.apache.streampipes.service.base.BaseNetworkingConfig;
 import org.apache.streampipes.service.base.StreamPipesServiceBase;
 import org.apache.streampipes.service.core.migrations.MigrationsHandler;
 import org.apache.streampipes.storage.api.IPipelineStorage;
+import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage;
 import org.apache.streampipes.storage.couchdb.utils.CouchDbViewGenerator;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
@@ -72,11 +76,13 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
   private static final int LOG_FETCH_INTERVAL = 60;
   private static final TimeUnit LOG_FETCH_UNIT = TimeUnit.SECONDS;
 
-  private static final int HEALTH_CHECK_INTERVAL = 60;
+  private static final int HEALTH_CHECK_INTERVAL = 30;
   private static final TimeUnit HEALTH_CHECK_UNIT = TimeUnit.SECONDS;
 
-  private static final int SERVICE_HEALTH_CHECK_INTERVAL = 60;
-  private static final TimeUnit SERVICE_HEALTH_CHECK_UNIT = TimeUnit.SECONDS;
+  private final ISpCoreConfigurationStorage coreConfigStorage = StorageDispatcher.INSTANCE
+      .getNoSqlStore().getSpCoreConfigurationStorage();
+
+  private final CoreServiceStatusManager coreStatusManager = new CoreServiceStatusManager(coreConfigStorage);
 
   public static void main(String[] args) {
     StreamPipesCoreApplication application = new StreamPipesCoreApplication();
@@ -109,9 +115,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
   @PostConstruct
   public void init() {
     var executorService = Executors.newSingleThreadScheduledExecutor();
-    var healthCheckExecutorService = Executors.newSingleThreadScheduledExecutor();
     var logCheckExecutorService = Executors.newSingleThreadScheduledExecutor();
-    var serviceHealthCheckExecutorService = Executors.newSingleThreadScheduledExecutor();
 
     new StreamPipesEnvChecker().updateEnvironmentVariables();
     new CouchDbViewGenerator().createGenericDatabaseIfNotExists();
@@ -119,22 +123,20 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
     if (!isConfigured()) {
       doInitialSetup();
     } else {
+      // Check needs to be present since core configuration is part of migration
+      if (coreConfigStorage.exists()) {
+        coreStatusManager.updateCoreStatus(SpCoreConfigurationStatus.MIGRATING);
+      }
       new MigrationsHandler().performMigrations();
     }
+    coreStatusManager.updateCoreStatus(SpCoreConfigurationStatus.READY);
 
-    executorService.schedule(new PostStartupTask(getAllPipelines()), 10, TimeUnit.SECONDS);
+    executorService.schedule(new PostStartupTask(getPipelineStorage()), 10, TimeUnit.SECONDS);
 
-    LOG.info("Service health check will run every {} seconds", SERVICE_HEALTH_CHECK_INTERVAL);
-    serviceHealthCheckExecutorService.scheduleAtFixedRate(new ServiceHealthCheck(),
-        SERVICE_HEALTH_CHECK_INTERVAL,
-        SERVICE_HEALTH_CHECK_INTERVAL,
-        SERVICE_HEALTH_CHECK_UNIT);
-
-    LOG.info("Pipeline health check will run every {} seconds", HEALTH_CHECK_INTERVAL);
-    healthCheckExecutorService.scheduleAtFixedRate(new PipelineHealthCheck(),
-        HEALTH_CHECK_INTERVAL,
-        HEALTH_CHECK_INTERVAL,
-        HEALTH_CHECK_UNIT);
+    scheduleHealthChecks(List.of(
+        new ServiceHealthCheck(),
+        new PipelineHealthCheck(),
+        new AdapterHealthCheck()));
 
     LOG.info("Extensions logs will be fetched every {} seconds", LOG_FETCH_INTERVAL);
     logCheckExecutorService.scheduleAtFixedRate(new ExtensionsServiceLogExecutor(),
@@ -143,6 +145,21 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
         LOG_FETCH_UNIT);
   }
 
+  private void scheduleHealthChecks(List<Runnable> checks) {
+    var healthCheckExecutorService = Executors.newSingleThreadScheduledExecutor();
+    checks.forEach(check -> {
+      LOG.info(
+          "Health check {} configured to run every {} {}",
+          check.getClass().getCanonicalName(),
+          HEALTH_CHECK_INTERVAL,
+          HEALTH_CHECK_UNIT);
+      healthCheckExecutorService.scheduleAtFixedRate(check,
+          HEALTH_CHECK_INTERVAL,
+          HEALTH_CHECK_INTERVAL,
+          HEALTH_CHECK_UNIT);
+    });
+  }
+
   private boolean isConfigured() {
     return BackendConfig.INSTANCE.isConfigured();
   }
@@ -163,8 +180,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
     }
   }
 
-
-
   @PreDestroy
   public void onExit() {
     LOG.info("Shutting down StreamPipes...");
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/AdapterMigration.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/AdapterMigration.java
index 94d2d7eecd..29fd205ab3 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/AdapterMigration.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/AdapterMigration.java
@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.service.core.migrations.v093;
 
+import org.apache.streampipes.manager.migration.AdapterDescriptionMigration093Provider;
 import org.apache.streampipes.model.connect.adapter.migration.MigrationHelpers;
 import org.apache.streampipes.model.connect.adapter.migration.utils.AdapterModels;
 import org.apache.streampipes.service.core.migrations.Migration;
@@ -45,7 +46,7 @@ public class AdapterMigration implements Migration {
   private final CouchDbClient adapterInstanceClient;
   private final CouchDbClient adapterDescriptionClient;
   private final List<JsonObject> adaptersToMigrate;
-  private final List<JsonObject> adapterDescriptionsToMigrate;
+  private final List<JsonObject> adapterDescriptionsToRemove;
 
   private final MigrationHelpers helpers;
 
@@ -54,7 +55,7 @@ public class AdapterMigration implements Migration {
     this.adapterInstanceClient = Utils.getCouchDbAdapterInstanceClient();
     this.adapterDescriptionClient = Utils.getCouchDbAdapterDescriptionClient();
     this.adaptersToMigrate = new ArrayList<>();
-    this.adapterDescriptionsToMigrate = new ArrayList<>();
+    this.adapterDescriptionsToRemove = new ArrayList<>();
     this.helpers = new MigrationHelpers();
   }
 
@@ -64,9 +65,9 @@ public class AdapterMigration implements Migration {
     var adapterDescriptionUri = getAllDocsUri(adapterDescriptionClient);
 
     findDocsToMigrate(adapterInstanceClient, adapterInstanceUri, adaptersToMigrate);
-    findDocsToMigrate(adapterDescriptionClient, adapterDescriptionUri, adapterDescriptionsToMigrate);
+    findDocsToMigrate(adapterDescriptionClient, adapterDescriptionUri, adapterDescriptionsToRemove);
 
-    return !adaptersToMigrate.isEmpty() || !adapterDescriptionsToMigrate.isEmpty();
+    return !adaptersToMigrate.isEmpty() || !adapterDescriptionsToRemove.isEmpty();
   }
 
   private void findDocsToMigrate(CouchDbClient adapterClient,
@@ -89,15 +90,19 @@ public class AdapterMigration implements Migration {
   public void executeMigration() {
     var adapterInstanceBackupClient = Utils.getCouchDbAdapterInstanceBackupClient();
 
-    adapterDescriptionsToMigrate.forEach(ad -> {
+    LOG.info("Deleting {} adapter descriptions, which will be regenerated after migration",
+        adapterDescriptionsToRemove.size());
+
+    adapterDescriptionsToRemove.forEach(ad -> {
+      String docId = helpers.getDocId(ad);
       var adapterType = ad.get("type").getAsString();
-      var appId = ad.get("appId");
-      if (isSetAdapter(adapterType)) {
-        LOG.info("Deleting adapter description data set {}", appId);
-        adapterDescriptionClient.remove(helpers.getDocId(ad), helpers.getRev(ad));
-      } else {
-        LOG.info("Migrating adapter description {} to new adapter model", appId);
-        getAdapterMigrator(adapterType).migrate(adapterDescriptionClient, ad);
+      String rev = helpers.getRev(ad);
+      String appId = helpers.getAppId(ad);
+      if (!isSetAdapter(adapterType)) {
+        AdapterDescriptionMigration093Provider.INSTANCE.addAppId(appId);
+      }
+      if (docId != null && rev != null) {
+        adapterDescriptionClient.remove(docId, rev);
       }
     });
 
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
index 5da0d0bef7..53857b12ca 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
@@ -22,6 +22,7 @@ import org.apache.streampipes.config.backend.BackendConfig;
 import org.apache.streampipes.config.backend.BackendConfigKeys;
 import org.apache.streampipes.model.configuration.DefaultMessagingSettings;
 import org.apache.streampipes.model.configuration.SpCoreConfiguration;
+import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
 import org.apache.streampipes.service.core.migrations.Migration;
 import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
@@ -71,6 +72,7 @@ public class ConsulConfigMigration implements Migration {
     newConf.setFilesDir(currConf.getFilesDir());
 
     newConf.setMessagingSettings(messagingSettings);
+    newConf.setServiceStatus(SpCoreConfigurationStatus.MIGRATING);
     storage.createElement(newConf);
   }
 
diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
index 8170c36ad9..0455dc1d8d 100644
--- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
+++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.svcdiscovery;
 
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
 import org.apache.streampipes.storage.api.CRUDStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
@@ -62,7 +63,7 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery {
         .stream()
         .filter(service -> allFiltersSupported(service, filterByTags))
         .filter(service -> !restrictToHealthy
-            || service.isHealthy())
+            || service.getStatus() != SpServiceStatus.UNHEALTHY)
         .map(this::makeServiceUrl)
         .collect(Collectors.toList());
   }
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CoreRequestSubmitter.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CoreRequestSubmitter.java
new file mode 100644
index 0000000000..ab7f631d02
--- /dev/null
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CoreRequestSubmitter.java
@@ -0,0 +1,54 @@
+/*
+ * 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.service.extensions;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+public class CoreRequestSubmitter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CoreRequestSubmitter.class);
+
+  private static final int RETRY_INTERVAL_SECONDS = 3;
+
+  public void submitRepeatedRequest(Supplier<Boolean> request,
+                                    String successMessage,
+                                    String failureMessage) {
+    try {
+      request.get();
+      LOG.info(successMessage);
+    } catch (SpRuntimeException e) {
+      LOG.warn(
+          failureMessage + " Trying again in {} seconds",
+          RETRY_INTERVAL_SECONDS
+      );
+      try {
+        TimeUnit.SECONDS.sleep(RETRY_INTERVAL_SECONDS);
+        submitRepeatedRequest(request, successMessage, failureMessage);
+      } catch (InterruptedException ex) {
+        throw new RuntimeException(ex);
+      }
+    }
+  }
+}
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
index cbdb3992f9..24f128cc47 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
@@ -52,7 +52,13 @@ public abstract class ExtensionsModelSubmitter extends StreamPipesExtensionsServ
 
     // register all migrations at StreamPipes Core
     var migrationConfigs = serviceDef.getMigrators().stream().map(IModelMigrator::config).toList();
-    client.adminApi().registerMigrations(migrationConfigs, serviceId());
+    new CoreRequestSubmitter().submitRepeatedRequest(
+        () -> {
+          client.adminApi().registerMigrations(migrationConfigs, serviceId());
+          return true;
+        },
+        "Successfully sent migration request",
+        "Core currently doesn't accept migration requests.");
 
     // initialize all function instances
     StreamPipesFunctionHandler.INSTANCE.initializeFunctions(serviceDef.getServiceGroup());
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
index c13b1153a8..0913fd2587 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.service.extensions;
 
 import org.apache.streampipes.client.StreamPipesClient;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver;
 import org.apache.streampipes.extensions.management.init.DeclarersSingleton;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
@@ -40,7 +39,6 @@ import jakarta.annotation.PreDestroy;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 
 public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServiceBase {
 
@@ -96,23 +94,17 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic
   }
 
   private void registerService(SpServiceRegistration serviceRegistration) {
-    StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance();
-    try {
-      client.adminApi().registerService(serviceRegistration);
-      LOG.info("Successfully registered service at core.");
-    } catch (SpRuntimeException e) {
-      LOG.warn(
-          "Could not register at core at url {}. Trying again in {} seconds",
-          client.getConnectionConfig().getBaseUrl(),
-          RETRY_INTERVAL_SECONDS
-      );
-      try {
-        TimeUnit.SECONDS.sleep(RETRY_INTERVAL_SECONDS);
-        registerService(serviceRegistration);
-      } catch (InterruptedException ex) {
-        throw new RuntimeException(ex);
-      }
-    }
+    var client = new StreamPipesClientResolver().makeStreamPipesClientInstance();
+    new CoreRequestSubmitter().submitRepeatedRequest(
+        () -> {
+          client.adminApi().registerService(serviceRegistration);
+          return true;
+        },
+        "Successfully registered service at core.",
+        String.format(
+            "Could not register service at core at url %s",
+            client.getConnectionConfig().getBaseUrl()
+        ));
   }
 
   protected List<SpServiceTag> getServiceTags() {
diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java
index 829dc09ce9..2d0f8ae657 100644
--- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java
+++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java
@@ -24,6 +24,8 @@ import java.util.List;
 
 public interface ISpCoreConfigurationStorage {
 
+  boolean exists();
+
   List<SpCoreConfiguration> getAll();
 
   void createElement(SpCoreConfiguration element);
diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java
index 7ff0a99c04..9b25b5ac80 100644
--- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java
+++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java
@@ -33,6 +33,11 @@ public class CoreConfigurationStorageImpl extends AbstractDao<SpCoreConfiguratio
     super(Utils::getCouchDbGeneralConfigStorage, SpCoreConfiguration.class);
   }
 
+  @Override
+  public boolean exists() {
+    return !findAll().isEmpty();
+  }
+
   @Override
   public List<SpCoreConfiguration> getAll() {
     return findAll();
diff --git a/ui/cypress/support/builder/PipelineElementBuilder.ts b/ui/cypress/support/builder/PipelineElementBuilder.ts
index 9464317f26..fa56292ce0 100644
--- a/ui/cypress/support/builder/PipelineElementBuilder.ts
+++ b/ui/cypress/support/builder/PipelineElementBuilder.ts
@@ -18,6 +18,7 @@
 
 import { UserInput } from '../model/UserInput';
 import { PipelineElementInput } from '../model/PipelineElementInput';
+import { UserInputType } from '../model/UserInputType';
 
 export class PipelineElementBuilder {
     pipelineElementInput: PipelineElementInput;
@@ -32,7 +33,7 @@ export class PipelineElementBuilder {
         return new PipelineElementBuilder(name);
     }
 
-    public addInput(type: string, selector: string, value: string) {
+    public addInput(type: UserInputType, selector: string, value: string) {
         const userInput = new UserInput();
         userInput.type = type;
         userInput.selector = selector;
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 6407e14bed..4a55c98883 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -121,7 +121,7 @@
         "resize-observer-polyfill": "^1.5.1",
         "to-string-loader": "^1.2.0",
         "typescript": "4.8.2",
-        "webpack": "^5.88.2",
+        "webpack": "^5.89.0",
         "webpack-cli": "^5.1.4",
         "webpack-merge": "^5.8.0"
       }
@@ -27184,9 +27184,9 @@
       "integrity": "sha512-2JAn3z8AR6rjK8Sm8orRC0h/bcl/DqL7tRPdGZ4I1CjdF+EaMLmYxBHyXuKL849eucPFhvBoxMsflfOb8kxaeQ=="
     },
     "node_modules/webpack": {
-      "version": "5.88.2",
-      "resolved": "https://registry.npmjs.org/webpack/-/webpack-5.88.2.tgz",
-      "integrity": "sha512-JmcgNZ1iKj+aiR0OvTYtWQqJwq37Pf683dY9bVORwVbUrDhLhdn/PlO2sHsFHPkj7sHNQF3JwaAkp49V+Sq1tQ==",
+      "version": "5.89.0",
+      "resolved": "https://registry.npmjs.org/webpack/-/webpack-5.89.0.tgz",
+      "integrity": "sha512-qyfIC10pOr70V+jkmud8tMfajraGCZMBWJtrmuBymQKCrLTRejBI8STDp1MCyZu/QTdZSeacCQYpYNQVOzX5kw==",
       "dev": true,
       "dependencies": {
         "@types/eslint-scope": "^3.7.3",
@@ -49233,9 +49233,9 @@
       "integrity": "sha512-2JAn3z8AR6rjK8Sm8orRC0h/bcl/DqL7tRPdGZ4I1CjdF+EaMLmYxBHyXuKL849eucPFhvBoxMsflfOb8kxaeQ=="
     },
     "webpack": {
-      "version": "5.88.2",
-      "resolved": "https://registry.npmjs.org/webpack/-/webpack-5.88.2.tgz",
-      "integrity": "sha512-JmcgNZ1iKj+aiR0OvTYtWQqJwq37Pf683dY9bVORwVbUrDhLhdn/PlO2sHsFHPkj7sHNQF3JwaAkp49V+Sq1tQ==",
+      "version": "5.89.0",
+      "resolved": "https://registry.npmjs.org/webpack/-/webpack-5.89.0.tgz",
+      "integrity": "sha512-qyfIC10pOr70V+jkmud8tMfajraGCZMBWJtrmuBymQKCrLTRejBI8STDp1MCyZu/QTdZSeacCQYpYNQVOzX5kw==",
       "dev": true,
       "requires": {
         "@types/eslint-scope": "^3.7.3",
diff --git a/ui/package.json b/ui/package.json
index b2afd10dfa..9674465075 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -143,7 +143,7 @@
     "resize-observer-polyfill": "^1.5.1",
     "to-string-loader": "^1.2.0",
     "typescript": "4.8.2",
-    "webpack": "^5.88.2",
+    "webpack": "^5.89.0",
     "webpack-cli": "^5.1.4",
     "webpack-merge": "^5.8.0"
   },
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index 47f06b81f2..85f9a230b6 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -16,10 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 /* tslint:disable */
 /* eslint-disable */
 // @ts-nocheck
-// Generated using typescript-generator version 3.2.1263 on 2023-10-27 10:43:45.
+// Generated using typescript-generator version 3.2.1263 on 2023-10-30 22:49:29.
 
 export class NamedStreamPipesEntity {
     '@class':
@@ -3599,12 +3600,12 @@ export class SpServiceConfiguration {
 export class SpServiceRegistration {
     firstTimeSeenUnhealthy: number;
     healthCheckPath: string;
-    healthy: boolean;
     host: string;
     port: number;
     rev: string;
     scheme: string;
     serviceUrl: string;
+    status: SpServiceStatus;
     svcGroup: string;
     svcId: string;
     svcType: string;
@@ -3620,12 +3621,12 @@ export class SpServiceRegistration {
         const instance = target || new SpServiceRegistration();
         instance.firstTimeSeenUnhealthy = data.firstTimeSeenUnhealthy;
         instance.healthCheckPath = data.healthCheckPath;
-        instance.healthy = data.healthy;
         instance.host = data.host;
         instance.port = data.port;
         instance.rev = data.rev;
         instance.scheme = data.scheme;
         instance.serviceUrl = data.serviceUrl;
+        instance.status = data.status;
         instance.svcGroup = data.svcGroup;
         instance.svcId = data.svcId;
         instance.svcType = data.svcType;
@@ -4105,6 +4106,12 @@ export type SpProtocol = 'KAFKA' | 'JMS' | 'MQTT' | 'NATS' | 'PULSAR';
 
 export type SpQueryStatus = 'OK' | 'TOO_MUCH_DATA';
 
+export type SpServiceStatus =
+    | 'REGISTERED'
+    | 'MIGRATING'
+    | 'HEALTHY'
+    | 'UNHEALTHY';
+
 export type SpServiceTagPrefix =
     | 'SYSTEM'
     | 'SP_GROUP'
diff --git a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html
index 32f1534086..c0a1e74b5e 100644
--- a/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html
+++ b/ui/src/app/configuration/extensions-service-management/registered-extensions-services/registered-extensions-services.component.html
@@ -38,11 +38,14 @@
                 mat-cell
                 *matCellDef="let element"
             >
-                <span *ngIf="element.healthy" fxLayoutAlign="center center">
+                <span
+                    *ngIf="element.status === 'HEALTHY'"
+                    fxLayoutAlign="center center"
+                >
                     <mat-icon class="service-icon-passing">lens</mat-icon>
                 </span>
                 <span
-                    *ngIf="element.healthy === false"
+                    *ngIf="element.status !== 'HEALTHY'"
                     fxLayoutAlign="center center"
                 >
                     <mat-icon class="service-icon-critical">lens</mat-icon>
diff --git a/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html b/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
index 85da7bdb47..eee4c83479 100644
--- a/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
+++ b/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
@@ -33,7 +33,11 @@
                 *ngIf="!updateInfo.canAutoMigrate"
                 fxLayoutAlign="start center"
             >
-                <i class="material-icons">warning</i>
+                <i
+                    class="material-icons"
+                    data-cy="sp-connect-adapter-edit-warning"
+                    >warning</i
+                >
                 &nbsp;Pipeline&nbsp;<b>{{ updateInfo.pipelineName }}</b
                 >&nbsp;will be stopped and needs manual review.
             </div>
@@ -50,6 +54,7 @@
             mat-raised-button
             color="accent"
             (click)="startUpdateEmitter.emit()"
+            data-cy="btn-update-adapter-migrate-pipelines"
         >
             Update adapter and migrate pipelines
         </button>
diff --git a/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html b/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
index 17ea0eb25f..a879e6b374 100644
--- a/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
+++ b/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
@@ -35,6 +35,7 @@
                 matTooltip="Configure Element"
                 [matTooltipPosition]="'above'"
                 (click)="customizeElement(pipelineElement)"
+                data-cy="settings-pipeline-element-button"
             >
                 <i class="material-icons options-icon-size">settings</i>
             </button>
diff --git a/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html b/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
index a428276dd5..6e3805e1bb 100644
--- a/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
+++ b/ui/src/app/pipelines/components/pipeline-overview/pipeline-overview.component.html
@@ -82,13 +82,17 @@
                                         ? 's'
                                         : ''
                                 }}"
+                                data-cy="pipeline-warning-icon"
                             >
                                 <i class="material-icons pipeline-notification"
                                     >warning</i
                                 >
                             </button>
                         </div>
-                        <div class="ml-10 notification-icon">
+                        <div
+                            class="ml-10 notification-icon"
+                            data-cy="pipeline-sync-problem-icon"
+                        >
                             <mat-icon
                                 color="warn"
                                 matTooltip="Invalid pipeline: Modify to fix problems"
@@ -108,6 +112,7 @@
                     mat-icon-button
                     matTooltip="Start pipeline"
                     matTooltipPosition="above"
+                    data-cy="start-pipeline-button"
                     (click)="
                         pipelineOperationsService.startPipeline(
                             pipeline._id,
@@ -205,6 +210,7 @@
                                     pipeline._id
                                 )
                             "
+                            data-cy="modify-pipeline-btn"
                         >
                             <i class="material-icons">mode_edit</i>
                         </button>


Test patch:
diff --git a/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParserTest.java b/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParserTest.java
new file mode 100644
index 0000000000..20a31073e7
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-plc/src/test/java/org/apache/streampipes/extensions/connectors/plc/adapter/s7/config/ConfigurationParserTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.connectors.plc.adapter.s7.config;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.sdk.utils.Datatypes;
+
+import org.junit.Test;
+
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ConfigurationParserTest {
+
+  @Test
+  public void testGetNodeInformationFromCodePropertyWithComments() {
+    var configBlock = """
+        // This code block can be used to manually specify the addresses of the PLC registers.
+        // The syntax is based on the PLC4X syntax, see [1].
+        // Address Pattern:
+        // propertyName=%{Memory-Area}{start-address}:{Data-Type}[{array-size}]
+        //
+        // Sample:
+        temperature=%I0.0:INT
+
+
+        // [1] https://plc4x.apache.org/users/protocols/s7.html
+        """;
+    var result = new ConfigurationParser().getNodeInformationFromCodeProperty(configBlock);
+
+    assertEquals(1, result.size());
+    assertEquals(Set.of("temperature"), result.keySet());
+    assertEquals("%I0.0:INT", result.get("temperature"));
+  }
+
+  @Test
+  public void testGetNodeInformationFromCodePropertyMultipleEntries() {
+    var configBlock = """
+        v1=%I0.0:INT
+        v2=%I0.0:BOOL
+        """;
+    var result = new ConfigurationParser().getNodeInformationFromCodeProperty(configBlock);
+
+    assertEquals(2, result.size());
+    assertEquals(Set.of("v1", "v2"), result.keySet());
+    assertEquals("%I0.0:INT", result.get("v1"));
+    assertEquals("%I0.0:BOOL", result.get("v2"));
+  }
+
+  @Test
+  public void testGetStreamPipesDataTypeArray() throws AdapterException {
+    var plcType = "INT[100]";
+    var result = new ConfigurationParser().getStreamPipesDataType(plcType);
+
+    assertEquals(Datatypes.Integer, result);
+  }
+
+  @Test
+  public void testGetStreamPipesDataTypeBasic() throws AdapterException {
+    var plcType = "INT";
+    var result = new ConfigurationParser().getStreamPipesDataType(plcType);
+
+    assertEquals(Datatypes.Integer, result);
+  }
+
+
+  @Test
+  public void testGetNodeInformationFromCodePropertyNoEntries() {
+    var configBlock = "";
+    var result = new ConfigurationParser().getNodeInformationFromCodeProperty(configBlock);
+
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testIsPLCArray() {
+    var result = new ConfigurationParser().isPLCArray("%DB3.DB0:BOOL[100]");
+    assertTrue(result);
+  }
+
+
+  @Test
+  public void testIsNoPLCArray() {
+    var result = new ConfigurationParser().isPLCArray("%DB3.DB0:BOOL");
+    assertFalse(result);
+  }
+}
diff --git a/streampipes-sdk/src/test/java/org/apache/streampipes/sdk/StaticPropertiesTest.java b/streampipes-sdk/src/test/java/org/apache/streampipes/sdk/StaticPropertiesTest.java
new file mode 100644
index 0000000000..532ed187ee
--- /dev/null
+++ b/streampipes-sdk/src/test/java/org/apache/streampipes/sdk/StaticPropertiesTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.sdk;
+
+import org.apache.streampipes.sdk.helpers.CodeLanguage;
+import org.apache.streampipes.sdk.helpers.Label;
+import org.apache.streampipes.sdk.helpers.Labels;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class StaticPropertiesTest {
+
+  private static final String TEST_PROPERTY_LABEL = "test-property-id";
+  private static final Label defaultLabel = Labels.from("", TEST_PROPERTY_LABEL, "");
+
+  @Test
+  public void codeStaticProperty() {
+    String codeTemplate = "// This is a test";
+    var result = StaticProperties.codeStaticProperty(defaultLabel, CodeLanguage.Javascript, codeTemplate);
+
+    assertEquals(TEST_PROPERTY_LABEL, result.getLabel());
+    assertEquals(CodeLanguage.Javascript.name(), result.getLanguage());
+    assertEquals(codeTemplate, result.getCodeTemplate());
+  }
+}
\ No newline at end of file
diff --git a/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts b/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
index 28e4fe04a2..ba8a1f54e0 100644
--- a/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
+++ b/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
@@ -42,7 +42,6 @@ describe('Test Edit Adapter', () => {
         const newUserConfiguration = AdapterBuilder.create(
             'Machine_Data_Simulator',
         )
-            // .setName(name)
             .addInput('input', 'wait-time-ms', '2000')
             .addInput('radio', 'selected', 'simulator-option-pressure')
             .build();
diff --git a/ui/cypress/tests/adapter/editAdapterSettingsAndPipeline.spec.ts b/ui/cypress/tests/adapter/editAdapterSettingsAndPipeline.spec.ts
new file mode 100644
index 0000000000..ca0eb4f88d
--- /dev/null
+++ b/ui/cypress/tests/adapter/editAdapterSettingsAndPipeline.spec.ts
@@ -0,0 +1,113 @@
+/*
+ *  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.
+ *
+ */
+
+import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
+import { ConnectBtns } from '../../support/utils/connect/ConnectBtns';
+import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
+import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
+import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
+import { DashboardUtils } from '../../support/utils/DashboardUtils';
+
+describe('Test Edit Adapter and Pipeline', () => {
+    beforeEach('Setup Test', () => {
+        // To set up test add a stream adapter that can be configured
+        cy.initStreamPipesTest();
+        ConnectUtils.addMachineDataSimulator('simulator');
+
+        const adapterName = 'simulator';
+        const pipelineInput = PipelineBuilder.create('Pipeline Test')
+            .addSource(adapterName)
+            .addProcessingElement(
+                PipelineElementBuilder.create('numerical_filter')
+                    .addInput('input', 'value', '10')
+                    .build(),
+            )
+            .addSink(
+                PipelineElementBuilder.create('data_lake')
+                    .addInput('input', 'db_measurement', 'demo')
+                    .build(),
+            )
+            .build();
+
+        PipelineUtils.addPipeline(pipelineInput);
+    });
+
+    it('Edit adapter and test Pipeline behaviour', () => {
+        // Edit Apater and select pressure
+        ConnectUtils.goToConnect();
+        ConnectBtns.editAdapter().should('not.be.disabled');
+        ConnectBtns.editAdapter().click();
+        const newUserConfiguration = AdapterBuilder.create(
+            'Machine_Data_Simulator',
+        )
+            .addInput('input', 'wait-time-ms', '2000')
+            .addInput('radio', 'selected', 'simulator-option-pressure')
+            .build();
+
+        ConnectUtils.configureAdapter(newUserConfiguration);
+
+        // Update event schema
+        ConnectUtils.finishEventSchemaConfiguration();
+        ConnectBtns.storeEditAdapter().click();
+
+        // Check for warning message
+        cy.dataCy('sp-connect-adapter-edit-warning', {
+            timeout: 60000,
+        }).should('be.visible');
+        cy.dataCy('btn-update-adapter-migrate-pipelines').click();
+        ConnectUtils.closeAdapterPreview();
+
+        // Go to pipelines, check for warning icon and edit pipeline
+        cy.visit('#/pipelines').wait(1000);
+
+        cy.dataCy('pipeline-warning-icon', {
+            timeout: 60000,
+        }).should('be.visible');
+
+        cy.dataCy('pipeline-sync-problem-icon', {
+            timeout: 60000,
+        }).should('be.visible');
+
+        cy.dataCy('modify-pipeline-btn').click();
+        cy.dataCy('settings-pipeline-element-button').eq(0).click();
+        cy.dataCy('number-mapping').contains('pressure').click({ force: true });
+        cy.dataCy('sp-element-configuration-save').click({ force: true });
+        cy.dataCy('sp-editor-save-pipeline').click();
+        cy.dataCy('sp-editor-save').click();
+        cy.dataCy('start-pipeline-button').click();
+        cy.dataCy('sp-pipeline-dialog-close').click();
+
+        // Visit dashboard
+        cy.wait(1000);
+        DashboardUtils.goToDashboard();
+
+        // Add new dashboard widget and check if edited adapter appears
+        const dashboardName = 'testDashboard';
+        DashboardUtils.addAndEditDashboard(dashboardName);
+        cy.dataCy('dashboard-add-widget').click();
+        cy.dataCy('dashboard-visualize-pipeline-Pipeline_Test').click();
+        cy.dataCy('dashboard-select-widget-table').click();
+        cy.get('[type="checkbox"]').check();
+        cy.dataCy('dashboard-new-widget-next-btn').click();
+        cy.dataCy('dashboard-save-edit-mode').click();
+        cy.get('.main-panel').should('include.text', 'pressure');
+        cy.get('.main-panel').should('include.text', 'sensorId');
+        cy.get('.main-panel').should('include.text', 'time');
+    });
+});
diff --git a/ui/cypress/tests/adapter/editAdapterValuesAndFields.spec.ts b/ui/cypress/tests/adapter/editAdapterValuesAndFields.spec.ts
new file mode 100644
index 0000000000..60ce694b21
--- /dev/null
+++ b/ui/cypress/tests/adapter/editAdapterValuesAndFields.spec.ts
@@ -0,0 +1,133 @@
+/*
+ *  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.
+ *
+ */
+
+import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
+import { ConnectBtns } from '../../support/utils/connect/ConnectBtns';
+import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
+
+describe('Test Edit Adapter', () => {
+    beforeEach('Setup Test', () => {
+        cy.initStreamPipesTest();
+    });
+
+    it('Test configuration of adapter fields ', () => {
+        // Set up new adapter
+        const builder = AdapterBuilder.create('Machine_Data_Simulator')
+            .setName('Machine_Data_Simulator')
+            .addInput('input', 'wait-time-ms', '1000');
+        const configuration = builder.build();
+        ConnectUtils.goToConnect();
+        ConnectUtils.goToNewAdapterPage();
+        ConnectUtils.selectAdapter(configuration.adapterType);
+        cy.contains('Next').click();
+
+        // Add new property and edit field
+        cy.dataCy('connect-add-static-property').click();
+        cy.dataCy('connect-add-field-name').type('test-property-1');
+        cy.dataCy('connect-add-field-name-button').click();
+        cy.dataCy('edit-density').click();
+        // Change runtime name
+        cy.dataCy('connect-edit-field-runtime-name')
+            .clear()
+            .type('test-density');
+        // Change field semantic type
+        cy.get('[id="domainproperty"]')
+            .clear()
+            .type('http://schema.org/Numbers');
+        // Change field data type
+        cy.dataCy('connect-change-runtime-type')
+            .click()
+            .get('mat-option')
+            .contains('Double')
+            .click();
+        // Provide correction value and operator
+        cy.dataCy('connect-schema-correction-value').type('2');
+        cy.dataCy('connect-schema-correction-operator')
+            .click()
+            .get('mat-option')
+            .contains('Multiply')
+            .click();
+        cy.dataCy('sp-save-edit-property').click();
+        cy.dataCy('sp-event-schema-next-button').click();
+
+        // Fill in adapter name and close view
+        cy.dataCy('sp-adapter-name').type('Test Adapter');
+        cy.dataCy('adapter-settings-start-adapter-btn').click();
+        ConnectUtils.closeAdapterPreview();
+
+        // Edit adapter and check if given values and added property still provided
+        ConnectBtns.editAdapter().should('not.be.disabled');
+        ConnectBtns.editAdapter().click();
+        cy.contains('Next').click();
+        cy.dataCy('edit-density').click();
+        cy.dataCy('connect-edit-field-runtime-name').should(
+            'have.value',
+            'test-density',
+        );
+        cy.get('[id="domainproperty"]').should(
+            'have.value',
+            'http://schema.org/Numbers',
+        );
+        cy.dataCy('connect-change-runtime-type').should(
+            'include.text',
+            'Double',
+        );
+        cy.dataCy('connect-schema-correction-value').should('have.value', '2');
+        cy.dataCy('connect-schema-correction-operator').should(
+            'include.text',
+            'Multiply',
+        );
+
+        // Delete inserted values in edit field
+        cy.dataCy('connect-edit-field-runtime-name').clear();
+        cy.get('[id="domainproperty"]').clear();
+        cy.dataCy('connect-change-runtime-type')
+            .click()
+            .get('mat-option')
+            .contains('Float')
+            .click();
+        cy.dataCy('connect-schema-correction-value').clear();
+        cy.dataCy('sp-save-edit-property').click();
+        cy.get('[class="general-panel"]').should(
+            'include.text',
+            'test-property-1',
+        );
+        cy.dataCy('sp-event-schema-next-button').click();
+        cy.dataCy('store-edit-adapter').click();
+        ConnectUtils.closeAdapterPreview();
+
+        // Configure adapter with pressure instead of flowrate
+        ConnectBtns.editAdapter().click();
+        const newUserConfiguration = AdapterBuilder.create(
+            'Machine_Data_Simulator',
+        )
+            .addInput('input', 'wait-time-ms', '2000')
+            .addInput('radio', 'selected', 'simulator-option-pressure')
+            .build();
+        ConnectUtils.configureAdapter(newUserConfiguration);
+
+        // Check if given property still exists and close view afterwards
+        cy.get('[class="general-panel"]').should(
+            'include.text',
+            'test-property-1',
+        );
+        cy.dataCy('sp-event-schema-next-button').click();
+        cy.dataCy('store-edit-adapter').click();
+        ConnectUtils.closeAdapterPreview();
+    });
+});