[streampipes] branch 1956-npe-in-consulconfigmigration updated: fix(#1956): Check Pulsar URL in consul migration runner

[ri...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

riemer pushed a commit to branch 1956-npe-in-consulconfigmigration
in repository https://gitbox.apache.org/repos/asf/streampipes.git


The following commit(s) were added to refs/heads/1956-npe-in-consulconfigmigration by this push:
     new b3751df5a fix(#1956): Check Pulsar URL in consul migration runner
b3751df5a is described below

commit b3751df5a5741d34de1154e09940ac915c8261ca
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Fri Sep 22 16:57:14 2023 +0200

    fix(#1956): Check Pulsar URL in consul migration runner
---
 .../service/core/migrations/v093/ConsulConfigMigration.java | 13 +++++++++++--
 .../org/apache/streampipes/svcdiscovery/api/SpConfig.java   |  2 ++
 .../streampipes/svcdiscovery/consul/ConsulSpConfig.java     |  7 ++++++-
 3 files changed, 19 insertions(+), 3 deletions(-)

diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
index 899c9a5ca..5da0d0bef 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java
@@ -19,10 +19,13 @@
 package org.apache.streampipes.service.core.migrations.v093;
 
 import org.apache.streampipes.config.backend.BackendConfig;
+import org.apache.streampipes.config.backend.BackendConfigKeys;
+import org.apache.streampipes.model.configuration.DefaultMessagingSettings;
 import org.apache.streampipes.model.configuration.SpCoreConfiguration;
 import org.apache.streampipes.service.core.migrations.Migration;
 import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
 
 public class ConsulConfigMigration implements Migration {
 
@@ -34,7 +37,7 @@ public class ConsulConfigMigration implements Migration {
 
   @Override
   public boolean shouldExecute() {
-    return storage.getAll().size() == 0;
+    return storage.getAll().isEmpty();
   }
 
   @Override
@@ -42,6 +45,12 @@ public class ConsulConfigMigration implements Migration {
     var currConf = BackendConfig.INSTANCE;
     var newConf = new SpCoreConfiguration();
 
+    var pulsarUrlExists = SpServiceDiscovery
+        .getSpConfig("backend")
+        .exists(BackendConfigKeys.PULSAR_URL);
+
+    var pulsarUrl = pulsarUrlExists ? currConf.getPulsarUrl() : new DefaultMessagingSettings().make().getPulsarUrl();
+
     var messagingSettings = currConf.getMessagingSettings();
     newConf.setLocalAuthConfig(currConf.getLocalAuthConfig());
     newConf.setEmailConfig(currConf.getEmailConfig());
@@ -55,7 +64,7 @@ public class ConsulConfigMigration implements Migration {
     messagingSettings.setNatsPort(currConf.getNatsPort());
     messagingSettings.setKafkaHost(currConf.getKafkaHost());
     messagingSettings.setKafkaPort(currConf.getKafkaPort());
-    messagingSettings.setPulsarUrl(currConf.getPulsarUrl());
+    messagingSettings.setPulsarUrl(pulsarUrl);
     messagingSettings.setZookeeperHost(currConf.getZookeeperHost());
     messagingSettings.setZookeeperPort(currConf.getZookeeperPort());
     newConf.setAssetDir(currConf.getAssetDir());
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java
index e6b0f8fcf..6ba4a2acd 100644
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java
+++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java
@@ -47,6 +47,8 @@ public interface SpConfig {
 
   String getString(String key);
 
+  boolean exists(String key);
+
   <T> T getObject(String key, Class<T> clazz, T defaultValue);
 
   ConfigItem getConfigItem(String key);
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 ad67689c8..48b3a93ed 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
@@ -135,6 +135,12 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
     return getConfigItem(key).getValue();
   }
 
+  @Override
+  public boolean exists(String key) {
+    var os = kvClient.getKVValue(addSn(key)).getValue();
+    return os != null;
+  }
+
   @Override
   public <T> T getObject(String key, Class<T> clazz, T defaultValue) {
     var os = kvClient.getKVValue(addSn(key));
@@ -153,7 +159,6 @@ public class ConsulSpConfig extends AbstractConsulService implements SpConfig {
   @Override
   public ConfigItem getConfigItem(String key) {
     var os = kvClient.getKVValue(addSn(key)).getValue();
-
     return fromJson(os.getDecodedValue());
   }