diff options
Diffstat (limited to 'sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl')
3 files changed, 20 insertions, 15 deletions
diff --git a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/MountpointRegistrarImpl.java b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/MountpointRegistrarImpl.java index 32d68ee62..deff2e336 100644 --- a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/MountpointRegistrarImpl.java +++ b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/MountpointRegistrarImpl.java @@ -95,7 +95,11 @@ public class MountpointRegistrarImpl implements AutoCloseable, IConfigChangedLis @Override public void onConfigChanged() { if (generalConfig == null) { // Included as NullPointerException observed once in docker logs - LOG.warn("onConfigChange cannot be handled. Unexpected Null"); + LOG.warn("onConfigChange cannot be handled. Unexpected Null for generalConfig"); + return; + } + if (strimziKafkaConfig == null) { // Included as NullPointerException observed once in docker logs + LOG.warn("onConfigChange cannot be handled. Unexpected Null for strimziKafkaConfig"); return; } LOG.info("Service configuration state changed. Enabled: {}", strimziKafkaConfig.getEnabled()); diff --git a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerImpl.java b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerImpl.java index 76870271d..b6090e3b5 100644 --- a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerImpl.java +++ b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerImpl.java @@ -25,7 +25,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.List; import java.util.Properties; import java.util.concurrent.ExecutionException; -import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.Admin; import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.GeneralConfig; import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.kafka.VESMsgKafkaConsumer; @@ -47,8 +46,9 @@ public abstract class StrimziKafkaVESMsgConsumerImpl protected final GeneralConfig generalConfig; Admin kafkaAdminClient = null; - protected StrimziKafkaVESMsgConsumerImpl(GeneralConfig generalConfig) { + protected StrimziKafkaVESMsgConsumerImpl(GeneralConfig generalConfig, Admin kafkaAdminClient) { this.generalConfig = generalConfig; + this.kafkaAdminClient = kafkaAdminClient; } /* @@ -107,9 +107,6 @@ public abstract class StrimziKafkaVESMsgConsumerImpl */ @Override public void init(Properties strimziKafkaProperties, Properties consumerProperties) { - Properties props = new Properties(); - props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, strimziKafkaProperties.getProperty("bootstrapServers")); - kafkaAdminClient = Admin.create(props); try { this.consumer = new VESMsgKafkaConsumer(strimziKafkaProperties, consumerProperties); @@ -158,6 +155,7 @@ public abstract class StrimziKafkaVESMsgConsumerImpl */ @Override public void stopConsumer() { + consumer.stop(); running = false; } diff --git a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerMain.java b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerMain.java index 03573d85b..c0c0f1836 100644 --- a/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerMain.java +++ b/sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/StrimziKafkaVESMsgConsumerMain.java @@ -23,6 +23,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Properties; +import org.apache.kafka.clients.admin.Admin; import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.FaultConfig; import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.GeneralConfig; import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.MessageConfig; @@ -62,6 +63,7 @@ public class StrimziKafkaVESMsgConsumerMain implements Runnable { private ProvisioningConfig provisioningConfig; private StndDefinedFaultConfig stndDefinedFaultConfig; private StrimziKafkaConfig strimziKafkaConfig; + private Admin kafkaAdminClient = null; public StrimziKafkaVESMsgConsumerMain(Map<String, MessageConfig> configMap, GeneralConfig generalConfig) { this.generalConfig = generalConfig; @@ -72,6 +74,7 @@ public class StrimziKafkaVESMsgConsumerMain implements Runnable { StrimziKafkaConfig strimziKafkaConfig) { this.generalConfig = generalConfig; this.strimziKafkaConfig = strimziKafkaConfig; + kafkaAdminClient = Admin.create(getStrimziKafkaProps(strimziKafkaConfig)); configMap.forEach(this::initialize); } @@ -148,10 +151,10 @@ public class StrimziKafkaVESMsgConsumerMain implements Runnable { private Properties getStrimziKafkaProps(StrimziKafkaConfig strimziKafkaConfig) { if (strimziKafkaProperties.size() == 0) { - strimziKafkaProperties.put("bootstrapServers", strimziKafkaConfig.getBootstrapServers()); - strimziKafkaProperties.put("securityProtocol", strimziKafkaConfig.getSecurityProtocol()); - strimziKafkaProperties.put("saslMechanism", strimziKafkaConfig.getSaslMechanism()); - strimziKafkaProperties.put("saslJaasConfig", strimziKafkaConfig.getSaslJaasConfig()); + strimziKafkaProperties.put("bootstrap.servers", strimziKafkaConfig.getBootstrapServers()); + strimziKafkaProperties.put("security.protocol", strimziKafkaConfig.getSecurityProtocol()); + strimziKafkaProperties.put("sasl.mechanism", strimziKafkaConfig.getSaslMechanism()); + strimziKafkaProperties.put("sasl.jaas.config", strimziKafkaConfig.getSaslJaasConfig()); } return strimziKafkaProperties; } @@ -170,13 +173,13 @@ public class StrimziKafkaVESMsgConsumerMain implements Runnable { StrimziKafkaVESMsgConsumerImpl consumer = null; if (consumerType.equalsIgnoreCase(_PNFREG_DOMAIN)) - consumer = new StrimziKafkaPNFRegVESMsgConsumer(generalConfig); + consumer = new StrimziKafkaPNFRegVESMsgConsumer(generalConfig, kafkaAdminClient); else if (consumerType.equalsIgnoreCase(_FAULT_DOMAIN)) - consumer = new StrimziKafkaFaultVESMsgConsumer(generalConfig); + consumer = new StrimziKafkaFaultVESMsgConsumer(generalConfig, kafkaAdminClient); else if (consumerType.equalsIgnoreCase(_CM_DOMAIN)) - consumer = new StrimziKafkaCMVESMsgConsumer(generalConfig); + consumer = new StrimziKafkaCMVESMsgConsumer(generalConfig, kafkaAdminClient); else if (consumerType.equals(_STNDDEFINED_FAULT_DOMAIN)) - consumer = new StrimziKafkaStndDefinedFaultVESMsgConsumer(generalConfig); + consumer = new StrimziKafkaStndDefinedFaultVESMsgConsumer(generalConfig, kafkaAdminClient); handleConsumer(consumer, consumerProperties, strimziKafkaProps, consumers); return !consumers.isEmpty(); @@ -216,7 +219,7 @@ public class StrimziKafkaVESMsgConsumerMain implements Runnable { Thread.currentThread().interrupt(); } } - + kafkaAdminClient.close(); LOG.info("No listener threads running - exiting"); } |