Replace consul-client with consul-api library (#1158)

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

streampipes-client-python/docs/examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb

@@ -610,4 +610,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}

streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java

@@ -19,35 +19,37 @@
 
 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 String getValueOrDefault(String defaultValue) {
     return this.exists() ? this.getValue() : defaultValue;
   }
 
+  public String getDefaultValue() {
+    return defaultValue;
+  }
 }

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);
+  }
+}

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();
+}

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();
+  }
+}

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());
+  }
+}

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();
+}

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);
+
+}

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);
+  }
+}

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;
+  }
+}