Repository: apache/streampipes
Pull request #145 (2022-11-18 15:06:51)

[STREAMPIPES-637] Remove incubating references from source code
<!--
  ~ 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: `[STREAMPIPES-<Jira issue #>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][STREAMPIPES-<Jira issue #>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Be sure to keep the PR description updated to reflect all changes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. Make sure tests pass via `mvn clean install`.
8. (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.
-->

This PR removes all references to the incubator phase:
* DISCLAIMER removed
* pom files updated
* README files updated

### Approach
<!--
Describe how you are implementing the solutions along with the design details.
-->

### Samples
<!--
Provide high-level details about the samples related to this feature.
-->

### Remarks
<!--
List related issues/PRs, link to discussions in the mailing list, todo items, or any other notes related to the PR.
-->
Fixes: <Jira issue link>



Patch:
diff --git a/.github/workflows/pr-validation.yml b/.github/workflows/pr-validation.yml
index 484fb2dad0..b63242649b 100644
--- a/.github/workflows/pr-validation.yml
+++ b/.github/workflows/pr-validation.yml
@@ -36,6 +36,7 @@ jobs:
 
   run-cypress-e2e-tests:
     runs-on: ubuntu-latest
+    needs: [python-quality-checks, python-unit-tests]
     steps:
       - name: Checkout
         uses: actions/checkout@v1
@@ -77,3 +78,73 @@ jobs:
           wait-on-timeout: 120
           spec: cypress/tests/**/*.smoke.spec.ts
           working-directory: ui
+
+  python-quality-checks:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
+
+      - name: Cache Python dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip${{ hashFiles('**/setup.py') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install Python dependencies
+        run: |
+          cd ./streampipes-client-python
+          python -m pip install --upgrade pip
+          pip install wheel
+          pip install -e ".[dev]"
+
+      - name: Run style & code checks
+        run: |
+          cd ./streampipes-client-python
+          pre-commit run --files streampipes_client/*
+          pre-commit run --files tests/*
+          make mypy
+          make lint
+
+
+  python-unit-tests:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        python: [3.8, 3.9, "3.10"]
+        os: [ubuntu-latest, macos-latest]
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Python v${{ matrix.python }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python }}
+
+      - name: Cache Python dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~~.cache/pip
+          key: ${{ runner.os }}-${{ matrix.python }}-pip-${{ hashFiles('**/setup.py') }}
+          restore-keys: |
+            ${{ runner.os }}-${{ matrix.python }}-pip-
+
+      - name: Install Python dependencies
+        run: |
+          cd ./streampipes-client-python
+          python -m pip install --upgrade pip
+          pip install wheel
+          pip install -e ".[dev]"
+
+      - name: Run unit-tests
+        run: |
+          cd ./streampipes-client-python
+          make unit-tests
\ No newline at end of file
diff --git a/DISCLAIMER b/DISCLAIMER
deleted file mode 100644
index bb81ab3d29..0000000000
--- a/DISCLAIMER
+++ /dev/null
@@ -1,5 +0,0 @@
-Apache StreamPipes is an effort undergoing incubation at The Apache Software Foundation (ASF), sponsored by the
-Apache Incubator. Incubation is required of all newly accepted projects until a further review indicates that the
-infrastructure, communications, and decision making process have stabilized in a manner consistent with other successful
-ASF projects. While incubation status is not necessarily a reflection of the completeness or stability of the code, it
-does indicate that the project has yet to be fully endorsed by the ASF.
\ No newline at end of file
diff --git a/NOTICE b/NOTICE
index 026f925283..5011add9d5 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,4 +1,4 @@
-Apache StreamPipes (incubating)
+Apache StreamPipes
 Copyright 2022 The Apache Software Foundation.
 
 This product includes software developed at
diff --git a/README.md b/README.md
index 1cf80545fd..d5861da59d 100644
--- a/README.md
+++ b/README.md
@@ -61,7 +61,7 @@
 
 ## About Apache StreamPipes
 
-Apache StreamPipes (incubating) makes industrial data analytics easy!
+Apache StreamPipes makes industrial data analytics easy!
 
 StreamPipes is an end-to-end toolbox for the industrial IoT. 
 It comes with a rich graphical user interface targeted at non-technical users and provides the following features:
diff --git a/RELEASE_VALIDATION.md b/RELEASE_VALIDATION.md
index e4003ab131..7baf4c34de 100644
--- a/RELEASE_VALIDATION.md
+++ b/RELEASE_VALIDATION.md
@@ -18,9 +18,9 @@
 
 ## Release Validation
 
-An Apache StreamPipes (incubating) release consists of a single source release starting from version 0.69.0:
+An Apache StreamPipes release consists of a single source release starting from version 0.69.0:
 
-* incubator-streampipes
+* streampipes
 
 To validate and test a release candidate, please visit our validation guide: 
 [https://cwiki.apache.org/confluence/display/STREAMPIPES/Validating+a+release](https://cwiki.apache.org/confluence/display/STREAMPIPES/Validating+a+release)
diff --git a/installer/README.md b/installer/README.md
index ee51e2ab2b..47d8eaf39f 100644
--- a/installer/README.md
+++ b/installer/README.md
@@ -16,7 +16,7 @@
   ~
   -->
 
-# Apache StreamPipes (incubating) installer
+# Apache StreamPipes installer
 This project contains a variety of StreamPipes installation and operation options, namely:
 
 * **[StreamPipes Compose](./compose)** - The User's Choice
diff --git a/pom.xml b/pom.xml
index 3a23ff6dd7..fbd500d57b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -209,7 +209,7 @@
 	<protobuf.version>3.16.3</protobuf.version>
 </properties>
 
-<name>Apache StreamPipes (incubating)</name>
+<name>Apache StreamPipes</name>
 <description>A self-service (Industrial) IoT toolbox to enable non-technical users to connect, analyze and explore
 IoT data streams.
 		</description>
@@ -1453,7 +1453,7 @@ IoT data streams.
 									<fileSet>
 										<directory>${project.build.directory}</directory>
 										<includes>
-											<include>apache-streampipes-${project.version}-incubating-source-release.zip
+											<include>apache-streampipes-${project.version}-source-release.zip
 										</include>
 									</includes>
 								</fileSet>
@@ -1549,7 +1549,7 @@ IoT data streams.
 					<artifactId>maven-assembly-plugin</artifactId>
 					<version>${maven-assembly-plugin.version}</version>
 					<configuration>
-						<finalName>apache-streampipes-${project.version}-incubating</finalName>
+						<finalName>apache-streampipes-${project.version}</finalName>
 						<descriptors>
 							<descriptor>tools/maven/assembly.xml</descriptor>
 						</descriptors>
@@ -1848,10 +1848,10 @@ IoT data streams.
 	</build>
 
 	<scm>
-		<developerConnection>scm:git:ssh://git@github.com/apache/incubator-streampipes.git</developerConnection>
-		<connection>scm:git:ssh://git@github.com/apache/incubator-streampipes.git</connection>
-		<url>https://github.com/apache/incubator-streampipes</url>
-		<tag>rel/0.70.0</tag>
+		<developerConnection>scm:git:ssh://git@github.com/apache/streampipes.git</developerConnection>
+		<connection>scm:git:ssh://git@github.com/apache/streampipes.git</connection>
+		<url>https://github.com/apache/streampipes</url>
+		<tag>dev</tag>
 	</scm>
 
 	<issueManagement>
diff --git a/streampipes-client-python/.pre-commit-config.yaml b/streampipes-client-python/.pre-commit-config.yaml
index 76b2829944..d353a24d79 100644
--- a/streampipes-client-python/.pre-commit-config.yaml
+++ b/streampipes-client-python/.pre-commit-config.yaml
@@ -24,9 +24,17 @@ repos:
       - id: end-of-file-fixer
       - id: trailing-whitespace
       - id: mixed-line-ending
+      - id: check-docstring-first
+      - id: check-merge-conflict
 
   - repo: local
     hooks:
+      - id: interrogate
+        name: interrogate
+        language: python
+        types: [ python ]
+        entry:  interrogate -vv --fail-under 95 --omit-covered-files --ignore-init-method --ignore-module --ignore-magic
+
       - id: pyupgrade
         name: pyupgrade
         language: python
diff --git a/streampipes-client-python/README.md b/streampipes-client-python/README.md
index 5c61e1fe29..69241a87df 100644
--- a/streampipes-client-python/README.md
+++ b/streampipes-client-python/README.md
@@ -19,7 +19,7 @@
 
 <h1 align="center">
   <br>
-   <img src="https://raw.githubusercontent.com/apache/incubator-streampipes/STREAMPIPES-607/streampipes-client-python/docs/img/streampipes-python.png"
+   <img src="https://raw.githubusercontent.com/apache/incubator-streampipes/dev/streampipes-client-python/docs/img/streampipes-python.png"
    alt="StreamPipes Logo with Python" title="Apache StreamPipes Logo with Python" width="75%"/>
   <br>
 </h1>
@@ -99,4 +99,4 @@ StreamPipesApiKeyCredentials.from_env(username_env="USER", api_key_env="API-KEY"
 `username` is always the username that is used to log in into StreamPipes. <br>
 The `api_key` can be generated within the UI as demonstrated below:
 
-![Howto API Key](https://raw.githubusercontent.com/apache/incubator-streampipes/STREAMPIPES-607/streampipes-client-python/docs/img/how-to-get-api-key.gif)
\ No newline at end of file
+![Howto API Key](https://raw.githubusercontent.com/apache/incubator-streampipes/dev/streampipes-client-python/docs/img/how-to-get-api-key.gif)
\ No newline at end of file
diff --git a/streampipes-client-python/setup.py b/streampipes-client-python/setup.py
index f36a88fda2..ee1d50c287 100644
--- a/streampipes-client-python/setup.py
+++ b/streampipes-client-python/setup.py
@@ -40,6 +40,7 @@ dev_packages = base_packages + [
     "autoflake>=1.7.7",
     "black>=22.10.0",
     "flake8>=5.0.4",
+    "interrogate>=1.5.0",
     "isort>=5.10.1",
     "mypy>=0.990",
     "pandas-stubs>=1.5.1.221024",
diff --git a/streampipes-client-python/streampipes_client/model/common.py b/streampipes-client-python/streampipes_client/model/common.py
index c9eec9a187..dd1cd393d9 100644
--- a/streampipes-client-python/streampipes_client/model/common.py
+++ b/streampipes-client-python/streampipes_client/model/common.py
@@ -38,9 +38,15 @@ def _snake_to_camel_case(snake_case_string: str) -> str:
 
 
 class BasicModel(BaseModel):
+    """Basic model class used for the whole Python StreamPipes data model."""
+
     element_id: Optional[StrictStr]
 
     class Config:
+        """Configuration class for Pydantic.
+        Defines alias generator to convert field names from camelCase (API) to snake_case (Python codebase).
+        """
+
         alias_generator = _snake_to_camel_case
 
 
diff --git a/streampipes-client-python/streampipes_client/model/container/resource_container.py b/streampipes-client-python/streampipes_client/model/container/resource_container.py
index 90bf2b109b..0b16c430ec 100644
--- a/streampipes-client-python/streampipes_client/model/container/resource_container.py
+++ b/streampipes-client-python/streampipes_client/model/container/resource_container.py
@@ -56,6 +56,13 @@ class StreamPipesDataModelError(Exception):
         super().__init__(self._generate_error_message())
 
     def _generate_error_message(self) -> str:
+        """Generate a specific error message for this exception.
+        Error message contains information derived from the specific `ValidationError`.
+
+        Returns
+        -------
+        The error description (`str`)
+        """
         return (
             f"\nOops, there seems to be a problem with our internal StreamPipes data model.\n"
             f"This should not occur, but unfortunately did.\n"
@@ -89,6 +96,13 @@ class StreamPipesResourceContainerJSONError(Exception):
         super().__init__(self._generate_error_message())
 
     def _generate_error_message(self) -> str:
+        """Generate a specific error message for this exception.
+        Error message contains information about the related container model and the causing JSON string.
+
+        Returns
+        -------
+        The error description (`str`)
+        """
         return (
             f"\nOops, there seems to be a problem when parsing the response of the StreamPipes API."
             f"This should not occur, but unfortunately did.\n"
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
index afee4fa083..5440661e3a 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
+++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
@@ -33,6 +33,7 @@ import org.influxdb.dto.Point;
 import org.influxdb.dto.Pong;
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -178,7 +179,7 @@ public class InfluxStore {
     }
 
     if (missingFields.size() > 0) {
-      LOG.warn("Ignored {} fields which were present in the schema, but not in the provided event: {}",
+      LOG.debug("Ignored {} fields which were present in the schema, but not in the provided event: {}",
           missingFields.size(),
           String.join(", ", missingFields));
     }
@@ -191,7 +192,7 @@ public class InfluxStore {
   }
 
   private void handleMeasurementProperty(Point.Builder p,
-                                         EventPropertyPrimitive ep,
+                                         @NotNull EventPropertyPrimitive ep,
                                          String preparedRuntimeName,
                                          PrimitiveField eventPropertyPrimitiveField) {
     try {
diff --git a/streampipes-extensions/README.md b/streampipes-extensions/README.md
index 7339eef641..73fbfe9bdc 100644
--- a/streampipes-extensions/README.md
+++ b/streampipes-extensions/README.md
@@ -18,7 +18,7 @@
 [![Github Actions](https://img.shields.io/github/workflow/status/apache/incubator-streampipes-extensions/Build%20and%20Deploy%20Extensions)](https://github.com/apache/incubator-streampipes-extensions/actions/)
 # Apache StreamPipes
 
-Apache StreamPipes (incubating) enables flexible modeling of stream processing pipelines by providing a graphical 
+Apache StreamPipes enables flexible modeling of stream processing pipelines by providing a graphical 
 modeling editor on top of existing stream processing frameworks.
 
 It empowers non-technical users to quickly define and execute processing pipelines based on an easily extensible 
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarConsumer.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarConsumer.java
deleted file mode 100644
index 1dc30ec391..0000000000
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarConsumer.java
+++ /dev/null
@@ -1,85 +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.connect.iiot.protocol.stream.pulsar;
-
-import org.apache.commons.lang3.RandomStringUtils;
-import org.apache.pulsar.client.api.*;
-import org.apache.streampipes.messaging.InternalEventProcessor;
-
-public class PulsarConsumer implements Runnable {
-
-  private InternalEventProcessor<byte[]> adapterConsumer;
-  private String brokerUrl;
-  private String topic;
-  private Integer maxElementsToReceive = -1;
-  private Integer messageCount = 0;
-  private Boolean running = true;
-
-  private PulsarClient pulsarClient;
-
-
-  public PulsarConsumer(String brokerUrl, String topic, InternalEventProcessor<byte[]> adapterConsumer) {
-    this.brokerUrl = brokerUrl;
-    this.topic = topic;
-    this.adapterConsumer = adapterConsumer;
-  }
-
-  public PulsarConsumer(String brokerUrl, String topic, InternalEventProcessor<byte[]> adapterConsumer,
-                        int maxElementsToReceive) {
-    this(brokerUrl, topic, adapterConsumer);
-    this.maxElementsToReceive = maxElementsToReceive;
-  }
-
-  @Override
-  public void run() {
-    this.running = true;
-    try {
-      this.pulsarClient = PulsarUtils.makePulsarClient(this.brokerUrl);
-
-      Consumer<byte[]> consumer = this.pulsarClient.newConsumer()
-              .topic(this.topic)
-              .subscriptionName(RandomStringUtils.randomAlphanumeric(10))
-              .subscriptionType(SubscriptionType.Shared)
-              .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
-              .subscribe();
-
-      while (running && ((maxElementsToReceive == -1) || (this.messageCount <= maxElementsToReceive))) {
-        Message msg = consumer.receive();
-        adapterConsumer.onEvent(msg.getData());
-        consumer.acknowledge(msg);
-        this.messageCount++;
-      }
-
-      consumer.close();
-      this.pulsarClient.close();
-    } catch (PulsarClientException e) {
-      e.printStackTrace();
-    }
-  }
-
-  public void stop() throws PulsarClientException {
-    this.running = false;
-    this.pulsarClient.close();
-  }
-
-  public Integer getMessageCount() {
-    return this.messageCount;
-  }
-
-
-}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarProtocol.java
index baad678d7f..9e24d5bde1 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/pulsar/PulsarProtocol.java
@@ -19,11 +19,16 @@ package org.apache.streampipes.connect.iiot.protocol.stream.pulsar;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageListener;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Reader;
+
+import org.apache.streampipes.commons.exceptions.SpConfigurationException;
 import org.apache.streampipes.connect.SendToPipeline;
 import org.apache.streampipes.connect.adapter.model.generic.Protocol;
 import org.apache.streampipes.connect.adapter.sdk.ParameterExtractor;
@@ -32,24 +37,24 @@ import org.apache.streampipes.connect.api.IFormat;
 import org.apache.streampipes.connect.api.IParser;
 import org.apache.streampipes.connect.api.exception.ParseException;
 import org.apache.streampipes.connect.iiot.protocol.stream.BrokerProtocol;
-import org.apache.streampipes.container.api.ResolvesContainerProvidedOptions;
-import org.apache.streampipes.messaging.InternalEventProcessor;
+import org.apache.streampipes.container.api.SupportsRuntimeConfig;
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
-import org.apache.streampipes.model.staticproperty.Option;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.sdk.builder.adapter.ProtocolDescriptionBuilder;
 import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
 import org.apache.streampipes.sdk.helpers.AdapterSourceType;
 import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
 import org.apache.streampipes.sdk.utils.Assets;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
 
-public class PulsarProtocol extends BrokerProtocol implements ResolvesContainerProvidedOptions {
+public class PulsarProtocol extends BrokerProtocol implements SupportsRuntimeConfig {
 
   private static final Logger LOG = LoggerFactory.getLogger(PulsarProtocol.class);
 
@@ -58,16 +63,18 @@ public class PulsarProtocol extends BrokerProtocol implements ResolvesContainerP
   public static final String PULSAR_BROKER_HOST = "pulsar-broker-host";
   public static final String PULSAR_BROKER_PORT = "pulsar-broker-port";
   public static final String PULSAR_TOPIC = "pulsar-topic";
+  public static final String PULSAR_SUBSCRIPTION_NAME = "pulsar-subscription-name";
 
-  private Thread thread;
-  private PulsarConsumer pulsarConsumer;
+  private String subscriptionName;
+  private Consumer<byte[]> consumer;
 
   public PulsarProtocol() {
 
   }
 
-  public PulsarProtocol(IParser parser, IFormat format, String brokerUrl, String topic) {
+  public PulsarProtocol(IParser parser, IFormat format, String brokerUrl, String topic, String subscriptionName) {
     super(parser, format, brokerUrl, topic);
+    this.subscriptionName = subscriptionName;
   }
 
   @Override
@@ -100,8 +107,9 @@ public class PulsarProtocol extends BrokerProtocol implements ResolvesContainerP
     Integer brokerPort = extractor.singleValue(PULSAR_BROKER_PORT, Integer.class);
     String brokerUrl = brokerHost + ":" + brokerPort;
     String topic = extractor.singleValue(PULSAR_TOPIC, String.class);
+    String subscriptionName = extractor.singleValue(PULSAR_SUBSCRIPTION_NAME, String.class);
 
-    return new PulsarProtocol(parser, format, brokerUrl, topic);
+    return new PulsarProtocol(parser, format, brokerUrl, topic, subscriptionName);
   }
 
   @Override
@@ -114,28 +122,44 @@ public class PulsarProtocol extends BrokerProtocol implements ResolvesContainerP
             .requiredTextParameter(Labels.withId(PULSAR_BROKER_HOST))
             .requiredIntegerParameter(Labels.withId(PULSAR_BROKER_PORT), 6650)
             .requiredTextParameter(Labels.withId(PULSAR_TOPIC))
-//            .requiredSingleValueSelectionFromContainer(Labels.from(PULSAR_TOPIC, "Topic",
-//                    "Example: topic"), Arrays.asList(PULSAR_BROKER_HOST, PULSAR_BROKER_PORT))
+            .requiredTextParameter(Labels.withId(PULSAR_SUBSCRIPTION_NAME))
             .build();
   }
 
   @Override
   public void run(IAdapterPipeline adapterPipeline) {
     SendToPipeline stk = new SendToPipeline(format, adapterPipeline);
-    this.pulsarConsumer = new PulsarConsumer(this.brokerUrl,
-            this.topic, stk::emit);
 
-    thread = new Thread(this.pulsarConsumer);
-    thread.start();
+    try {
+      if (consumer != null) {
+        consumer.close();
+      }
+      PulsarClient client = PulsarUtils.makePulsarClient(brokerUrl);
+      consumer = client.newConsumer()
+              .topic(topic)
+              .subscriptionName(subscriptionName)
+              .messageListener((MessageListener<byte[]>) (consumer, msg) -> {
+                try {
+                  stk.emit(msg.getValue());
+                } catch (ParseException e) {
+                  LOG.error("Failed to parse message.", e);
+                }
+              }).subscribe();
+    } catch (PulsarClientException e) {
+      throw new RuntimeException(e);
+    }
   }
 
   @Override
   public void stop() {
-    try {
-      this.pulsarConsumer.stop();
-    } catch (PulsarClientException e) {
-      e.printStackTrace();
+    if(consumer != null) {
+      try {
+        consumer.close();
+      } catch (PulsarClientException e) {
+        throw new RuntimeException(e);
+      }
     }
+    consumer = null;
   }
 
   @Override
@@ -144,15 +168,16 @@ public class PulsarProtocol extends BrokerProtocol implements ResolvesContainerP
   }
 
   @Override
-  public List<Option> resolveOptions(String requestId, StaticPropertyExtractor extractor) {
+  public StaticProperty resolveConfiguration(String staticPropertyInternalName, StaticPropertyExtractor extractor) throws
+          SpConfigurationException {
     String brokerHost = extractor.singleValueParameter(PULSAR_BROKER_HOST, String.class);
     Integer brokerPort = extractor.singleValueParameter(PULSAR_BROKER_PORT, Integer.class);
 
     try {
       PulsarClient client = PulsarUtils.makePulsarClient(brokerHost + ":" + brokerPort);
-      return new ArrayList<>();
+      return null;
     } catch (PulsarClientException e) {
-      return new ArrayList<>();
+      throw new SpConfigurationException(e);
     }
   }
 }
diff --git a/streampipes-wrapper-python/LICENSE b/streampipes-wrapper-python/LICENSE
deleted file mode 100644
index 56ee3c8c4c..0000000000
--- a/streampipes-wrapper-python/LICENSE
+++ /dev/null
@@ -1,201 +0,0 @@
-                                Apache License
-                           Version 2.0, January 2004
-                        http://www.apache.org/licenses/
-
-   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
-   1. Definitions.
-
-      "License" shall mean the terms and conditions for use, reproduction,
-      and distribution as defined by Sections 1 through 9 of this document.
-
-      "Licensor" shall mean the copyright owner or entity authorized by
-      the copyright owner that is granting the License.
-
-      "Legal Entity" shall mean the union of the acting entity and all
-      other entities that control, are controlled by, or are under common
-      control with that entity. For the purposes of this definition,
-      "control" means (i) the power, direct or indirect, to cause the
-      direction or management of such entity, whether by contract or
-      otherwise, or (ii) ownership of fifty percent (50%) or more of the
-      outstanding shares, or (iii) beneficial ownership of such entity.
-
-      "You" (or "Your") shall mean an individual or Legal Entity
-      exercising permissions granted by this License.
-
-      "Source" form shall mean the preferred form for making modifications,
-      including but not limited to software source code, documentation
-      source, and configuration files.
-
-      "Object" form shall mean any form resulting from mechanical
-      transformation or translation of a Source form, including but
-      not limited to compiled object code, generated documentation,
-      and conversions to other media types.
-
-      "Work" shall mean the work of authorship, whether in Source or
-      Object form, made available under the License, as indicated by a
-      copyright notice that is included in or attached to the work
-      (an example is provided in the Appendix below).
-
-      "Derivative Works" shall mean any work, whether in Source or Object
-      form, that is based on (or derived from) the Work and for which the
-      editorial revisions, annotations, elaborations, or other modifications
-      represent, as a whole, an original work of authorship. For the purposes
-      of this License, Derivative Works shall not include works that remain
-      separable from, or merely link (or bind by name) to the interfaces of,
-      the Work and Derivative Works thereof.
-
-      "Contribution" shall mean any work of authorship, including
-      the original version of the Work and any modifications or additions
-      to that Work or Derivative Works thereof, that is intentionally
-      submitted to Licensor for inclusion in the Work by the copyright owner
-      or by an individual or Legal Entity authorized to submit on behalf of
-      the copyright owner. For the purposes of this definition, "submitted"
-      means any form of electronic, verbal, or written communication sent
-      to the Licensor or its representatives, including but not limited to
-      communication on electronic mailing lists, source code control systems,
-      and issue tracking systems that are managed by, or on behalf of, the
-      Licensor for the purpose of discussing and improving the Work, but
-      excluding communication that is conspicuously marked or otherwise
-      designated in writing by the copyright owner as "Not a Contribution."
-
-      "Contributor" shall mean Licensor and any individual or Legal Entity
-      on behalf of whom a Contribution has been received by Licensor and
-      subsequently incorporated within the Work.
-
-   2. Grant of Copyright License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      copyright license to reproduce, prepare Derivative Works of,
-      publicly display, publicly perform, sublicense, and distribute the
-      Work and such Derivative Works in Source or Object form.
-
-   3. Grant of Patent License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      (except as stated in this section) patent license to make, have made,
-      use, offer to sell, sell, import, and otherwise transfer the Work,
-      where such license applies only to those patent claims licensable
-      by such Contributor that are necessarily infringed by their
-      Contribution(s) alone or by combination of their Contribution(s)
-      with the Work to which such Contribution(s) was submitted. If You
-      institute patent litigation against any entity (including a
-      cross-claim or counterclaim in a lawsuit) alleging that the Work
-      or a Contribution incorporated within the Work constitutes direct
-      or contributory patent infringement, then any patent licenses
-      granted to You under this License for that Work shall terminate
-      as of the date such litigation is filed.
-
-   4. Redistribution. You may reproduce and distribute copies of the
-      Work or Derivative Works thereof in any medium, with or without
-      modifications, and in Source or Object form, provided that You
-      meet the following conditions:
-
-      (a) You must give any other recipients of the Work or
-          Derivative Works a copy of this License; and
-
-      (b) You must cause any modified files to carry prominent notices
-          stating that You changed the files; and
-
-      (c) You must retain, in the Source form of any Derivative Works
-          that You distribute, all copyright, patent, trademark, and
-          attribution notices from the Source form of the Work,
-          excluding those notices that do not pertain to any part of
-          the Derivative Works; and
-
-      (d) If the Work includes a "NOTICE" text file as part of its
-          distribution, then any Derivative Works that You distribute must
-          include a readable copy of the attribution notices contained
-          within such NOTICE file, excluding those notices that do not
-          pertain to any part of the Derivative Works, in at least one
-          of the following places: within a NOTICE text file distributed
-          as part of the Derivative Works; within the Source form or
-          documentation, if provided along with the Derivative Works; or,
-          within a display generated by the Derivative Works, if and
-          wherever such third-party notices normally appear. The contents
-          of the NOTICE file are for informational purposes only and
-          do not modify the License. You may add Your own attribution
-          notices within Derivative Works that You distribute, alongside
-          or as an addendum to the NOTICE text from the Work, provided
-          that such additional attribution notices cannot be construed
-          as modifying the License.
-
-      You may add Your own copyright statement to Your modifications and
-      may provide additional or different license terms and conditions
-      for use, reproduction, or distribution of Your modifications, or
-      for any such Derivative Works as a whole, provided Your use,
-      reproduction, and distribution of the Work otherwise complies with
-      the conditions stated in this License.
-
-   5. Submission of Contributions. Unless You explicitly state otherwise,
-      any Contribution intentionally submitted for inclusion in the Work
-      by You to the Licensor shall be under the terms and conditions of
-      this License, without any additional terms or conditions.
-      Notwithstanding the above, nothing herein shall supersede or modify
-      the terms of any separate license agreement you may have executed
-      with Licensor regarding such Contributions.
-
-   6. Trademarks. This License does not grant permission to use the trade
-      names, trademarks, service marks, or product names of the Licensor,
-      except as required for reasonable and customary use in describing the
-      origin of the Work and reproducing the content of the NOTICE file.
-
-   7. Disclaimer of Warranty. Unless required by applicable law or
-      agreed to in writing, Licensor provides the Work (and each
-      Contributor provides its Contributions) on an "AS IS" BASIS,
-      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-      implied, including, without limitation, any warranties or conditions
-      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
-      PARTICULAR PURPOSE. You are solely responsible for determining the
-      appropriateness of using or redistributing the Work and assume any
-      risks associated with Your exercise of permissions under this License.
-
-   8. Limitation of Liability. In no event and under no legal theory,
-      whether in tort (including negligence), contract, or otherwise,
-      unless required by applicable law (such as deliberate and grossly
-      negligent acts) or agreed to in writing, shall any Contributor be
-      liable to You for damages, including any direct, indirect, special,
-      incidental, or consequential damages of any character arising as a
-      result of this License or out of the use or inability to use the
-      Work (including but not limited to damages for loss of goodwill,
-      work stoppage, computer failure or malfunction, or any and all
-      other commercial damages or losses), even if such Contributor
-      has been advised of the possibility of such damages.
-
-   9. Accepting Warranty or Additional Liability. While redistributing
-      the Work or Derivative Works thereof, You may choose to offer,
-      and charge a fee for, acceptance of support, warranty, indemnity,
-      or other liability obligations and/or rights consistent with this
-      License. However, in accepting such obligations, You may act only
-      on Your own behalf and on Your sole responsibility, not on behalf
-      of any other Contributor, and only if You agree to indemnify,
-      defend, and hold each Contributor harmless for any liability
-      incurred by, or claims asserted against, such Contributor by reason
-      of your accepting any such warranty or additional liability.
-
-   END OF TERMS AND CONDITIONS
-
-   APPENDIX: How to apply the Apache License to your work.
-
-      To apply the Apache License to your work, attach the following
-      boilerplate notice, with the fields enclosed by brackets "[]"
-      replaced with your own identifying information. (Don't include
-      the brackets!)  The text should be enclosed in the appropriate
-      comment syntax for the file format. We also recommend that a
-      file or class name and description of purpose be included on the
-      same "printed page" as the copyright notice for easier
-      identification within third-party archives.
-
-   Copyright [yyyy] [name of copyright owner]
-
-   Licensed 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.
\ No newline at end of file
diff --git a/streampipes-wrapper-python/NOTICE b/streampipes-wrapper-python/NOTICE
deleted file mode 100644
index 17b697e002..0000000000
--- a/streampipes-wrapper-python/NOTICE
+++ /dev/null
@@ -1,7 +0,0 @@
-Apache StreamPipes (incubating)
-Copyright 2020 The Apache Software Foundation.
-
-This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
-
-============================================================================
\ No newline at end of file
diff --git a/streampipes-wrapper-python/README.md b/streampipes-wrapper-python/README.md
index 9d31a294ea..96255615a9 100644
--- a/streampipes-wrapper-python/README.md
+++ b/streampipes-wrapper-python/README.md
@@ -29,7 +29,7 @@
 > The StreamPipes wrapper for python is currently under development. Thus, the processor model description still needs to be implemented externally in Java.
 
 ## Apache StreamPipes
-Apache StreamPipes (incubating) enables flexible modeling of stream processing pipelines by providing a graphical 
+Apache StreamPipes enables flexible modeling of stream processing pipelines by providing a graphical 
 modeling editor on top of existing stream processing frameworks.
 
 It leverages non-technical users to quickly define and execute processing pipelines based on an easily extensible 
@@ -98,13 +98,3 @@ def main():
 if __name__ == '__main__':
     main()
 ````
-
-### DISCLAIMER (WIP)
-Apache StreamPipes is an effort undergoing incubation at The Apache Software Foundation (ASF), sponsored by the Apache
-Incubator. Incubation is required of all newly accepted projects until a further review indicates that the
-infrastructure, communications, and decision making process have stabilized in a manner consistent with other successful ASF projects. While incubation status is not necessarily a reflection of the completeness or stability of the code, it does indicate that the project has yet to be fully endorsed by the ASF.
-
-Some of the incubating project’s releases may not be fully compliant with ASF policy. For example, releases may have incomplete or un-reviewed licensing conditions. What follows is a list of known issues the project is currently aware of (note that this list, by definition, is likely to be incomplete):
-(currently no issues are known)
-
-If you are planning to incorporate this work into your product/project, please be aware that you will need to conduct a thorough licensing review to determine the overall implications of including this work. For the current status of this project through the Apache Incubator visit: https://incubator.apache.org/projects/streampipes.html
diff --git a/ui/package.json b/ui/package.json
index 1a1e790d79..833a04193a 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -1,7 +1,7 @@
 {
   "name": "apache-streampipes",
   "version": "0.71.0-SNAPSHOT",
-  "description": "Apache StreamPipes (incubating) user interface",
+  "description": "Apache StreamPipes user interface",
   "homepage": "http://streampipes.apache.org",
   "repository": {
     "type": "git",


Test patch:
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/PulsarAdapterTester.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/PulsarAdapterTester.java
index 566a835e21..27cdbb3274 100644
--- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/PulsarAdapterTester.java
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/PulsarAdapterTester.java
@@ -69,6 +69,8 @@ public class PulsarAdapterTester extends AdapterTesterBase {
                 pulsarContainer.getBrokerPort())
             .requiredTextParameter(
                 Labels.withId(PulsarProtocol.PULSAR_TOPIC), TOPIC)
+            .requiredTextParameter(
+                Labels.withId(PulsarProtocol.PULSAR_SUBSCRIPTION_NAME), "test-sub")
             .build())
         .addRules(List.of(new DebugSinkRuleDescription()))
         .build();