Repository: apache/streampipes
Pull request #1169 (2023-01-26 21:04:56)

Replace consul-client with consul-api library (#1158)
<!--
  ~ 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.
-->

Replace consul library and add an initial version of improved handling of environment variables to facilitate easier testing.

### 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): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/docker-compose.yml b/docker-compose.yml
index 96be1ae7ba..e324ecf825 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -76,11 +76,13 @@ services:
       spnet:
 
   consul:
-    image: fogsyio/consul:1.9.6
+    image: consul:1.14.3
     environment:
-      - "CONSUL_LOCAL_CONFIG={\"disable_update_check\": true}"
+      - "CONSUL_LOCAL_CONFIG={\"rpc_streaming\": false, \"disable_update_check\": true, \"rpc\": {\"enable_streaming\": false}, \"use_streaming_backend\": false}"
       - "CONSUL_BIND_INTERFACE=eth0"
       - "CONSUL_HTTP_ADDR=0.0.0.0"
+      - "CONSUL_RPC_ENABLE_STREAMING=false"
+      - "CONSUL_USE_STREAMING_BACKEND=false"
     entrypoint:
       - consul
       - agent
diff --git a/installer/cli/deploy/standalone/consul/docker-compose.yml b/installer/cli/deploy/standalone/consul/docker-compose.yml
index 49dca0e953..dd26e0af96 100644
--- a/installer/cli/deploy/standalone/consul/docker-compose.yml
+++ b/installer/cli/deploy/standalone/consul/docker-compose.yml
@@ -16,11 +16,13 @@
 version: "3.4"
 services:
   consul:
-    image: fogsyio/consul:1.9.6
+    image: consul:1.14.3
     environment:
-      - "CONSUL_LOCAL_CONFIG={\"disable_update_check\": true}"
+      - "CONSUL_LOCAL_CONFIG={\"rpc_streaming\": false, \"disable_update_check\": true, \"rpc\": {\"enable_streaming\": false}, \"use_streaming_backend\": false}"
       - "CONSUL_BIND_INTERFACE=eth0"
       - "CONSUL_HTTP_ADDR=0.0.0.0"
+      - "CONSUL_RPC_ENABLE_STREAMING=false"
+      - "CONSUL_USE_STREAMING_BACKEND=false"
     entrypoint:
       - consul
       - agent
diff --git a/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml b/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml
index b01ab7c0fa..8cfad035c3 100644
--- a/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml
+++ b/installer/cli/deploy/standalone/iotdb/docker-compose.dev.yml
@@ -16,5 +16,8 @@
 version: "3.4"
 services:
   iotdb:
+    # 0.13.3 is the only recommended production ready version for edge side (standalone) deployment
+    image: apache/iotdb:0.13.3-node
     ports:
-      - 6667:6667
+      # rpc port
+      - "6667:6667"
diff --git a/installer/cli/deploy/standalone/iotdb/docker-compose.yml b/installer/cli/deploy/standalone/iotdb/docker-compose.yml
index eb315f3433..f45807b9d6 100644
--- a/installer/cli/deploy/standalone/iotdb/docker-compose.yml
+++ b/installer/cli/deploy/standalone/iotdb/docker-compose.yml
@@ -16,9 +16,10 @@
 version: "3.4"
 services:
   iotdb:
-    image: apache/iotdb
+    # 0.13.3 is the only recommended production ready version for edge side (standalone) deployment
+    image: apache/iotdb:0.13.3-node
     volumes:
-      - iotdb:/iotdb/data
+      - iotdb:/iotdb
     logging:
       driver: "json-file"
       options:
diff --git a/installer/compose/docker-compose.full.yml b/installer/compose/docker-compose.full.yml
index 83bf879a46..687719e866 100644
--- a/installer/compose/docker-compose.full.yml
+++ b/installer/compose/docker-compose.full.yml
@@ -82,7 +82,7 @@ services:
       spnet:
 
   consul:
-    image: fogsyio/consul:1.9.6
+    image: consul:1.14.3
     environment:
       - "CONSUL_LOCAL_CONFIG={\"disable_update_check\": true}"
       - "CONSUL_BIND_INTERFACE=eth0"
diff --git a/installer/compose/docker-compose.nats.yml b/installer/compose/docker-compose.nats.yml
index ba115d92d4..e977e333e4 100644
--- a/installer/compose/docker-compose.nats.yml
+++ b/installer/compose/docker-compose.nats.yml
@@ -55,7 +55,7 @@ services:
       spnet:
 
   consul:
-    image: fogsyio/consul:1.9.6
+    image: consul:1.14.3
     environment:
       - "CONSUL_LOCAL_CONFIG={\"disable_update_check\": true}"
       - "CONSUL_BIND_INTERFACE=eth0"
diff --git a/installer/compose/docker-compose.yml b/installer/compose/docker-compose.yml
index a5d7e4727c..0657b86017 100644
--- a/installer/compose/docker-compose.yml
+++ b/installer/compose/docker-compose.yml
@@ -54,7 +54,7 @@ services:
       spnet:
 
   consul:
-    image: fogsyio/consul:1.9.6
+    image: consul:1.14.3
     environment:
       - "CONSUL_LOCAL_CONFIG={\"disable_update_check\": true}"
       - "CONSUL_BIND_INTERFACE=eth0"
diff --git a/installer/k8s/templates/external/consul/consul-deployment.yaml b/installer/k8s/templates/external/consul/consul-deployment.yaml
index b80fc11aa7..18fd60801c 100644
--- a/installer/k8s/templates/external/consul/consul-deployment.yaml
+++ b/installer/k8s/templates/external/consul/consul-deployment.yaml
@@ -33,7 +33,7 @@ spec:
             claimName: consul-pvc
       containers:
         - name: consul
-          image: fogsyio/consul:{{ .Values.external.consulVersion }}
+          image: consul:{{ .Values.external.consulVersion }}
           imagePullPolicy: {{ .Values.pullPolicy }}
           command:
             - "/bin/sh"
diff --git a/installer/k8s/values.yaml b/installer/k8s/values.yaml
index cea3b6fbf1..d322e36cf2 100644
--- a/installer/k8s/values.yaml
+++ b/installer/k8s/values.yaml
@@ -25,7 +25,7 @@ streampipes:
   registry: "apachestreampipes"
 
 external:
-  consulVersion: 1.9.6
+  consulVersion: 1.14.3
   couchdbVersion: 2.3.1
   flinkVersion: 1.13.5-scala_2.11
   kafkaVersion: 2.2.0
diff --git a/pom.xml b/pom.xml
index ec429e77c2..045286e085 100644
--- a/pom.xml
+++ b/pom.xml
@@ -116,7 +116,7 @@
 	<retrofit.version>2.5.0</retrofit.version>
 	<roaster.version>2.28.0.Final</roaster.version>
 	<siddhi.version>5.1.27</siddhi.version>
-	<simple-java-mail.version>7.6.0</simple-java-mail.version>
+	<simple-java-mail.version>7.8.0</simple-java-mail.version>
 	<slf4j.version>2.0.6</slf4j.version>
 	<snakeyaml.version>1.32</snakeyaml.version>
 	<snappy-java.version>1.1.7.7</snappy-java.version>
@@ -157,7 +157,6 @@
 	<errorprone-annotations.version>2.2.0</errorprone-annotations.version>
 	<google-maps-services.version>0.10.0</google-maps-services.version>
 	<graalvm.js.version>22.3.1</graalvm.js.version>
-	<iotdb.version>1.0.0</iotdb.version>
 	<java-websocket.version>1.5.0</java-websocket.version>
 	<jakarta-websocket-client-api.version>2.1.0</jakarta-websocket-client-api.version>
 	<jaxb-runtime.version>2.3.2</jaxb-runtime.version>
@@ -213,7 +212,7 @@
 	<maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
 	<maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version>
 	<maven.dependency.check.plugin.version>6.5.1</maven.dependency.check.plugin.version>
-	<maven-dependency-plugin.version>3.2.0</maven-dependency-plugin.version>
+	<maven-dependency-plugin.version>3.5.0</maven-dependency-plugin.version>
 	<maven-javadoc-plugin.version>3.3.1</maven-javadoc-plugin.version>
 	<mave-release-plugin.version>3.0.0-M5</mave-release-plugin.version>
 	<maven-source-plugin.version>3.2.1</maven-source-plugin.version>
@@ -1283,17 +1282,6 @@
                 <artifactId>flink-runtime_2.11</artifactId>
                 <version>${flink.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.apache.iotdb</groupId>
-                <artifactId>iotdb-jdbc</artifactId>
-                <version>${iotdb.version}</version>
-                <exclusions>
-                    <exclusion>
-                        <groupId>javax.servlet</groupId>
-                        <artifactId>servlet-api</artifactId>
-                    </exclusion>
-                </exclusions>
-            </dependency>
             <dependency>
                 <groupId>org.apache.opennlp</groupId>
                 <artifactId>opennlp-tools</artifactId>
diff --git a/streampipes-client-python/docs/examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb b/streampipes-client-python/docs/examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb
index 3b69c065db..3f3a3c9c31 100644
--- a/streampipes-client-python/docs/examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb
+++ b/streampipes-client-python/docs/examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb
@@ -610,4 +610,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}
\ No newline at end of file
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index 9ad589bb6b..5460500583 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -19,35 +19,37 @@ package org.apache.streampipes.commons.constants;
 
 public enum Envs {
 
-  SP_HOST("SP_HOST"),
-  SP_PORT("SP_PORT"),
+  SP_HOST("SP_HOST", null),
+  SP_PORT("SP_PORT", null),
 
   @Deprecated(since = "0.90.0", forRemoval = true)
-  SP_CONSUL_LOCATION("CONSUL_LOCATION"),
+  SP_CONSUL_LOCATION("CONSUL_LOCATION", "consul"),
 
-  SP_CONSUL_HOST("SP_CONSUL_HOST"),
-  SP_CONSUL_PORT("SP_CONSUL_PORT"),
-  SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS"),
-  SP_JWT_SECRET("JWT_SECRET"),
-  SP_JWT_SIGNING_MODE("SP_JWT_SIGNING_MODE"),
-  SP_JWT_PRIVATE_KEY_LOC("SP_JWT_PRIVATE_KEY_LOC"),
-  SP_JWT_PUBLIC_KEY_LOC("SP_JWT_PUBLIC_KEY_LOC"),
-  SP_INITIAL_ADMIN_EMAIL("SP_INITIAL_ADMIN_EMAIL"),
-  SP_INITIAL_ADMIN_PASSWORD("SP_INITIAL_ADMIN_PASSWORD"),
-  SP_INITIAL_SERVICE_USER("SP_INITIAL_SERVICE_USER"),
-  SP_INITIAL_SERVICE_USER_SECRET("SP_INITIAL_SERVICE_USER_SECRET"),
-  SP_SETUP_INSTALL_PIPELINE_ELEMENTS("SP_SETUP_INSTALL_PIPELINE_ELEMENTS"),
-  SP_EXT_AUTH_MODE("SP_EXT_AUTH_MODE"),
-  SP_CLIENT_USER("SP_CLIENT_USER"),
-  SP_CLIENT_SECRET("SP_CLIENT_SECRET"),
-  SP_ENCRYPTION_PASSCODE("SP_ENCRYPTION_PASSCODE"),
-  SP_DEBUG("SP_DEBUG"),
-  SP_MAX_WAIT_TIME_AT_SHUTDOWN("SP_MAX_WAIT_TIME_AT_SHUTDOWN");
+  SP_CONSUL_HOST("SP_CONSUL_HOST", "consul"),
+  SP_CONSUL_PORT("SP_CONSUL_PORT", "8500"),
+  SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS", null),
+  SP_JWT_SECRET("JWT_SECRET", null),
+  SP_JWT_SIGNING_MODE("SP_JWT_SIGNING_MODE", null),
+  SP_JWT_PRIVATE_KEY_LOC("SP_JWT_PRIVATE_KEY_LOC", null),
+  SP_JWT_PUBLIC_KEY_LOC("SP_JWT_PUBLIC_KEY_LOC", null),
+  SP_INITIAL_ADMIN_EMAIL("SP_INITIAL_ADMIN_EMAIL", null),
+  SP_INITIAL_ADMIN_PASSWORD("SP_INITIAL_ADMIN_PASSWORD", null),
+  SP_INITIAL_SERVICE_USER("SP_INITIAL_SERVICE_USER", null),
+  SP_INITIAL_SERVICE_USER_SECRET("SP_INITIAL_SERVICE_USER_SECRET", null),
+  SP_SETUP_INSTALL_PIPELINE_ELEMENTS("SP_SETUP_INSTALL_PIPELINE_ELEMENTS", null),
+  SP_EXT_AUTH_MODE("SP_EXT_AUTH_MODE", null),
+  SP_CLIENT_USER("SP_CLIENT_USER", null),
+  SP_CLIENT_SECRET("SP_CLIENT_SECRET", null),
+  SP_ENCRYPTION_PASSCODE("SP_ENCRYPTION_PASSCODE", null),
+  SP_DEBUG("SP_DEBUG", "false"),
+  SP_MAX_WAIT_TIME_AT_SHUTDOWN("SP_MAX_WAIT_TIME_AT_SHUTDOWN", null);
 
   private final String envVariableName;
+  private final String defaultValue;
 
-  Envs(String envVariableName) {
+  Envs(String envVariableName, String defaultValue) {
     this.envVariableName = envVariableName;
+    this.defaultValue = defaultValue;
   }
 
   public boolean exists() {
@@ -82,4 +84,7 @@ public enum Envs {
     return this.exists() ? this.getValue() : defaultValue;
   }
 
+  public String getDefaultValue() {
+    return defaultValue;
+  }
 }
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
new file mode 100644
index 0000000000..0c051ef166
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
@@ -0,0 +1,47 @@
+/*
+ * 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.commons.environment;
+
+import org.apache.streampipes.commons.constants.Envs;
+import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
+
+public class DefaultEnvironment implements Environment {
+
+  @Override
+  public StringEnvironmentVariable getConsulHost() {
+    return new StringEnvironmentVariable(Envs.SP_CONSUL_HOST);
+  }
+
+  @Override
+  public IntEnvironmentVariable getConsulPort() {
+    return new IntEnvironmentVariable(Envs.SP_CONSUL_PORT);
+  }
+
+  @Override
+  public BooleanEnvironmentVariable getSpDebug() {
+    return new BooleanEnvironmentVariable(Envs.SP_DEBUG);
+  }
+
+  @Override
+  public StringEnvironmentVariable getConsulLocation() {
+    return new StringEnvironmentVariable(Envs.SP_CONSUL_LOCATION);
+  }
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
new file mode 100644
index 0000000000..b3a3256d20
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
@@ -0,0 +1,35 @@
+/*
+ * 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.commons.environment;
+
+import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
+
+public interface Environment {
+
+  StringEnvironmentVariable getConsulHost();
+
+  IntEnvironmentVariable getConsulPort();
+
+  BooleanEnvironmentVariable getSpDebug();
+
+  @Deprecated(since = "0.90.0", forRemoval = true)
+  StringEnvironmentVariable getConsulLocation();
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environments.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environments.java
new file mode 100644
index 0000000000..631e72dcc7
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environments.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.commons.environment;
+
+public class Environments {
+
+  public static Environment getEnvironment() {
+    return new DefaultEnvironment();
+  }
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/BooleanEnvironmentVariable.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/BooleanEnvironmentVariable.java
new file mode 100644
index 0000000000..2c7b3e0649
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/BooleanEnvironmentVariable.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.commons.environment.variable;
+
+import org.apache.streampipes.commons.constants.Envs;
+
+public class BooleanEnvironmentVariable extends EnvironmentVariable<Boolean> {
+
+  public BooleanEnvironmentVariable(Envs envVariable) {
+    super(envVariable);
+  }
+
+  @Override
+  public Boolean parse(String value) {
+    return Boolean.parseBoolean(value.toLowerCase());
+  }
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvResolver.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvResolver.java
new file mode 100644
index 0000000000..591f0d0f4b
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvResolver.java
@@ -0,0 +1,24 @@
+/*
+ * 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.commons.environment.variable;
+
+public interface EnvResolver<T> {
+
+  T resolve();
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java
new file mode 100644
index 0000000000..769d5fea95
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/EnvironmentVariable.java
@@ -0,0 +1,62 @@
+/*
+ * 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.commons.environment.variable;
+
+import org.apache.streampipes.commons.constants.CustomEnvs;
+import org.apache.streampipes.commons.constants.Envs;
+
+public abstract class EnvironmentVariable<T> {
+
+  private T defaultValue;
+  private String envVariableName;
+
+  public EnvironmentVariable(String envVariableName,
+                             T defaultValue) {
+    this.envVariableName = envVariableName;
+    this.defaultValue = defaultValue;
+  }
+
+  public EnvironmentVariable(Envs envVariable) {
+    this.envVariableName = envVariable.getEnvVariableName();
+    this.defaultValue = parse(envVariable.getDefaultValue());
+  }
+
+  public T getValue() {
+    return parse(CustomEnvs.getEnv(envVariableName));
+  }
+
+  public boolean exists() {
+    return CustomEnvs.exists(envVariableName);
+  }
+
+  public T getValueOrDefault() {
+    return exists() ? getValue() : defaultValue;
+  }
+
+  public T getValueOrReturn(T defaultValue) {
+    return exists() ? getValue() : defaultValue;
+  }
+
+  public T getValueOrResolve(EnvResolver<T> resolver) {
+    return resolver.resolve();
+  }
+
+  public abstract T parse(String value);
+
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/IntEnvironmentVariable.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/IntEnvironmentVariable.java
new file mode 100644
index 0000000000..6bfb35c2da
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/IntEnvironmentVariable.java
@@ -0,0 +1,32 @@
+/*
+ * 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.commons.environment.variable;
+
+import org.apache.streampipes.commons.constants.Envs;
+
+public class IntEnvironmentVariable extends EnvironmentVariable<Integer> {
+  public IntEnvironmentVariable(Envs envVariable) {
+    super(envVariable);
+  }
+
+  @Override
+  public Integer parse(String value) {
+    return Integer.parseInt(value);
+  }
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/StringEnvironmentVariable.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/StringEnvironmentVariable.java
new file mode 100644
index 0000000000..eaf891b4bc
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/StringEnvironmentVariable.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.commons.environment.variable;
+
+import org.apache.streampipes.commons.constants.Envs;
+
+public class StringEnvironmentVariable extends EnvironmentVariable<String> {
+
+  public StringEnvironmentVariable(Envs envVariable) {
+    super(envVariable);
+  }
+
+  @Override
+  public String parse(String value) {
+    return value;
+  }
+}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java
index 89bc9a7000..5c00462a4a 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java
@@ -17,8 +17,8 @@
  */
 package org.apache.streampipes.extensions.management.config;
 
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
 import org.apache.streampipes.svcdiscovery.api.SpConfig;
-import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig;
 
 import java.io.Serializable;
 
@@ -27,7 +27,7 @@ public class ConfigExtractor implements Serializable {
   private SpConfig config;
 
   private ConfigExtractor(String serviceGroup) {
-    this.config = new ConsulSpConfig(serviceGroup);
+    this.config = SpServiceDiscovery.getSpConfig(serviceGroup);
   }
 
   public static ConfigExtractor from(String serviceGroup) {
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
index 0cea72e1df..16128fd3d6 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
@@ -55,7 +55,10 @@
         </dependency>
         <dependency>
             <groupId>org.apache.iotdb</groupId>
-            <artifactId>iotdb-jdbc</artifactId>
+            <artifactId>iotdb-session</artifactId>
+            <!-- 0.13.3 is the only recommended production ready version for edge side (standalone) deployment -->
+            <!-- Should be worked for IoTDB version >= 0.13.0. -->
+            <version>0.13.3</version>
         </dependency>
         <dependency>
             <groupId>org.eclipse.milo</groupId>
@@ -108,6 +111,18 @@
             <artifactId>bcpkix-jdk15on</artifactId>
         </dependency>
 
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <!-- remove these javax dependencies whenever possible -->
         <dependency>
             <groupId>javax.xml.bind</groupId>
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
index 7e023167fb..df8300f710 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
@@ -18,166 +18,112 @@
 
 package org.apache.streampipes.sinks.databases.jvm.iotdb;
 
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.logging.api.Logger;
 import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventPropertyPrimitive;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.JdbcClient;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.DbDataTypeFactory;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.DbDataTypes;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.ParameterInformation;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.SupportedDbEngines;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.utils.SQLStatementUtils;
+import org.apache.streampipes.model.runtime.field.AbstractField;
 import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
 import org.apache.streampipes.wrapper.runtime.EventSink;
 
-import java.sql.SQLException;
-import java.sql.Statement;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Binary;
 
-public class IotDb extends JdbcClient implements EventSink<IotDbParameters> {
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 
-  private IotDbParameters params;
-  private static Logger log;
+public class IotDb implements EventSink<IotDbParameters> {
 
-  private String timestampField;
+  private static Logger logger;
 
-  private final SupportedDbEngines dbEngine = SupportedDbEngines.IOT_DB;
+  private String timestampFieldId;
+  private String deviceId;
+
+  // IoTDB provides a connection pool (SessionPool) for Native API.
+  // Using the interface, we need to define the pool size.
+  // If we can not get a session connection in 60 seconds, there is a warning log but the program will hang.
+  // If a session has finished an operation, it will be put back to the pool automatically.
+  // If a session connection is broken, the session will be removed automatically and the pool will try to create a
+  // new session and redo the operation.
+  private SessionPool sessionPool;
 
   @Override
-  public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext)
-      throws SpRuntimeException {
-
-    this.params = parameters;
-    log = parameters.getGraph().getLogger(IotDb.class);
-    timestampField = parameters.getTimestampField();
-
-    // tablename is the identifier for the storage group in the IoTDB Adapter (e.g. root.data.table1) in which all
-    // time series are written
-    //TODO: Add better regular expression
-    initializeJdbc(
-        parameters.getGraph().getInputStreams().get(0).getEventSchema(),
-        parameters,
-        dbEngine,
-        log);
+  public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext) {
+    logger = parameters.getGraph().getLogger(IotDb.class);
+
+    deviceId = "root." + parameters.getDatabase() + "." + parameters.getDevice();
+    timestampFieldId = parameters.getTimestampField();
 
+    // In our case, the pool size is set to 2.
+    // One connection is for current requests, and the other is a backup for fast-recovery when connection dies.
+    sessionPool = new SessionPool.Builder().maxSize(2).enableCompression(false).host(parameters.getHost())
+        .port(parameters.getPort()).user(parameters.getUser()).password(parameters.getPassword()).build();
   }
 
   @Override
   public void onEvent(Event event) {
-    try {
-      if (event.getRaw().containsKey("value")) {
-        // Renaming value. Very ugly
-        event.addField("value_1", event.getFieldBySelector("s0::value").getRawValue());
-        event.removeFieldBySelector("s0::value");
-      }
-      save(event);
-    } catch (SpRuntimeException e) {
-      log.error(e.getMessage());
+    if (event == null) {
+      return;
     }
-  }
 
-  @Override
-  public void onDetach() throws SpRuntimeException {
-    closeAll();
-  }
-
-  @Override
-  protected void save(final Event event) throws SpRuntimeException {
-    checkConnected();
-    try {
-      Long timestampValue = event.getFieldBySelector(timestampField).getAsPrimitive().getAsLong();
-      event.removeFieldBySelector(timestampField);
-      Statement statement;
-      statement = connection.createStatement();
-      StringBuilder sb1 = new StringBuilder();
-      StringBuilder sb2 = new StringBuilder();
-      //TODO: Check for SQL-Injection
-      // Timestamp must be in the beginning of the values
-      sb1.append("INSERT INTO ").append(this.params.getDbTable()).append("(timestamp, ");
-      sb2.append(" VALUES (").append(timestampValue).append(", ");
-      for (String s : event.getRaw().keySet()) {
-        sb1.append(s).append(", ");
-        if (event.getFieldByRuntimeName(s).getRawValue() instanceof String) {
-          sb2.append("\"").append(event.getFieldByRuntimeName(s).getRawValue().toString()).append("\", ");
-        } else {
-          sb2.append(event.getFieldByRuntimeName(s).getRawValue().toString()).append(", ");
-        }
-      }
-      sb1.setLength(sb1.length() - 2);
-      sb2.setLength(sb2.length() - 2);
-      sb1.append(") ").append(sb2).append(")");
-      statement.execute(sb1.toString());
-    } catch (SQLException e) {
-      e.printStackTrace();
+    final AbstractField timestampAbstractField = event.getFieldBySelector(timestampFieldId);
+    final Long timestamp = timestampAbstractField.getAsPrimitive().getAsLong();
+    if (timestamp == null) {
+      return;
     }
-  }
 
-  @Override
-  protected void ensureDatabaseExists(String url, String databaseName) throws SpRuntimeException {
-    SQLStatementUtils.checkRegEx(this.params.getDbTable(), "Storage Group name", this.dbDescription);
-    try {
-      Statement statement = connection.createStatement();
-      statement.execute("SET STORAGE GROUP TO " + this.params.getDbTable());
-    } catch (SQLException e) {
-      // Storage group already exists
-      //TODO: Catch other exceptions
+    final Map<String, Object> measurementValuePairs = event.getRaw();
+    // should be at least a timestamp field and a measurement field
+    if (measurementValuePairs.size() <= 1) {
+      return;
     }
-  }
 
-  /**
-   * Needs to be reimplemented since the IoTDB JDBC implementation does not support the methods used in the
-   * JDBC-Client class
-   *
-   * @param url The JDBC url containing the needed information (e.g. "jdbc:iotdb://127.0.0.1:6667/")
-   * @throws SpRuntimeException
-   */
-  @Override
-  protected void ensureTableExists(String url, String databaseName) throws SpRuntimeException {
-    int index = 1;
-    this.statementHandler.putEventParameterMap("timestamp",
-        new ParameterInformation(index++, DbDataTypeFactory.getLong(dbEngine)));
-    for (EventProperty eventProperty : this.tableDescription.getEventSchema().getEventProperties()) {
-      try {
-        if (eventProperty.getRuntimeName().equals(timestampField.substring(4))) {
-          continue;
-        }
-        Statement statement = null;
-        statement = connection.createStatement();
-        // The identifier cannot be called "value"
-        //TODO: Do not simply add a _1 but look instead, if the name is already taken
-        String runtimeName = eventProperty.getRuntimeName();
-        if (eventProperty.getRuntimeName().equals("value")) {
-          runtimeName = "value_1";
-        }
-        DbDataTypes datatype = extractAndAddEventPropertyRuntimeType(eventProperty, index++);
-
-        statement.execute("CREATE TIMESERIES "
-            + params.getDbTable()
-            + "."
-            + runtimeName
-            + " WITH DATATYPE="
-            + datatype.toString()
-            + ", ENCODING=PLAIN");
-      } catch (SQLException e) {
-        // Probably because it already exists
-        //TODO: Add better exception handling
-        e.printStackTrace();
+    final int measurementFieldCount = measurementValuePairs.size() - 1;
+    final List<String> measurements = new ArrayList<>(measurementFieldCount);
+    final List<TSDataType> types = new ArrayList<>(measurementFieldCount);
+    final List<Object> values = new ArrayList<>(measurementFieldCount);
+
+    for (Map.Entry<String, Object> measurementValuePair : measurementValuePairs.entrySet()) {
+      if (timestampAbstractField.getFieldNameIn().equals(measurementValuePair.getKey())) {
+        continue;
+      }
+
+      measurements.add(measurementValuePair.getKey());
+
+      final Object value = measurementValuePair.getValue();
+      if (value instanceof Integer) {
+        types.add(TSDataType.INT32);
+        values.add(value);
+      } else if (value instanceof Long) {
+        types.add(TSDataType.INT64);
+        values.add(value);
+      } else if (value instanceof Float) {
+        types.add(TSDataType.FLOAT);
+        values.add(value);
+      } else if (value instanceof Double) {
+        types.add(TSDataType.DOUBLE);
+        values.add(value);
+      } else if (value instanceof Boolean) {
+        types.add(TSDataType.BOOLEAN);
+        values.add(value);
+      } else {
+        types.add(TSDataType.TEXT);
+        values.add(Binary.valueOf(value.toString()));
       }
     }
-    //tableExists = true;
-  }
 
-  private DbDataTypes extractAndAddEventPropertyRuntimeType(EventProperty eventProperty, int index) {
-    // Supported datatypes can be found here: https://iotdb.apache.org/#/Documents/0.8.0/chap2/sec2
-    DbDataTypes dataType = DbDataTypes.TEXT;
-    if (eventProperty instanceof EventPropertyPrimitive) {
-      dataType = DbDataTypeFactory.getFromUri(((EventPropertyPrimitive) eventProperty).getRuntimeType(),
-          SupportedDbEngines.IOT_DB);
-      this.statementHandler.putEventParameterMap(eventProperty.getRuntimeName(),
-          new ParameterInformation(index, dataType));
+    try {
+      sessionPool.insertRecord(deviceId, timestamp, measurements, types, values);
+    } catch (IoTDBConnectionException | StatementExecutionException e) {
+      logger.error("Failed to save event to IoTDB, because: " + e.getMessage());
+      e.printStackTrace();
     }
+  }
 
-    return dataType;
+  @Override
+  public void onDetach() {
+    sessionPool.close();
   }
 }
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
index 6731397a4f..7bc974bbac 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
@@ -34,52 +34,46 @@ import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDec
 
 public class IotDbController extends StandaloneEventSinkDeclarer<IotDbParameters> {
 
-  private static final String DATABASE_HOST_KEY = "db_host";
-  private static final String DATABASE_PORT_KEY = "db_port";
-  private static final String STORAGE_GROUP_KEY = "db_storage_group";
-  private static final String DATABASE_USER_KEY = "db_user";
-  private static final String DATABASE_PASSWORD_KEY = "db_password";
-  private static final String TIMESTAMPE_MAPPING_KEY = "timestamp_mapping";
+  private static final String HOST_KEY = "db_host";
+  private static final String PORT_KEY = "db_port";
+
+  private static final String DATABASE_KEY = "db_database";
+  private static final String DEVICE_KEY = "db_device";
+
+  private static final String USER_KEY = "db_user";
+  private static final String PASSWORD_KEY = "db_password";
+
+  private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
 
   @Override
   public DataSinkDescription declareModel() {
-    return DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb")
-        .withLocales(Locales.EN)
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-        .category(DataSinkType.DATABASE)
-        .requiredStream(StreamRequirementsBuilder.create()
-            .requiredPropertyWithUnaryMapping(
-                EpRequirements.timestampReq(),
-                Labels.withId(TIMESTAMPE_MAPPING_KEY),
-                PropertyScope.NONE)
-            .build())
-        .requiredTextParameter(Labels.withId(DATABASE_HOST_KEY))
-        .requiredIntegerParameter(Labels.withId(DATABASE_PORT_KEY), 6667)
-        .requiredTextParameter(Labels.withId(STORAGE_GROUP_KEY))
-        .requiredTextParameter(Labels.withId(DATABASE_USER_KEY))
-        .requiredSecret(Labels.withId(DATABASE_PASSWORD_KEY))
-        .build();
+    return DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb").withLocales(Locales.EN)
+        .withAssets(Assets.DOCUMENTATION, Assets.ICON).category(DataSinkType.DATABASE).requiredStream(
+            StreamRequirementsBuilder.create()
+                .requiredPropertyWithUnaryMapping(EpRequirements.timestampReq(), Labels.withId(TIMESTAMP_MAPPING_KEY),
+                    PropertyScope.NONE).build()).requiredTextParameter(Labels.withId(HOST_KEY))
+        .requiredIntegerParameter(Labels.withId(PORT_KEY), 6667).requiredTextParameter(Labels.withId(USER_KEY), "root")
+        .requiredSecret(Labels.withId(PASSWORD_KEY)).requiredTextParameter(Labels.withId(DATABASE_KEY))
+        .requiredTextParameter(Labels.withId(DEVICE_KEY)).build();
   }
 
   @Override
   public ConfiguredEventSink<IotDbParameters> onInvocation(DataSinkInvocation graph,
                                                            DataSinkParameterExtractor extractor) {
 
-    String hostname = extractor.singleValueParameter(DATABASE_HOST_KEY, String.class);
-    Integer port = extractor.singleValueParameter(DATABASE_PORT_KEY, Integer.class);
-    String dbStorageGroup = extractor.singleValueParameter(STORAGE_GROUP_KEY, String.class);
-    String user = extractor.singleValueParameter(DATABASE_USER_KEY, String.class);
-    String password = extractor.secretValue(DATABASE_PASSWORD_KEY);
-    String timestampField = extractor.mappingPropertyValue(TIMESTAMPE_MAPPING_KEY);
+    final String host = extractor.singleValueParameter(HOST_KEY, String.class);
+    final Integer port = extractor.singleValueParameter(PORT_KEY, Integer.class);
+
+    final String database = extractor.singleValueParameter(DATABASE_KEY, String.class);
+    final String device = extractor.singleValueParameter(DEVICE_KEY, String.class);
+
+    final String user = extractor.singleValueParameter(USER_KEY, String.class);
+    final String password = extractor.secretValue(PASSWORD_KEY);
+
+    final String timestampField = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
 
-    IotDbParameters params = new IotDbParameters(graph,
-        hostname,
-        port,
-        dbStorageGroup,
-        user,
-        password,
-        false, // SSL connection not yet implemented for IoT DB
-        timestampField);
+    final IotDbParameters params =
+        new IotDbParameters(graph, host, port, user, password, database, device, timestampField);
 
     return new ConfiguredEventSink<>(params, IotDb::new);
   }
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
index 3150717244..4f41610c8e 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
@@ -19,36 +19,68 @@
 package org.apache.streampipes.sinks.databases.jvm.iotdb;
 
 import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.JdbcConnectionParameters;
+import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
 
-public class IotDbParameters extends JdbcConnectionParameters {
+public class IotDbParameters extends EventSinkBindingParams {
 
-  private String timestampField;
+  private final String host;
+  private final Integer port;
+
+  private final String user;
+  private final String password;
+
+  private final String database;
+  // entity
+  private final String device;
+
+  private final String timestampField;
 
   public IotDbParameters(DataSinkInvocation graph,
-                         String iotDbHost,
-                         Integer iotDbPort,
-                         String dbStorageGroup,
+                         String host,
+                         Integer port,
                          String user,
                          String password,
-                         boolean sslEnabled,
+                         String database,
+                         String device,
                          String timestampField) {
-    super(
-        graph,
-        iotDbHost,
-        iotDbPort,
-        "",
-        user,
-        password,
-        "root." + dbStorageGroup,
-        sslEnabled,
-        null,
-        false
-    );
+    super(graph);
+
+    this.host = host;
+    this.port = port;
+
+    this.user = user;
+    this.password = password;
+
+    this.database = database;
+    this.device = device;
 
     this.timestampField = timestampField;
   }
 
+  public String getHost() {
+    return host;
+  }
+
+  public Integer getPort() {
+    return port;
+  }
+
+  public String getUser() {
+    return user;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+
+  public String getDatabase() {
+    return database;
+  }
+
+  public String getDevice() {
+    return device;
+  }
+
   public String getTimestampField() {
     return timestampField;
   }
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java
index 14887d3bec..206535b6dd 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypeFactory.java
@@ -29,8 +29,6 @@ public class DbDataTypeFactory {
     switch (sqlEngine) {
       case MY_SQL:
         return DbDataTypes.INT;
-      case IOT_DB:
-        return DbDataTypes.INT32;
       case POSTGRESQL:
         return DbDataTypes.INTEGER;
     }
@@ -40,8 +38,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getLong(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
-        return DbDataTypes.INT64;
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.BIGINT;
@@ -52,7 +48,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getFloat(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
       case MY_SQL:
         return DbDataTypes.FLOAT;
       case POSTGRESQL:
@@ -63,8 +58,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getDouble(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
-        return DbDataTypes.DOUBLE;
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.DOUBLE_PRECISION;
@@ -74,8 +67,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getShortString(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
-        return DbDataTypes.TEXT;
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.VAR_CHAR;
@@ -85,7 +76,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getLongString(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.TEXT;
@@ -95,7 +85,6 @@ public class DbDataTypeFactory {
 
   public static DbDataTypes getBoolean(SupportedDbEngines sqlEngine) throws SpRuntimeException {
     switch (sqlEngine) {
-      case IOT_DB:
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.BOOLEAN;
@@ -108,8 +97,6 @@ public class DbDataTypeFactory {
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.TIMESTAMP;
-      case IOT_DB:
-        return getLong(sqlEngine);
     }
     throw new SpRuntimeException("Database engine " + sqlEngine + " does not support timestamps.");
   }
@@ -119,7 +106,6 @@ public class DbDataTypeFactory {
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.DATE;
-      case IOT_DB:
     }
     throw new SpRuntimeException("Database engine " + sqlEngine + " does not support dates.");
   }
@@ -129,7 +115,6 @@ public class DbDataTypeFactory {
       case MY_SQL:
       case POSTGRESQL:
         return DbDataTypes.TIME;
-      case IOT_DB:
     }
     throw new SpRuntimeException("Database engine " + sqlEngine + " does not support time.");
   }
@@ -138,7 +123,6 @@ public class DbDataTypeFactory {
     switch (sqlEngine) {
       case MY_SQL:
         return DbDataTypes.DATETIME;
-      case IOT_DB:
       case POSTGRESQL:
     }
     throw new SpRuntimeException("Database engine " + sqlEngine + " does not support datetime.");
@@ -189,17 +173,14 @@ public class DbDataTypeFactory {
   public static Datatypes getDataType(DbDataTypes dbDataType) throws SpRuntimeException {
 
     switch (dbDataType) {
-      case BOOL:
       case BOOLEAN:
         return Datatypes.Boolean;
       case TEXT:
       case VAR_CHAR:
-      case STRING:
       case TIMESTAMP:
       case DATE:
       case TIME:
         return Datatypes.String;
-      case DOUBLE:
       case DOUBLE_PRECISION:
         return Datatypes.Double;
       case FLOAT:
@@ -208,10 +189,8 @@ public class DbDataTypeFactory {
         return Datatypes.Float;
       case BIGINT:
       case TINYINT:
-      case INT64:
         return Datatypes.Long;
       case INT:
-      case INT32:
       case INTEGER:
         return Datatypes.Integer;
       default:
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java
index 5691a31f43..25b0ecf6a3 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/DbDataTypes.java
@@ -21,19 +21,14 @@ package org.apache.streampipes.sinks.databases.jvm.jdbcclient.model;
 
 public enum DbDataTypes {
 
-  BOOL("BOOL"),
   BOOLEAN("BOOLEAN"),
   TEXT("TEXT"),
   VAR_CHAR("VARCHAR255"),
-  STRING("STRING"),
-  DOUBLE("DOUBLE"),
   DOUBLE_PRECISION("DOUBLE PRECISION"),
   FLOAT("FLOAT"),
   REAL("REAL"),
   BIGINT("BIGINT"),
   TINYINT("TINYINT"),
-  INT64("INT64"),
-  INT32("INT32"),
   INT("INT"),
   INTEGER("INTEGER"),
   TIMESTAMP("TIMESTAMP"),
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java
index 21d6176537..55a0a191d3 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/SupportedDbEngines.java
@@ -20,7 +20,6 @@ package org.apache.streampipes.sinks.databases.jvm.jdbcclient.model;
 
 public enum SupportedDbEngines {
   MY_SQL("mysql", "com.mysql.cj.jdbc.Driver", ".*"),
-  IOT_DB("iotdb", "org.apache.iotdb.jdbc.IoTDBDriver", ".*"),
   POSTGRESQL("postgresql", "org.postgresql.Driver", "^[a-zA-Z_][a-zA-Z0-9_]*$");
 
   private final String urlName;
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md
index d02b0d429d..eacab0a2e0 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/documentation.md
@@ -28,6 +28,10 @@
 
 Stores events in a IoTDB database.
 
+Events will be stored in timeseries `root.${Database Name}.${Device (Entity) Name}.${Event Key}`.
+
+Please reference to [https://iotdb.apache.org/](https://iotdb.apache.org/) for more information.
+
 ***
 
 ## Required input
@@ -46,11 +50,6 @@ The hostname of the IoTDB instance.
 
 The port of the IoTDB instance (default 6667).
 
-### Storage Group Name
-
-The name of the storage group where events will be stored (will be created if it does not exist).
-For each element of the stream a new time series will be created.
-
 ### Username
 
 The username for the IoTDB Server.
@@ -59,6 +58,28 @@ The username for the IoTDB Server.
 
 The password for the IoTDB Server.
 
+### **Database Name**
+
+The name of the database where events will be stored (will be created if it does not exist).
+
+A database is a group of devices (entities). Users can create any prefix path as a database.
+
+### **Device (Entity) Name**
+
+The name of the device (entity) where events will be stored.
+
+A device (also called entity) is an equipped with measurements in real scenarios. In IoTDB, all measurements should have
+their corresponding devices.
+
+### **Measurements**
+
+All keys of fields in an event will be automatically converted to suffixes of timeseries.
+
+A measurement is information measured by detection equipment in an actual scene and can transform the sensed information
+into an electrical signal or other desired form of information output and send it to IoTDB.
+
+In IoTDB, all data and paths stored are organized in units of measurements.
+
 ## Output
 
 (not applicable for data sinks)
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en
index 107b64b02f..cb16982637 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.iotdb/strings.en
@@ -15,8 +15,8 @@
 # limitations under the License.
 #
 
-org.apache.streampipes.sinks.databases.jvm.iotdb.title=IoTDB
-org.apache.streampipes.sinks.databases.jvm.iotdb.description=Stores events in a IoTDB database.
+org.apache.streampipes.sinks.databases.jvm.iotdb.title=Apache IoTDB
+org.apache.streampipes.sinks.databases.jvm.iotdb.description=Stores events in a IoTDB database. Events will be stored in timeseries root.${Database Name}.${Device (Entity) Name}.${Event Key}.
 
 db_host.title=Hostname
 db_host.description=The hostname of the IoTDB instance
@@ -24,14 +24,17 @@ db_host.description=The hostname of the IoTDB instance
 db_port.title=Port
 db_port.description=The port of the IoTDB instance (default 6667)
 
-db_storage_group.title=Storage Group Name
-db_storage_group.description=The name of the storage group where events will be stored
+db_database.title=Database Name
+db_database.description=The name of the database where events will be stored. A database is a group of devices (entities). Users can create any prefix path as a database.
+
+db_device.title=Device (Entity) Name
+db_device.description=The name of the device (entity) where events will be stored. A device (also called entity) is an equipped with measurements in real scenarios. In IoTDB, all measurements should have their corresponding devices.
 
 db_user.title=Username
-db_user.description=The username for the IoTDB Server
+db_user.description=The username for the IoTDB Server (default "root")
 
 db_password.title=Password
-db_password.description=The password for the IoTDB Server
+db_password.description=The password for the IoTDB Server (default "root")
 
 timestamp_mapping.title=Timestamp
 timestamp_mapping.description=The timestamp of the event
diff --git a/streampipes-sdk/pom.xml b/streampipes-sdk/pom.xml
index 0f7b7aaf31..889aa9e5df 100644
--- a/streampipes-sdk/pom.xml
+++ b/streampipes-sdk/pom.xml
@@ -47,6 +47,10 @@
         </dependency>
 
         <!-- External dependencies -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.github.drapostolos</groupId>
             <artifactId>type-parser</artifactId>
@@ -71,4 +75,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
\ No newline at end of file
+</project>
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java
index 6c7583df10..4b859f3c67 100644
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java
+++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java
@@ -21,8 +21,6 @@ import java.util.Map;
 
 public interface ISpKvManagement {
 
-  <T> T getValueForRoute(String route, Class<T> type);
-
   Map<String, String> getKeyValue(String route);
 
   void updateConfig(String key, String entry, boolean password);
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
index c3dcc04156..803f4e43b9 100644
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
+++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
@@ -18,6 +18,8 @@
 
 package org.apache.streampipes.svcdiscovery.api.model;
 
+import java.util.Objects;
+
 public class ConfigItem {
 
   private String key;
@@ -117,4 +119,26 @@ public class ConfigItem {
   public void setConfigurationScope(ConfigurationScope configurationScope) {
     this.configurationScope = configurationScope;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ConfigItem that = (ConfigItem) o;
+    return isPassword == that.isPassword
+        && Objects.equals(key, that.key)
+        && Objects.equals(description, that.description)
+        && Objects.equals(value, that.value)
+        && Objects.equals(valueType, that.valueType)
+        && configurationScope == that.configurationScope;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(key, description, value, valueType, configurationScope, isPassword);
+  }
 }
diff --git a/streampipes-service-discovery-consul/pom.xml b/streampipes-service-discovery-consul/pom.xml
index b1cedc09ef..a10cad9efb 100644
--- a/streampipes-service-discovery-consul/pom.xml
+++ b/streampipes-service-discovery-consul/pom.xml
@@ -46,8 +46,13 @@
 
         <!-- third-party dependencies -->
         <dependency>
-            <groupId>com.orbitz.consul</groupId>
-            <artifactId>consul-client</artifactId>
+            <groupId>com.ecwid.consul</groupId>
+            <artifactId>consul-api</artifactId>
+            <version>1.4.5</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
         </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
index aff0f18c3b..d89b387ac3 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
@@ -17,17 +17,19 @@
  */
 package org.apache.streampipes.svcdiscovery.consul;
 
-import com.orbitz.consul.Consul;
+import org.apache.streampipes.commons.environment.Environment;
+
+import com.ecwid.consul.v1.ConsulClient;
 
 public abstract class AbstractConsulService {
 
-  protected ConsulHealthServiceManager consul;
+  private final Environment environment;
 
-  public AbstractConsulService() {
-    this.consul = ConsulHealthServiceManager.INSTANCE;
+  public AbstractConsulService(Environment environment) {
+    this.environment = environment;
   }
 
-  public Consul consulInstance() {
-    return this.consul.consulInstance();
+  public ConsulClient consulInstance() {
+    return ConsulProvider.INSTANCE.getConsulInstance(environment);
   }
 }
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
index d8d2d642b6..cb1710483e 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
@@ -17,90 +17,64 @@
  */
 package org.apache.streampipes.svcdiscovery.consul;
 
-import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
+import org.apache.streampipes.commons.environment.Environment;
 
-import com.orbitz.consul.Consul;
-import com.orbitz.consul.HealthClient;
-import com.orbitz.consul.cache.ServiceHealthCache;
-import com.orbitz.consul.model.health.ServiceHealth;
-import com.orbitz.consul.option.QueryOptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.ecwid.consul.v1.QueryParams;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.health.HealthServicesRequest;
+import com.ecwid.consul.v1.health.model.Check;
+import com.ecwid.consul.v1.health.model.HealthService;
 
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-public enum ConsulHealthServiceManager {
+public class ConsulHealthServiceManager extends AbstractConsulService {
 
-  INSTANCE;
-
-  private static final Logger LOG = LoggerFactory.getLogger(ConsulHealthServiceManager.class);
   private static final int MAX_RETRIES = 3;
-  private final Consul client;
-  private final Map<String, ServiceHealthCache> serviceHealthCaches;
 
-  ConsulHealthServiceManager() {
-    serviceHealthCaches = new HashMap<>();
-    client = new ConsulProvider().consulInstance();
-    initializeAll();
+  public ConsulHealthServiceManager(Environment environment) {
+    super(environment);
+    initializeCache();
   }
 
-  public void initializeAll() {
-    initialize(DefaultSpServiceGroups.CORE);
-    initialize(DefaultSpServiceGroups.EXT);
-  }
+  public void initializeCache() {
 
-  public void initialize(String serviceGroup) {
-    HealthClient healthClient = client.healthClient();
-    ServiceHealthCache svHealth = ServiceHealthCache.newCache(healthClient, serviceGroup, false, 9, QueryOptions.BLANK);
-    svHealth.start();
-    try {
-      svHealth.awaitInitialized(5, TimeUnit.SECONDS);
-    } catch (InterruptedException e) {
-      e.printStackTrace();
-    }
-    serviceHealthCaches.put(serviceGroup, svHealth);
   }
 
   public List<String> getServiceEndpoints(String serviceGroup,
                                           boolean restrictToHealthy,
                                           List<String> filterByTags) {
-    List<ServiceHealth> activeServices = findService(serviceGroup, 0);
+    List<HealthService> activeServices = findService(serviceGroup, 0);
 
     return activeServices
         .stream()
         .filter(service -> allFiltersSupported(service, filterByTags))
         .filter(service -> !restrictToHealthy
-            || service.getChecks().stream().allMatch(check -> check.getStatus().equals("passing")))
+            || service.getChecks().stream().allMatch(check -> check.getStatus() == Check.CheckStatus.PASSING))
         .map(this::makeServiceUrl)
         .collect(Collectors.toList());
   }
 
-  private String makeServiceUrl(ServiceHealth service) {
+  private String makeServiceUrl(HealthService service) {
     return service.getService().getAddress() + ":" + service.getService().getPort();
   }
 
-  private boolean allFiltersSupported(ServiceHealth service,
+  private boolean allFiltersSupported(HealthService service,
                                       List<String> filterByTags) {
-    return service.getService().getTags().containsAll(filterByTags);
+    return new HashSet<>(service.getService().getTags()).containsAll(filterByTags);
   }
 
-  private List<ServiceHealth> findService(String serviceGroup, int retryCount) {
-
-    if (serviceHealthCaches.containsKey(serviceGroup)
-        && serviceHealthCaches.get(serviceGroup).getMap() != null) {
-      ServiceHealthCache cache = serviceHealthCaches.get(serviceGroup);
-      return cache
-          .getMap()
-          .values()
-          .stream()
-          .filter((value) -> value.getService().getService().equals(serviceGroup))
-          .collect(Collectors.toList());
-    } else {
+  private List<HealthService> findService(String serviceGroup, int retryCount) {
+    HealthServicesRequest request = HealthServicesRequest.newBuilder()
+        .setPassing(true)
+        .setQueryParams(QueryParams.DEFAULT)
+        .build();
+    Response<List<HealthService>> healthyServicesResp = consulInstance().getHealthServices(serviceGroup, request);
+    var healthyServices = healthyServicesResp.getValue();
+    if (healthyServices.size() == 0) {
       if (retryCount < MAX_RETRIES) {
         try {
           retryCount++;
@@ -113,10 +87,10 @@ public enum ConsulHealthServiceManager {
       } else {
         return Collections.emptyList();
       }
+    } else {
+      return healthyServices;
     }
   }
 
-  public Consul consulInstance() {
-    return this.client;
-  }
+
 }
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
index 69654eaba2..cd7282d324 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
@@ -18,9 +18,9 @@
 package org.apache.streampipes.svcdiscovery.consul;
 
 import org.apache.streampipes.commons.constants.DefaultEnvValues;
-import org.apache.streampipes.commons.constants.Envs;
+import org.apache.streampipes.commons.environment.Environment;
 
-import com.orbitz.consul.Consul;
+import com.ecwid.consul.v1.ConsulClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,47 +29,52 @@ import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.concurrent.TimeUnit;
 
-public class ConsulProvider {
+public enum ConsulProvider {
+
+  INSTANCE;
 
   private static final Logger LOG = LoggerFactory.getLogger(ConsulProvider.class);
   private static final int CHECK_INTERVAL = 1;
 
-  private final String consulHost;
-  private final String consulUrlString;
-  private final int consulPort;
+  private ConsulClient consulClient;
+  private boolean initialized = false;
 
-  public ConsulProvider() {
-    this.consulHost = getConsulHost();
-    this.consulPort = getConsulPort();
-    this.consulUrlString = makeConsulUrl();
+  ConsulProvider() {
   }
 
-  public Consul consulInstance() {
-    boolean connected;
+  public ConsulClient getConsulInstance(Environment environment) {
+    if (!initialized) {
+      createConsulInstance(environment);
+    }
 
-    do {
-      LOG.info("Checking if consul is available on host {} and port {}", consulHost, consulPort);
-      connected = checkConsulAvailable();
+    return consulClient;
+  }
 
-      if (!connected) {
-        LOG.info("Retrying in {} second", CHECK_INTERVAL);
-        try {
-          TimeUnit.SECONDS.sleep(CHECK_INTERVAL);
-        } catch (InterruptedException e) {
-          e.printStackTrace();
-        }
-      }
-    } while (!connected);
+  private void createConsulInstance(Environment environment) {
+    var consulHost = getConsulHost(environment);
+    var consulPort = getConsulPort(environment);
+    var connected = false;
 
-    LOG.info("Successfully connected to Consul");
-    return Consul.builder().withUrl(consulUrlString).build();
-  }
+    LOG.info("Checking if consul is available on host {} and port {}", consulHost, consulPort);
+    connected = checkConsulAvailable(consulHost, consulPort);
 
-  public String makeConsulUrl() {
-    return "http://" + consulHost + ":" + consulPort;
+    if (connected) {
+      LOG.info("Successfully connected to Consul on host {}", consulHost);
+      this.consulClient = new ConsulClient(consulHost, consulPort);
+      this.initialized = true;
+    } else {
+      LOG.info("Retrying in {} second", CHECK_INTERVAL);
+      try {
+        TimeUnit.SECONDS.sleep(CHECK_INTERVAL);
+        createConsulInstance(environment);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
   }
 
-  private boolean checkConsulAvailable() {
+  private boolean checkConsulAvailable(String consulHost,
+                                       int consulPort) {
     try {
       InetSocketAddress sa = new InetSocketAddress(consulHost, consulPort);
       Socket ss = new Socket();
@@ -83,18 +88,18 @@ public class ConsulProvider {
     }
   }
 
-  private int getConsulPort() {
-    return Envs.SP_CONSUL_PORT.getValueAsIntOrDefault(DefaultEnvValues.CONSUL_PORT_DEFAULT);
+  private int getConsulPort(Environment environment) {
+    return environment.getConsulPort().getValueOrDefault();
   }
 
-  private String getConsulHost() {
-    if (Envs.SP_CONSUL_LOCATION.exists()) {
-      return Envs.SP_CONSUL_LOCATION.getValue();
+  private String getConsulHost(Environment environment) {
+    if (environment.getConsulLocation().exists()) {
+      return environment.getConsulLocation().getValue();
     } else {
-      if (Envs.SP_DEBUG.getValueAsBooleanOrDefault(false)) {
+      if (environment.getSpDebug().getValueOrReturn(false)) {
         return DefaultEnvValues.CONSUL_HOST_LOCAL;
       } else {
-        return Envs.SP_CONSUL_HOST.getValueOrDefault(DefaultEnvValues.CONSUL_HOST_DEFAULT);
+        return environment.getConsulHost().getValueOrDefault();
       }
     }
   }
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java
index 9f3f6ddeca..73770eeead 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java
@@ -18,35 +18,30 @@
 
 package org.apache.streampipes.svcdiscovery.consul;
 
+import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
 import org.apache.streampipes.svcdiscovery.api.SpConfig;
 import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
 import org.apache.streampipes.svcdiscovery.api.model.ConfigItemUtils;
 import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope;
 
+import com.ecwid.consul.v1.ConsulClient;
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.orbitz.consul.Consul;
-import com.orbitz.consul.KeyValueClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Map;
-import java.util.Optional;
-
 public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
 
   public static final String SERVICE_ROUTE_PREFIX = "sp/v1/";
   private static final Logger LOG = LoggerFactory.getLogger(ConsulSpConfig.class);
   private static final String SLASH = "/";
   private final String serviceName;
-  private final KeyValueClient kvClient;
-
-  // TODO Implement mechanism to update the client when some configuration parameters change in Consul
-  private Map<String, Object> configProps;
+  private final ConsulClient kvClient;
 
-  public ConsulSpConfig(String serviceName) {
-    Consul consul = consulInstance();
-    this.kvClient = consul.keyValueClient();
+  public ConsulSpConfig(String serviceName,
+                        Environment environment) {
+    super(environment);
+    this.kvClient = consulInstance();
     this.serviceName = serviceName;
   }
 
@@ -82,9 +77,9 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
 
   @Override
   public void registerObject(String key, Object defaultValue, String description) {
-    Optional<String> i = kvClient.getValueAsString(addSn(key));
-    if (!i.isPresent()) {
-      kvClient.putValue(addSn(key), toJson(defaultValue));
+    var i = kvClient.getKVValue(addSn(key));
+    if (i.getValue() == null) {
+      kvClient.setKVValue(addSn(key), toJson(defaultValue));
     }
   }
 
@@ -96,28 +91,28 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
   @Override
   public void register(ConfigItem configItem) {
     String key = addSn(configItem.getKey());
-    Optional<String> i = kvClient.getValueAsString(key);
+    var i = kvClient.getKVValue(key);
 
-    if (!i.isPresent()) {
+    if (i.getValue() == null) {
       // Set the value of environment variable as default
       String envVariable = System.getenv(configItem.getKey());
       if (envVariable != null) {
         configItem.setValue(envVariable);
-        kvClient.putValue(key, toJson(configItem));
+        kvClient.setKVValue(key, toJson(configItem));
       } else {
-        kvClient.putValue(key, toJson(configItem));
+        kvClient.setKVValue(key, toJson(configItem));
       }
     }
   }
 
-  private void register(String key, String defaultValue, String valueType, String description,
-                        ConfigurationScope configurationScope, boolean isPassword) {
+  private void register(String key,
+                        String defaultValue,
+                        String valueType,
+                        String description,
+                        ConfigurationScope configurationScope,
+                        boolean isPassword) {
     ConfigItem configItem = ConfigItem.from(key, defaultValue, description, valueType, configurationScope, isPassword);
     register(configItem);
-
-    if (configProps != null) {
-      configProps.put(key, getString(key));
-    }
   }
 
   @Override
@@ -142,10 +137,10 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
 
   @Override
   public <T> T getObject(String key, Class<T> clazz, T defaultValue) {
-    Optional<String> os = kvClient.getValueAsString(addSn(key));
-    if (os.isPresent()) {
+    var os = kvClient.getKVValue(addSn(key));
+    if (os.getValue() != null) {
       try {
-        return JacksonSerializer.getObjectMapper().readValue(os.get(), clazz);
+        return JacksonSerializer.getObjectMapper().readValue(os.getValue().getDecodedValue(), clazz);
       } catch (JsonProcessingException e) {
         LOG.info("Could not deserialize object", e);
         return defaultValue;
@@ -157,9 +152,9 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
 
   @Override
   public ConfigItem getConfigItem(String key) {
-    Optional<String> os = kvClient.getValueAsString(addSn(key));
+    var os = kvClient.getKVValue(addSn(key)).getValue();
 
-    return fromJson(os.get());
+    return fromJson(os.getDecodedValue());
   }
 
   @Override
@@ -179,12 +174,12 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
 
   @Override
   public void setString(String key, String value) {
-    kvClient.putValue(addSn(key), value);
+    kvClient.setKVValue(addSn(key), value);
   }
 
   @Override
   public void setObject(String key, Object value) {
-    kvClient.putValue(addSn(key), toJson(value));
+    kvClient.setKVValue(addSn(key), toJson(value));
   }
 
   private String addSn(String key) {
@@ -202,7 +197,9 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
     }
   }
 
-  private ConfigItem prepareConfigItem(String valueType, String description, ConfigurationScope configurationScope,
+  private ConfigItem prepareConfigItem(String valueType,
+                                       String description,
+                                       ConfigurationScope configurationScope,
                                        boolean password) {
     ConfigItem configItem = new ConfigItem();
     configItem.setValueType(valueType);
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
index 0138114a52..31247feca6 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
@@ -17,15 +17,11 @@
  */
 package org.apache.streampipes.svcdiscovery.consul;
 
-import org.apache.streampipes.serializers.json.JacksonSerializer;
+import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.svcdiscovery.api.ISpKvManagement;
-import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.orbitz.consul.Consul;
-import com.orbitz.consul.KeyValueClient;
-import com.orbitz.consul.model.ConsulResponse;
-import com.orbitz.consul.model.kv.Value;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.kv.model.GetValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -33,47 +29,28 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public class SpConsulKvManagement extends ConsulProvider implements ISpKvManagement {
+public class SpConsulKvManagement extends AbstractConsulService implements ISpKvManagement {
 
   private static final Logger LOG = LoggerFactory.getLogger(SpConsulKvManagement.class);
 
   private static final String CONSUL_NAMESPACE = "/sp/v1/";
 
-  public <T> T getValueForRoute(String route, Class<T> type) {
-    try {
-      String entry = getKeyValue(route)
-          .values()
-          .stream()
-          .findFirst()
-          .orElse(null);
-
-      if (type.equals(Integer.class)) {
-        return (T) Integer.valueOf(JacksonSerializer.getObjectMapper().readValue(entry, ConfigItem.class).getValue());
-      } else if (type.equals(Boolean.class)) {
-        return (T) Boolean.valueOf(JacksonSerializer.getObjectMapper().readValue(entry, ConfigItem.class).getValue());
-      } else {
-        return type.cast(JacksonSerializer.getObjectMapper().readValue(entry, ConfigItem.class).getValue());
-      }
-    } catch (JsonProcessingException e) {
-      e.printStackTrace();
-    }
-    throw new IllegalArgumentException("Cannot get entry from Consul");
+  public SpConsulKvManagement(Environment environment) {
+    super(environment);
   }
 
   public Map<String, String> getKeyValue(String route) {
-    Consul consul = consulInstance();
-    KeyValueClient keyValueClient = consul.keyValueClient();
+    var consul = consulInstance();
 
     Map<String, String> keyValues = new HashMap<>();
+    Response<List<GetValue>> consulResponseWithValues = consul.getKVValues(route);
 
-    ConsulResponse<List<Value>> consulResponseWithValues = keyValueClient.getConsulResponseWithValues(route);
-
-    if (consulResponseWithValues.getResponse() != null) {
-      for (Value value : consulResponseWithValues.getResponse()) {
+    if (consulResponseWithValues.getValue() != null) {
+      for (GetValue value : consulResponseWithValues.getValue()) {
         String key = value.getKey();
         String v = "";
-        if (value.getValueAsString().isPresent()) {
-          v = value.getValueAsString().get();
+        if (value.getValue() != null) {
+          v = value.getDecodedValue();
         }
         keyValues.put(key, v);
       }
@@ -82,16 +59,16 @@ public class SpConsulKvManagement extends ConsulProvider implements ISpKvManagem
   }
 
   public void updateConfig(String key, String entry, boolean password) {
-    Consul consul = consulInstance();
+    var consul = consulInstance();
     if (!password) {
       LOG.info("Updated config - key:" + key + " value: " + entry);
-      consul.keyValueClient().putValue(key, entry);
+      consul.setKVValue(key, entry);
     }
   }
 
   public void deleteConfig(String key) {
-    Consul consul = consulInstance();
+    var consul = consulInstance();
     LOG.info("Delete config: {}", key);
-    consul.keyValueClient().deleteKeys(CONSUL_NAMESPACE + key);
+    consul.deleteKVValue(CONSUL_NAMESPACE + key);
   }
 }
diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
index 8156e2545e..a00acf5752 100644
--- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
+++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
@@ -17,6 +17,7 @@
  */
 package org.apache.streampipes.svcdiscovery.consul;
 
+import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
 import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
 import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
@@ -24,13 +25,10 @@ import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationReques
 import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag;
 import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix;
 
-import com.orbitz.consul.AgentClient;
-import com.orbitz.consul.Consul;
-import com.orbitz.consul.model.agent.ImmutableRegCheck;
-import com.orbitz.consul.model.agent.ImmutableRegistration;
-import com.orbitz.consul.model.agent.Registration;
-import com.orbitz.consul.model.health.HealthCheck;
-import com.orbitz.consul.model.health.Service;
+import com.ecwid.consul.v1.Response;
+import com.ecwid.consul.v1.agent.model.Check;
+import com.ecwid.consul.v1.agent.model.NewService;
+import com.ecwid.consul.v1.agent.model.Service;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,9 +47,16 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I
   private static final String COLON = ":";
   private static final String HEALTH_CHECK_INTERVAL = "10s";
 
+  private ConsulHealthServiceManager healthServiceManager;
+
+  public SpConsulServiceDiscovery(Environment environment) {
+    super(environment);
+    this.healthServiceManager = new ConsulHealthServiceManager(environment);
+  }
+
   @Override
   public void registerService(SpServiceRegistrationRequest req) {
-    consulInstance().agentClient().register((createRegistrationBody(req)));
+    consulInstance().agentServiceRegister(createRegistrationBody(req));
     LOG.info("Successfully registered service at Consul: " + req.getSvcId());
   }
 
@@ -75,30 +80,32 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I
 
   @Override
   public List<String> getServiceEndpoints(String svcGroup, boolean restrictToHealthy, List<String> filterByTags) {
-    return ConsulHealthServiceManager.INSTANCE.getServiceEndpoints(svcGroup, restrictToHealthy, filterByTags);
+    return healthServiceManager.getServiceEndpoints(svcGroup, restrictToHealthy, filterByTags);
   }
 
   @Override
   public Map<String, String> getExtensionsServiceGroups() {
     LOG.info("Load pipeline element service status");
-    Consul consul = consulInstance();
-    AgentClient agent = consul.agentClient();
+    var consul = consulInstance();
 
-    Map<String, Service> services = consul.agentClient().getServices();
-    Map<String, HealthCheck> checks = agent.getChecks();
+    Response<Map<String, Service>> servicesResp = consul.getAgentServices();
+    Response<Map<String, Check>> checksResp = consul.getAgentChecks();
 
     Map<String, String> peSvcs = new HashMap<>();
-
-    for (Map.Entry<String, Service> entry : services.entrySet()) {
-      if (hasExtensionsTag(entry.getValue().getTags())) {
-        String serviceId = entry.getValue().getId();
-        String serviceStatus = "critical";
-        if (checks.containsKey("service:" + entry.getKey())) {
-          serviceStatus = checks.get("service:" + entry.getKey()).getStatus();
+    if (servicesResp.getValue() != null) {
+      var services = servicesResp.getValue();
+      var checks = checksResp.getValue();
+      for (Map.Entry<String, Service> entry : services.entrySet()) {
+        if (hasExtensionsTag(entry.getValue().getTags())) {
+          String serviceId = entry.getValue().getId();
+          String serviceStatus = "critical";
+          if (checks.containsKey("service:" + entry.getKey())) {
+            serviceStatus = checks.get("service:" + entry.getKey()).getStatus().name();
+          }
+          LOG.info("Service id: " + serviceId + " service status: " + serviceStatus);
+          String serviceGroup = extractServiceGroup(entry.getValue().getTags());
+          peSvcs.put(serviceGroup, serviceStatus);
         }
-        LOG.info("Service id: " + serviceId + " service status: " + serviceStatus);
-        String serviceGroup = extractServiceGroup(entry.getValue().getTags());
-        peSvcs.put(serviceGroup, serviceStatus);
       }
     }
     return peSvcs;
@@ -117,25 +124,33 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I
 
   @Override
   public void deregisterService(String svcId) {
-    Consul consul = consulInstance();
+    var consul = consulInstance();
     LOG.info("Deregister service: " + svcId);
-    consul.agentClient().deregister(svcId);
+    consul.agentServiceDeregister(svcId);
+  }
+
+  private NewService createRegistrationBody(SpServiceRegistrationRequest req) {
+    var service = new NewService();
+    service.setId(req.getSvcId());
+    service.setName(req.getSvcGroup());
+    service.setPort(req.getPort());
+    service.setAddress(HTTP_PROTOCOL + req.getHost());
+    service.setCheck(createServiceCheck(req));
+
+    service.setTags(asString(req.getTags()));
+    service.setEnableTagOverride(true);
+
+    return service;
   }
 
-  private Registration createRegistrationBody(SpServiceRegistrationRequest req) {
-    return ImmutableRegistration.builder()
-        .id(req.getSvcId())
-        .name(req.getSvcGroup())
-        .port(req.getPort())
-        .address(HTTP_PROTOCOL + req.getHost())
-        .check(ImmutableRegCheck.builder()
-            .http(HTTP_PROTOCOL + req.getHost() + COLON + req.getPort() + req.getHealthCheckPath())
-            .interval(HEALTH_CHECK_INTERVAL)
-            .deregisterCriticalServiceAfter("120s")
-            .status("passing")
-            .build())
-        .tags(asString(req.getTags()))
-        .enableTagOverride(true)
-        .build();
+  private NewService.Check createServiceCheck(SpServiceRegistrationRequest req) {
+    var serviceCheck = new NewService.Check();
+
+    serviceCheck.setHttp(HTTP_PROTOCOL + req.getHost() + COLON + req.getPort() + req.getHealthCheckPath());
+    serviceCheck.setInterval(HEALTH_CHECK_INTERVAL);
+    serviceCheck.setDeregisterCriticalServiceAfter("120s");
+    serviceCheck.setStatus("passing");
+
+    return serviceCheck;
   }
 }
diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java
index 8f836b0777..cead1334b0 100644
--- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java
+++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java
@@ -17,6 +17,8 @@
  */
 package org.apache.streampipes.svcdiscovery;
 
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.svcdiscovery.api.ISpKvManagement;
 import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
 import org.apache.streampipes.svcdiscovery.api.SpConfig;
@@ -27,15 +29,28 @@ import org.apache.streampipes.svcdiscovery.consul.SpConsulServiceDiscovery;
 public class SpServiceDiscovery {
 
   public static ISpServiceDiscovery getServiceDiscovery() {
-    return new SpConsulServiceDiscovery();
+    return new SpConsulServiceDiscovery(Environments.getEnvironment());
+  }
+
+  public static ISpServiceDiscovery getServiceDiscovery(Environment environment) {
+    return new SpConsulServiceDiscovery(environment);
   }
 
   public static ISpKvManagement getKeyValueStore() {
-    return new SpConsulKvManagement();
+    return new SpConsulKvManagement(Environments.getEnvironment());
+  }
+
+  public static ISpKvManagement getKeyValueStore(Environment environment) {
+    return new SpConsulKvManagement(environment);
   }
 
   public static SpConfig getSpConfig(String serviceGroup) {
-    return new ConsulSpConfig(serviceGroup);
+    return getSpConfig(serviceGroup, Environments.getEnvironment());
+  }
+
+  public static SpConfig getSpConfig(String serviceGroup,
+                                     Environment environment) {
+    return new ConsulSpConfig(serviceGroup, environment);
   }
 
 }
diff --git a/streampipes-storage-couchdb/pom.xml b/streampipes-storage-couchdb/pom.xml
index 4a0c28cdbd..f0346cd602 100644
--- a/streampipes-storage-couchdb/pom.xml
+++ b/streampipes-storage-couchdb/pom.xml
@@ -56,6 +56,10 @@
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.lightcouch</groupId>
             <artifactId>lightcouch</artifactId>


Test patch:
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java
new file mode 100644
index 0000000000..1bf3092529
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.sinks.databases.jvm.iotdb;
+
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+
+public class TestIotDbParameters {
+
+  @Test
+  public void testIotDbParameters() {
+    final String host = "localhost";
+    final Integer port = 6667;
+    final String user = "user";
+    final String password = "password";
+    final String database = "database";
+    final String device = "device";
+    final String timestampField = "timestampField";
+
+    final IotDbParameters parameters =
+        new IotDbParameters(mock(DataSinkInvocation.class), host, port, user, password, database, device,
+            timestampField);
+
+    Assert.assertEquals(host, parameters.getHost());
+    Assert.assertEquals(port, parameters.getPort());
+    Assert.assertEquals(user, parameters.getUser());
+    Assert.assertEquals(password, parameters.getPassword());
+    Assert.assertEquals(database, parameters.getDatabase());
+    Assert.assertEquals(device, parameters.getDevice());
+    Assert.assertEquals(timestampField, parameters.getTimestampField());
+  }
+}
\ No newline at end of file
diff --git a/streampipes-integration-tests/pom.xml b/streampipes-integration-tests/pom.xml
index d576bb5088..c4a2588ec0 100644
--- a/streampipes-integration-tests/pom.xml
+++ b/streampipes-integration-tests/pom.xml
@@ -56,12 +56,23 @@
             <classifier>embed</classifier>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-service-discovery</artifactId>
+            <version>0.91.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>
             <version>1.17.4</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>consul</artifactId>
+            <version>1.17.6</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/AbstractConsulTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/AbstractConsulTest.java
new file mode 100644
index 0000000000..76f3444f04
--- /dev/null
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/AbstractConsulTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.integration.svcdiscovery;
+
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+
+import org.mockito.Mockito;
+import org.testcontainers.consul.ConsulContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import static org.mockito.Mockito.mock;
+
+public class AbstractConsulTest {
+
+  static final ConsulContainer CONSUL_CONTAINER;
+
+  static {
+    CONSUL_CONTAINER = new ConsulContainer(
+        DockerImageName.parse("consul").withTag("1.14.3"))
+        .withExposedPorts(8500);
+    CONSUL_CONTAINER.start();
+  }
+
+  protected ConfigItem makeConfigItem(String key, String value) {
+    var item = new ConfigItem();
+    item.setKey(key);
+    item.setValue(value);
+    item.setPassword(false);
+
+    return item;
+  }
+
+  protected Environment mockEnvironment() {
+    var envMock = mock(Environment.class);
+    var hostVariableMock = mock(StringEnvironmentVariable.class);
+    var portVariableMock = mock(IntEnvironmentVariable.class);
+    var consulLocationMock = mock(StringEnvironmentVariable.class);
+    var spDebugMock = mock(BooleanEnvironmentVariable.class);
+
+    Mockito.when(hostVariableMock.getValueOrDefault()).thenReturn(CONSUL_CONTAINER.getHost());
+    Mockito.when(portVariableMock.getValueOrDefault()).thenReturn(CONSUL_CONTAINER.getMappedPort(8500));
+    Mockito.when(consulLocationMock.exists()).thenReturn(false);
+    Mockito.when(spDebugMock.exists()).thenReturn(false);
+    Mockito.when(spDebugMock.getValueOrReturn(false)).thenReturn(false);
+
+    Mockito.when(envMock.getConsulHost()).thenReturn(hostVariableMock);
+    Mockito.when(envMock.getConsulPort()).thenReturn(portVariableMock);
+    Mockito.when(envMock.getConsulLocation()).thenReturn(consulLocationMock);
+    Mockito.when(envMock.getSpDebug()).thenReturn(spDebugMock);
+
+    return envMock;
+
+  }
+}
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java
new file mode 100644
index 0000000000..5b14c0a636
--- /dev/null
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.integration.svcdiscovery;
+
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.ISpKvManagement;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+
+import com.google.gson.Gson;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class ConsulKvManagementTest extends AbstractConsulTest {
+
+  private ISpKvManagement kvManagement;
+
+  @Before
+  public void before() {
+    var environment = mockEnvironment();
+    this.kvManagement = SpServiceDiscovery
+        .getKeyValueStore(environment);
+  }
+
+  @Test
+  public void testKv() {
+    var test1Key = "a/test";
+    var test2Key = "a/test2";
+    var configItem1 = makeConfigItem(test1Key, "abc");
+    var configItem2 = makeConfigItem(test2Key, "def");
+    kvManagement.updateConfig(test1Key, serialize(configItem1), false);
+    kvManagement.updateConfig(test2Key, serialize(configItem2), false);
+
+    Map<String, String> result = kvManagement.getKeyValue("a");
+
+    assertEquals(2, result.size());
+    assertEquals(serialize(configItem1), result.get(test1Key));
+    assertEquals(serialize(configItem2), result.get(test2Key));
+  }
+
+  private String serialize(ConfigItem item) {
+    return new Gson().toJson(item);
+  }
+}
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpConfigTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpConfigTest.java
new file mode 100644
index 0000000000..7b64d3d391
--- /dev/null
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpConfigTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.integration.svcdiscovery;
+
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.SpConfig;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class ConsulSpConfigTest extends AbstractConsulTest {
+
+  private SpConfig spConfig;
+  @Before
+  public void before() {
+    var env = mockEnvironment();
+    this.spConfig = SpServiceDiscovery
+        .getSpConfig("test-service-group", env);
+  }
+
+  @Test
+  public void testKvString() {
+
+    spConfig.register("str", "abc", "desc");
+    var value = spConfig.getString("str");
+    assertEquals("abc", value);
+  }
+
+  @Test
+  public void testKvInteger() {
+    spConfig.register("int", 2, "desc");
+    var value = spConfig.getInteger("int");
+    assertEquals(2, value);
+  }
+
+  @Test
+  public void testKvConfigItem() {
+    var configItem = makeConfigItem("config", "value");
+    spConfig.register(configItem);
+    var value = spConfig.getConfigItem("config");
+    assertEquals(configItem, value);
+  }
+
+  @Test
+  public void testUpdateValue() {
+    spConfig.register("stru", "abc", "desc");
+    var value = spConfig.getString("stru");
+    assertEquals("abc", value);
+    spConfig.setString("stru", "abc2");
+    var result = spConfig.getString("stru");
+    assertEquals("abc2", result);
+  }
+}
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java
new file mode 100644
index 0000000000..0a4093c30e
--- /dev/null
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.integration.svcdiscovery;
+
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ConsulSpServiceDiscoveryTest extends AbstractConsulTest {
+
+  private ISpServiceDiscovery serviceDiscovery;
+
+  @Before
+  public void before() {
+    var env = mockEnvironment();
+    this.serviceDiscovery = SpServiceDiscovery
+        .getServiceDiscovery(env);
+  }
+
+  @Test
+  public void testServiceRegistration() {
+
+    serviceDiscovery.registerService(makeRequest(makeServiceTag(), "abc"));
+
+    var endpoints = getEndpoints(false);
+    assertEquals(1, endpoints.size());
+
+    serviceDiscovery.deregisterService("abc");
+
+    endpoints = getEndpoints(false);
+    assertEquals(0, endpoints.size());
+  }
+
+  @Test
+  public void testGetExtensionsServiceGroups() {
+    serviceDiscovery.registerService(makeRequest(List.of(DefaultSpServiceTags.PE, groupTag()), "abc"));
+    serviceDiscovery.registerService(makeRequest(
+        List.of(DefaultSpServiceTags.CORE, groupTag()), "def"));
+
+    var serviceGroups = serviceDiscovery.getExtensionsServiceGroups();
+
+    assertEquals(1, serviceGroups.size());
+    assertTrue(serviceGroups.containsKey("my-group"));
+
+    serviceDiscovery.deregisterService("abc");
+    serviceDiscovery.deregisterService("def");
+  }
+
+  private List<String> getEndpoints(boolean healthy) {
+    return serviceDiscovery.getServiceEndpoints(
+        DefaultSpServiceGroups.EXT,
+        healthy,
+        List.of(makeServiceTag().get(0).asString()));
+  }
+
+  private SpServiceRegistrationRequest makeRequest(List<SpServiceTag> serviceTags,
+                                                   String serviceId) {
+    var req = new SpServiceRegistrationRequest(
+        DefaultSpServiceGroups.EXT,
+        serviceId,
+        "localhost",
+        80,
+        serviceTags,
+        "/"
+    );
+
+    return req;
+  }
+
+  private SpServiceTag groupTag() {
+    return SpServiceTag.create(SpServiceTagPrefix.SP_GROUP, "my-group");
+  }
+
+  private List<SpServiceTag> makeServiceTag() {
+    return List.of(DefaultSpServiceTags.PE);
+  }
+}
diff --git a/streampipes-integration-tests/src/test/resources/logback-test.xml b/streampipes-integration-tests/src/test/resources/logback-test.xml
new file mode 100644
index 0000000000..d47f8f1cde
--- /dev/null
+++ b/streampipes-integration-tests/src/test/resources/logback-test.xml
@@ -0,0 +1,34 @@
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+<configuration debug="false" scan="true">
+
+    <appender name="STDOUT" target="System.out" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>DEBUG</level>
+        </filter>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT" />
+    </root>
+
+</configuration>