diff options
Diffstat (limited to 'prh-app-server/src/main/java/org')
18 files changed, 521 insertions, 233 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java index f668a581..c82c326f 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsBootstrapConfiguration.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,26 +23,36 @@ package org.onap.dcaegen2.services.bootstrap; import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration; +import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; + @Configuration @EnableConfigurationProperties(CbsProperties.class) public class CbsBootstrapConfiguration { - private static final CbsConfiguration CBS_CONFIGURATION = new CbsConfiguration(); + + private static CbsConfiguration CBS_CONFIGURATION = new CbsConfiguration(); + private static CbsConfigurationForAutoCommitDisabledMode CBS_CONFIGURATION_FOR_AUTO_COMMIT_DISABLED_MODE = + new CbsConfigurationForAutoCommitDisabledMode(); @Bean - public CbsProperties cbsProperties() { + public CbsProperties cbsProperties() + { return new CbsProperties(); } @Bean @ConditionalOnProperty(value = "cbs.enabled", matchIfMissing = true) + @Profile("!autoCommitDisabled") public CbsPropertySourceLocator cbsPropertySourceLocator( CbsProperties cbsProperties, CbsConfiguration cbsConfiguration) { + + System.out.println("Trying to return CbsPropertySourceLocator bean"); return new CbsPropertySourceLocator( cbsProperties, @@ -50,9 +61,33 @@ public class CbsBootstrapConfiguration { new CbsClientFactoryFacade(), cbsConfiguration); } + + @Bean + @ConditionalOnProperty(value = "cbs.enabled", matchIfMissing = true) + @Profile("autoCommitDisabled") + public CbsPropertySourceLocatorForAutoCommitDisabled cbsPropertySourceLocatorForAutoCommitDisabled(CbsProperties cbsProperties, + CbsConfigurationForAutoCommitDisabledMode cbsConfigurationforAutoCommitdisabledMode) { + + System.out.println("Trying to return CbsPropertySourceLocatorForAutoCommitDisabled bean"); + + CbsPropertySourceLocatorForAutoCommitDisabled cbsPropertySourceLocatorACDM = new CbsPropertySourceLocatorForAutoCommitDisabled(cbsProperties, + new CbsJsonToPropertyMapConverter(), new CbsClientConfigurationResolver(cbsProperties), + new CbsClientFactoryFacade(), cbsConfigurationforAutoCommitdisabledMode); + + return cbsPropertySourceLocatorACDM; + + } @Bean + @Profile("!autoCommitDisabled") public CbsConfiguration cbsConfiguration() { - return CBS_CONFIGURATION; + return CBS_CONFIGURATION; + } + + @Bean + @Profile("autoCommitDisabled") + public CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode() { + return CBS_CONFIGURATION_FOR_AUTO_COMMIT_DISABLED_MODE; } + } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java index 18d4021b..5fa4cdbe 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsProperties.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,34 +36,14 @@ public class CbsProperties { private Duration updatesInterval; @NestedConfigurationProperty private RetryProperties fetchRetries = new RetryProperties(); - private String hostname; - private Integer port; private String appName; CbsClientConfiguration toCbsClientConfiguration() { return ImmutableCbsClientConfiguration.builder() - .hostname(hostname) - .port(port) .appName(appName) .build(); } - public String getHostname() { - return hostname; - } - - public void setHostname(String hostname) { - this.hostname = hostname; - } - - public Integer getPort() { - return port; - } - - public void setPort(Integer port) { - this.port = port; - } - public String getAppName() { return appName; } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java index 2b5ac2e5..b4875eed 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocator.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2019-2021 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,6 +29,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnos import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.cloud.bootstrap.config.PropertySourceLocator; +import org.springframework.context.annotation.Profile; import org.springframework.core.env.Environment; import org.springframework.core.env.MapPropertySource; import org.springframework.core.env.PropertySource; @@ -35,6 +37,7 @@ import reactor.util.retry.Retry; import java.util.Map; +@Profile("!autoCommitDisabled") public class CbsPropertySourceLocator implements PropertySourceLocator { private static final Logger LOGGER = LoggerFactory.getLogger(CbsPropertySourceLocator.class); @@ -43,37 +46,40 @@ public class CbsPropertySourceLocator implements PropertySourceLocator { private final CbsClientConfigurationResolver cbsClientConfigurationResolver; private final CbsClientFactoryFacade cbsClientFactoryFacade; private final CbsConfiguration cbsConfiguration; - + public CbsPropertySourceLocator(CbsProperties cbsProperties, - CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter, - CbsClientConfigurationResolver cbsClientConfigurationResolver, - CbsClientFactoryFacade cbsClientFactoryFacade, - CbsConfiguration cbsConfiguration) { - this.cbsProperties = cbsProperties; - this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter; - this.cbsClientConfigurationResolver = cbsClientConfigurationResolver; - this.cbsClientFactoryFacade = cbsClientFactoryFacade; - this.cbsConfiguration = cbsConfiguration; + CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter, + CbsClientConfigurationResolver cbsClientConfigurationResolver, + CbsClientFactoryFacade cbsClientFactoryFacade, CbsConfiguration cbsConfiguration) { + + this.cbsProperties = cbsProperties; + this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter; + this.cbsClientConfigurationResolver = cbsClientConfigurationResolver; + this.cbsClientFactoryFacade = cbsClientFactoryFacade; + this.cbsConfiguration = cbsConfiguration; } @Override public PropertySource<?> locate(Environment environment) { + CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationResolver.resolveCbsClientConfiguration(); Map<String, Object> properties = cbsClientFactoryFacade.createCbsClient(cbsClientConfiguration) .flatMap(cbsClient -> cbsClient.get(CbsRequests.getAll(RequestDiagnosticContext.create()))) .doOnError(e -> LOGGER.warn("Failed loading configuration - retrying...", e)) - .retryWhen(Retry. - backoff(cbsProperties.getFetchRetries().getMaxAttempts(), cbsProperties.getFetchRetries().getFirstBackoff()). - maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff())) - .doOnNext(this::updateCbsConfig) - .map(cbsJsonToPropertyMapConverter::convertToMap) - .block(); + .retryWhen(Retry + .backoff(cbsProperties.getFetchRetries().getMaxAttempts(), + cbsProperties.getFetchRetries().getFirstBackoff()) + .maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff())) + .doOnNext(this::updateCbsConfig).map(cbsJsonToPropertyMapConverter::convertToMap).block(); + return new MapPropertySource("cbs", properties); } private void updateCbsConfig(JsonObject jsonObject) { try { + LOGGER.info("Updating CBS configuration"); cbsConfiguration.parseCBSConfig(jsonObject); + } catch (Exception e) { LOGGER.error("Failed parsing configuration", e); throw e; diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java new file mode 100644 index 00000000..b7aa1f58 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabled.java @@ -0,0 +1,94 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.bootstrap; + +import com.google.gson.JsonObject; +import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode; +import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests; +import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration; +import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.cloud.bootstrap.config.PropertySourceLocator; +import org.springframework.context.annotation.Profile; +import org.springframework.core.env.Environment; +import org.springframework.core.env.MapPropertySource; +import org.springframework.core.env.PropertySource; +import reactor.util.retry.Retry; +import java.util.Map; + +/** + * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on + * * 24/08/23 + * */ + +@Profile("autoCommitDisabled") +public class CbsPropertySourceLocatorForAutoCommitDisabled implements PropertySourceLocator { + private static final Logger LOGGER = LoggerFactory.getLogger(CbsPropertySourceLocatorForAutoCommitDisabled.class); + + private final CbsProperties cbsProperties; + private final CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter; + private final CbsClientConfigurationResolver cbsClientConfigurationResolver; + private final CbsClientFactoryFacade cbsClientFactoryFacade; + private final CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode; + + public CbsPropertySourceLocatorForAutoCommitDisabled(CbsProperties cbsProperties, + CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter, + CbsClientConfigurationResolver cbsClientConfigurationResolver, + CbsClientFactoryFacade cbsClientFactoryFacade, CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) { + + this.cbsProperties = cbsProperties; + this.cbsJsonToPropertyMapConverter = cbsJsonToPropertyMapConverter; + this.cbsClientConfigurationResolver = cbsClientConfigurationResolver; + this.cbsClientFactoryFacade = cbsClientFactoryFacade; + this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode; + + } + + @Override + public PropertySource<?> locate(Environment environment) { + + CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationResolver.resolveCbsClientConfiguration(); + Map<String, Object> properties = cbsClientFactoryFacade.createCbsClient(cbsClientConfiguration) + .flatMap(cbsClient -> cbsClient.get(CbsRequests.getAll(RequestDiagnosticContext.create()))) + .doOnError(e -> LOGGER.warn("Failed loading configuration - retrying...", e)) + .retryWhen(Retry + .backoff(cbsProperties.getFetchRetries().getMaxAttempts(), + cbsProperties.getFetchRetries().getFirstBackoff()) + .maxBackoff(cbsProperties.getFetchRetries().getMaxBackoff())) + .doOnNext(this::updateCbsConfig) + .map(cbsJsonToPropertyMapConverter::convertToMap).block(); + + return new MapPropertySource("cbs", properties); + } + + private void updateCbsConfig(JsonObject jsonObject) { + try { + LOGGER.info("Updating CBS configuration"); + cbsConfigurationForAutoCommitDisabledMode.parseCBSConfig(jsonObject); + + } catch (Exception e) { + LOGGER.error("Failed parsing configuration", e); + throw e; + } + } + +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java index 1d2a65d3..5a986517 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/MainApp.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,7 +23,6 @@ package org.onap.dcaegen2.services.prh; import java.util.Map; import java.util.UUID; - import org.slf4j.MDC; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -32,19 +32,22 @@ import org.springframework.context.annotation.Bean; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; - import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INVOCATION_ID; /** - * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18 + * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on + * 3/23/18 */ -@SpringBootApplication(exclude = {JacksonAutoConfiguration.class}) +@SpringBootApplication(exclude = { JacksonAutoConfiguration.class }) @EnableScheduling @EnableConfigurationProperties public class MainApp { + public static void main(String[] args) { + SpringApplication.run(MainApp.class, args); + } @Bean diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java index 39369329..cd0d8d2a 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/AaiHttpClientConfig.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +22,7 @@ package org.onap.dcaegen2.services.prh.configuration; import java.nio.charset.StandardCharsets; + import java.util.function.BiFunction; import org.onap.dcaegen2.services.prh.adapter.aai.api.AaiHttpClient; import org.onap.dcaegen2.services.prh.adapter.aai.api.AaiPnfResultModel; @@ -39,47 +41,48 @@ import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; @Configuration public class AaiHttpClientConfig { @Autowired - private CbsConfiguration cbsConfiguration; + private Config config; @Bean public AaiHttpClient<ConsumerDmaapModel, HttpResponse> getPatchClientFactory() { return createLazyConfigClient( - (config, client) -> new AaiHttpPatchClient(config, new AaiJsonBodyBuilderImpl(), client)); + (config, client) -> new AaiHttpPatchClient(config, new AaiJsonBodyBuilderImpl(), client)); } @Bean public AaiHttpClient<AaiServiceInstanceQueryModel, AaiServiceInstanceResultModel> getServiceInstanceClient() { return createLazyConfigClient( - (config, client) -> new AaiGetServiceInstanceClient(config, client) - .map(httpResponse -> { + (config, client) -> new AaiGetServiceInstanceClient(config, client).map(httpResponse -> { httpResponse.throwIfUnsuccessful(); - return httpResponse.bodyAsJson(StandardCharsets.UTF_8, - PrhModelAwareGsonBuilder.createGson(), AaiServiceInstanceResultModel.class); + return httpResponse.bodyAsJson(StandardCharsets.UTF_8, PrhModelAwareGsonBuilder.createGson(), + AaiServiceInstanceResultModel.class); })); } @Bean public AaiHttpClient<ConsumerDmaapModel, AaiPnfResultModel> getGetClient() { - return createLazyConfigClient( - (config, client) -> new AaiHttpGetClient(config, client) - .map(httpResponse -> { - httpResponse.throwIfUnsuccessful(); - return httpResponse.bodyAsJson(StandardCharsets.UTF_8, - PrhModelAwareGsonBuilder.createGson(), AaiPnfResultModel.class); - })); + + + + return createLazyConfigClient((config, client) -> new AaiHttpGetClient(config, client).map(httpResponse -> { + httpResponse.throwIfUnsuccessful(); + return httpResponse.bodyAsJson(StandardCharsets.UTF_8, PrhModelAwareGsonBuilder.createGson(), + AaiPnfResultModel.class); + })); } private <T, U> AaiHttpClient<T, U> createLazyConfigClient( - final BiFunction<AaiClientConfiguration, RxHttpClient, AaiHttpClient<T, U>> factoryMethod) { + final BiFunction<AaiClientConfiguration, RxHttpClient, AaiHttpClient<T, U>> factoryMethod) { +// System.out.println("pnf url in AAIClientConfiguration is: " + config.getAaiClientConfiguration().pnfUrl()); +// System.out.println("base url in AAIClientConfiguration is: " + config.getAaiClientConfiguration().baseUrl()); + return x -> factoryMethod.apply(config.getAaiClientConfiguration(), + new AaiHttpClientFactory(config.getAaiClientConfiguration()).build()).getAaiResponse(x); - return x -> factoryMethod.apply( - cbsConfiguration.getAaiClientConfiguration(), - new AaiHttpClientFactory(cbsConfiguration.getAaiClientConfiguration()).build() - ).getAaiResponse(x); } } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java index 8373018d..64fff9a7 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfiguration.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,35 +31,35 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRo import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.context.annotation.Profile; +@Profile("!autoCommitDisabled") public class CbsConfiguration implements Config { private static final Logger LOGGER = LoggerFactory.getLogger(CbsConfiguration.class); - private static final String CBS_CONFIG_MISSING = "CBS config missing"; - private AaiClientConfiguration aaiClientCBSConfiguration; - private MessageRouterPublisher messageRouterPublisher; - private MessageRouterSubscriber messageRouterSubscriber; - private MessageRouterPublishRequest messageRouterCBSPublishRequest; - private MessageRouterSubscribeRequest messageRouterCBSSubscribeRequest; - private MessageRouterPublishRequest messageRouterCBSUpdatePublishRequest; - - + protected static final String CBS_CONFIG_MISSING = "CBS config missing"; + protected AaiClientConfiguration aaiClientCBSConfiguration; + protected MessageRouterPublisher messageRouterPublisher; + protected MessageRouterSubscriber messageRouterSubscriber; + protected MessageRouterPublishRequest messageRouterCBSPublishRequest; + protected MessageRouterSubscribeRequest messageRouterCBSSubscribeRequest; + protected MessageRouterPublishRequest messageRouterCBSUpdatePublishRequest; + public void parseCBSConfig(JsonObject jsonObject) { + LOGGER.info("Received application configuration: {}", jsonObject); - CbsContentParser consulConfigurationParser = new CbsContentParser(jsonObject); - + CbsContentParser consulConfigurationParser = new CbsContentParser(jsonObject); aaiClientCBSConfiguration = consulConfigurationParser.getAaiClientConfig(); - messageRouterPublisher = DmaapClientFactory.createMessageRouterPublisher( - consulConfigurationParser.getMessageRouterPublisherConfig()); + messageRouterPublisher = DmaapClientFactory + .createMessageRouterPublisher(consulConfigurationParser.getMessageRouterPublisherConfig()); messageRouterCBSPublishRequest = consulConfigurationParser.getMessageRouterPublishRequest(); messageRouterCBSUpdatePublishRequest = consulConfigurationParser.getMessageRouterUpdatePublishRequest(); - messageRouterSubscriber = DmaapClientFactory.createMessageRouterSubscriber( - consulConfigurationParser.getMessageRouterSubscriberConfig()); + messageRouterSubscriber = DmaapClientFactory + .createMessageRouterSubscriber(consulConfigurationParser.getMessageRouterSubscriberConfig()); messageRouterCBSSubscribeRequest = consulConfigurationParser.getMessageRouterSubscribeRequest(); - } - + } @Override public MessageRouterPublisher getMessageRouterPublisher() { @@ -72,21 +73,27 @@ public class CbsConfiguration implements Config { @Override public MessageRouterPublishRequest getMessageRouterPublishRequest() { - return Optional.ofNullable(messageRouterCBSPublishRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); + return Optional.ofNullable(messageRouterCBSPublishRequest) + .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); } @Override public MessageRouterPublishRequest getMessageRouterUpdatePublishRequest() { - return Optional.ofNullable(messageRouterCBSUpdatePublishRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); + return Optional.ofNullable(messageRouterCBSUpdatePublishRequest) + .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); } @Override public AaiClientConfiguration getAaiClientConfiguration() { - return Optional.ofNullable(aaiClientCBSConfiguration).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); + return Optional.ofNullable(aaiClientCBSConfiguration) + .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); } @Override public MessageRouterSubscribeRequest getMessageRouterSubscribeRequest() { - return Optional.ofNullable(messageRouterCBSSubscribeRequest).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); + return Optional.ofNullable(messageRouterCBSSubscribeRequest) + .orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); } + + } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java new file mode 100644 index 00000000..b20cbad6 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledMode.java @@ -0,0 +1,67 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.prh.configuration; + +import java.util.Optional; +import org.onap.dcaegen2.services.prh.adapter.kafka.ImmutableKafkaConfiguration; +import org.onap.dcaegen2.services.prh.adapter.kafka.KafkaConfiguration; +import org.springframework.context.annotation.Profile; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; + +/** + * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on + * 24/08/23 + */ +@Profile("autoCommitDisabled") +public class CbsConfigurationForAutoCommitDisabledMode extends CbsConfiguration { + + protected KafkaConfiguration kafkaConfiguration; + + @Override + public void parseCBSConfig(JsonObject jsonObject) { + + super.parseCBSConfig(jsonObject); + JsonObject jsonObjectforAutoCommitDisabled = jsonObject.getAsJsonObject("config"); + JsonElement jsonObjectOfKafkaConfigurations = jsonObjectforAutoCommitDisabled.get("kafka-configurations"); + + kafkaConfiguration = new ImmutableKafkaConfiguration.Builder() + .kafkaBoostrapServerConfig( + ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaBoostrapServerConfig").getAsString()) + .groupIdConfig(((JsonObject) jsonObjectOfKafkaConfigurations).get("groupIdConfig").getAsString()) + .kafkaSaslMechanism( + ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaSaslMechanism").getAsString()) + .kafkaSecurityProtocol( + ((JsonObject) jsonObjectOfKafkaConfigurations).get("kafkaSecurityProtocol").getAsString()) + .kafkaJaasConfig(System.getenv("JAAS_CONFIG")) + .build(); + + } + + public KafkaConfiguration getKafkaConfig() { + return Optional.ofNullable(kafkaConfiguration).orElseThrow(() -> new RuntimeException(CBS_CONFIG_MISSING)); + } + + public void setKafkaConfiguration(KafkaConfiguration kafkaConfiguration) { + this.kafkaConfiguration = kafkaConfiguration; + } + +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java index ed935501..e1200119 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/CbsContentParser.java @@ -3,6 +3,7 @@ * PNF-REGISTRATION-HANDLER * ================================================================================ * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved. + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -139,4 +140,4 @@ class CbsContentParser { .timeout(Duration.ofMillis(jsonObject.get("dmaap.dmaapConsumerConfiguration.timeoutMs").getAsLong())) .build(); } -}
\ No newline at end of file +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java index 8affe281..baaf3b16 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfig.java @@ -17,80 +17,90 @@ * limitations under the License. * ============LICENSE_END========================================================= */ - package org.onap.dcaegen2.services.prh.configuration; +import java.util.HashMap; +import java.util.Map; +import org.springframework.context.annotation.Profile; import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Profile; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.listener.ContainerProperties; -import java.util.HashMap; -import java.util.Map; - -/** - * @author <a href="mailto:pravin.kokane@t-systems.com">Pravin Kokane</a> on 3/13/23 - */ + /** + * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on + * * 24/08/23 + * */ @Profile("autoCommitDisabled") @EnableKafka @Configuration -public class KafkaConfig -{ - String kafkaBoostrapServerConfig = System.getenv("kafkaBoostrapServerConfig"); - - String groupIdConfig = System.getenv("groupIdConfig"); - - - String kafkaSecurityProtocol = System.getenv("kafkaSecurityProtocol"); - - String kafkaSaslMechanism = System.getenv("kafkaSaslMechanism"); - - String kafkaUsername = System.getenv("kafkaUsername"); - - String kafkaPassword = System.getenv("kafkaPassword"); - - String kafkaJaasConfig = System.getenv("JAAS_CONFIG"); - - String kafkaLoginModuleClassConfig = System.getenv("Login_Module_Class"); +public class KafkaConfig { + + CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode; + + public String kafkaBoostrapServerConfig; + public String groupIdConfig; + public String kafkaSecurityProtocol; + public String kafkaSaslMechanism; + public String kafkaUsername; + public String kafkaPassword; + public String kafkaJaasConfigName; + public String kafkaLoginModuleClassConfig; + public String kafkaJaasConfig; + + public final String DEFAULT_KAFKA_SECURITY_PROTOCOL = "SASL_PLAINTEXT"; + public final String DEFAULT_KAFKA_SASL_MECHANISM = "SCRAM-SHA-512"; + + public KafkaConfig() { + + } @Bean - public ConsumerFactory<String, String> consumerFactory() - { - Map<String,Object> config = new HashMap<>(); - config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaBoostrapServerConfig); - config.put(ConsumerConfig.GROUP_ID_CONFIG,groupIdConfig); - config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); - config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer"); + public ConsumerFactory<String, String> consumerFactory(CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) { + this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode; + kafkaBoostrapServerConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig() + .kafkaBoostrapServerConfig(); + groupIdConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().groupIdConfig(); + kafkaSecurityProtocol = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaSecurityProtocol(); + kafkaSaslMechanism = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaSaslMechanism(); + kafkaJaasConfig = cbsConfigurationForAutoCommitDisabledMode.getKafkaConfig().kafkaJaasConfig(); + + Map<String, Object> config = new HashMap<>(); + config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBoostrapServerConfig); + + config.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig); + config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringDeserializer"); + config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringDeserializer"); config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - if(kafkaJaasConfig == null) { - kafkaJaasConfig = kafkaLoginModuleClassConfig + " required username=\"" - + kafkaUsername + "\" password=\"" + kafkaPassword + "\";"; - } - if(kafkaSecurityProtocol==null ) kafkaSecurityProtocol="SASL_PLAINTEXT"; + + if (kafkaSecurityProtocol == null) + kafkaSecurityProtocol = DEFAULT_KAFKA_SECURITY_PROTOCOL; config.put("security.protocol", kafkaSecurityProtocol); - if(kafkaSaslMechanism==null ) kafkaSaslMechanism="SCRAM-SHA-512"; + if (kafkaSaslMechanism == null) + kafkaSaslMechanism = DEFAULT_KAFKA_SASL_MECHANISM; config.put("sasl.mechanism", kafkaSaslMechanism); config.put("sasl.jaas.config", kafkaJaasConfig); return new DefaultKafkaConsumerFactory<>(config); + } @Bean - public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() - { + public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory( + CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode) { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); - factory.setConsumerFactory(consumerFactory()); + factory.setConsumerFactory(consumerFactory(cbsConfigurationForAutoCommitDisabledMode)); factory.setBatchListener(true); factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); return factory; } + } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java index 0b1f0e1c..fcbd10a5 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleController.java @@ -24,10 +24,10 @@ package org.onap.dcaegen2.services.prh.controllers; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.onap.dcaegen2.services.prh.tasks.ScheduledTasksRunner; -import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Profile; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.RequestMapping; @@ -36,15 +36,15 @@ import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; /** - * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/5/18 + * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on + * 4/5/18 */ @RestController -@Api(value = "ScheduleController", description = "Schedule Controller") +@Api(value = "ScheduleController") +@Profile("!autoCommitDisabled") public class ScheduleController { - private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleController.class); - private ScheduledTasksRunner scheduledTasksRunner; @@ -53,24 +53,21 @@ public class ScheduleController { this.scheduledTasksRunner = scheduledTasksRunner; } - - @RequestMapping(value = "start", method = RequestMethod.GET) @ApiOperation(value = "Start scheduling worker request") public Mono<ResponseEntity<String>> startTasks() { - return Mono.fromSupplier(scheduledTasksRunner::tryToStartTask).map(this::createStartTaskResponse); + LOGGER.trace("Receiving start scheduling worker request with Comit SchedulerController"); + return Mono.fromSupplier(scheduledTasksRunner::tryToStartTask).map(this::createStartTaskResponse); } - @RequestMapping(value = "stopPrh", method = RequestMethod.GET) @ApiOperation(value = "Receiving stop scheduling worker request") public Mono<ResponseEntity<String>> stopTask() { LOGGER.trace("Receiving stop scheduling worker request"); return Mono.defer(() -> { - scheduledTasksRunner.cancelTasks(); - return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK)); - } - ); + scheduledTasksRunner.cancelTasks(); + return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK)); + }); } private ResponseEntity<String> createStartTaskResponse(boolean wasScheduled) { diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java new file mode 100644 index 00000000..64e3a469 --- /dev/null +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabled.java @@ -0,0 +1,80 @@ +/* + * ============LICENSE_START======================================================= + * PNF-REGISTRATION-HANDLER + * ================================================================================ + * Copyright (C) 2023 Deutsche Telekom Intellectual Property. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.prh.controllers; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Profile; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; +import reactor.core.publisher.Mono; + +/** + * * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on + * * 24/08/23 + * */ +@RestController +@Api(value = "ScheduleController") +@Profile("autoCommitDisabled") +public class ScheduleControllerForAutoCommitDisabled { + + private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleControllerForAutoCommitDisabled.class); + + private ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit; + + @Autowired(required = false) + public ScheduleControllerForAutoCommitDisabled(ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit) { + this.scheduledTasksRunnerWithCommit = scheduledTasksRunnerWithCommit; + } + + @RequestMapping(value = "start", method = RequestMethod.GET) + @ApiOperation(value = "Start scheduling worker request") + public Mono<ResponseEntity<String>> startTasks() { + LOGGER.trace("Receiving start scheduling worker request with Comit SchedulerController"); + return Mono.fromSupplier(scheduledTasksRunnerWithCommit::tryToStartTaskWithCommit) + .map(this::createStartTaskResponse); + } + + @RequestMapping(value = "stopPrh", method = RequestMethod.GET) + @ApiOperation(value = "Receiving stop scheduling worker request") + public Mono<ResponseEntity<String>> stopTask() { + LOGGER.trace("Receiving stop scheduling worker request"); + return Mono.defer(() -> { + scheduledTasksRunnerWithCommit.cancelTasks(); + return Mono.just(new ResponseEntity<>("PRH Service has been stopped!", HttpStatus.OK)); + }); + } + + private ResponseEntity<String> createStartTaskResponse(boolean wasScheduled) { + if (wasScheduled) { + return new ResponseEntity<>("PRH Service has been started!", HttpStatus.CREATED); + } else { + return new ResponseEntity<>("PRH Service is already running!", HttpStatus.NOT_ACCEPTABLE); + } + } +} diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java index 4a7eef58..73131926 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AaiQueryTaskImpl.java @@ -79,8 +79,6 @@ public class AaiQueryTaskImpl implements AaiQueryTask { return getPnfModelClient .getAaiResponse(model) .flatMap(aaiModel -> Mono.just(model)); - - } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java index f305a925..b7c5c7ea 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java @@ -33,7 +33,6 @@ import org.slf4j.LoggerFactory; import org.slf4j.MDC; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.boot.configurationprocessor.json.JSONException; import org.springframework.context.annotation.Profile; import org.springframework.http.HttpStatus; import org.springframework.stereotype.Component; @@ -43,7 +42,6 @@ import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.function.Predicate; - import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID; import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE; diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java index 09e06da7..e90b0271 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java @@ -24,13 +24,8 @@ package org.onap.dcaegen2.services.prh.tasks; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ScheduledFuture; - import javax.annotation.PreDestroy; import org.onap.dcaegen2.services.prh.configuration.PrhProperties; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.slf4j.Marker; -import org.slf4j.MarkerFactory; import org.springframework.boot.context.event.ApplicationStartedEvent; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; @@ -38,31 +33,26 @@ import org.springframework.context.event.EventListener; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.annotation.EnableScheduling; - /** * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/13/18 */ + @Profile("!autoCommitDisabled") @Configuration @EnableScheduling public class ScheduledTasksRunner { - private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTasksRunner.class); - private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY"); - private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>(); + private static volatile List<ScheduledFuture> scheduledPrhTaskFutureList = new ArrayList<>(); private final TaskScheduler taskScheduler; private final ScheduledTasks scheduledTask; private final PrhProperties prhProperties; - public ScheduledTasksRunner(TaskScheduler taskScheduler, ScheduledTasks scheduledTask, PrhProperties prhProperties) { this.taskScheduler = taskScheduler; this.scheduledTask = scheduledTask; this.prhProperties = prhProperties; } - - String profile = System.getenv("SPRING_PROFILES_ACTIVE"); - + @EventListener public void onApplicationStartedEvent(ApplicationStartedEvent applicationStartedEvent) { tryToStartTask(); @@ -83,7 +73,6 @@ public class ScheduledTasksRunner { * @return status of operation execution: true - started, false - not started */ public synchronized boolean tryToStartTask() { - LOGGER.info(ENTRY, "Start scheduling PRH workflow"); if (scheduledPrhTaskFutureList.isEmpty()) { scheduledPrhTaskFutureList.add(taskScheduler .scheduleWithFixedDelay(scheduledTask::scheduleMainPrhEventTask, @@ -94,4 +83,3 @@ public class ScheduledTasksRunner { } } } - diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java index 30e6cff1..6b289f1c 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImpl.java @@ -22,10 +22,8 @@ package org.onap.dcaegen2.services.prh.tasks.commit; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel; +import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode; import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.configurationprocessor.json.JSONException; import org.springframework.context.annotation.Profile; import org.springframework.kafka.annotation.KafkaListener; @@ -33,50 +31,72 @@ import org.springframework.kafka.listener.BatchAcknowledgingMessageListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; - import java.util.ArrayList; import java.util.List; /** - * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on 3/13/23 + * @author <a href="mailto:ajinkya-patil@t-systems.com">Ajinkya Patil</a> on + * 3/13/23 */ @Profile("autoCommitDisabled") @Component public class KafkaConsumerTaskImpl implements KafkaConsumerTask, BatchAcknowledgingMessageListener<String, String> { - - private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerTaskImpl.class); - - @Autowired + + private DmaapConsumerJsonParser dmaapConsumerJsonParser; - @Autowired private EpochDateTimeConversion epochDateTimeConversion; + private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode; + private List<String> jsonEvent = new ArrayList<>(); + public List<String> getJsonEvent() { + return jsonEvent; + } + private Acknowledgment offset; - String kafkaTopic = System.getenv("kafkaTopic"); + public Acknowledgment getOffset() { + return offset; + } + + static String commonInURL = "/events/"; + + String kafkaTopic; - String groupIdConfig = System.getenv("groupIdConfig"); + String groupIdConfig; + + + public KafkaConsumerTaskImpl(CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode + ,DmaapConsumerJsonParser dmaapConsumerJsonParser,EpochDateTimeConversion epochDateTimeConversion) { + this.cbsConfigurationForAutoCommitDisabledMode = cbsConfigurationForAutoCommitDisabledMode; + this.dmaapConsumerJsonParser = dmaapConsumerJsonParser; + this.epochDateTimeConversion = epochDateTimeConversion; + String kafkaTopicURL = this.cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest() + .sourceDefinition().topicUrl(); + kafkaTopic = getTopicFromTopicUrl(kafkaTopicURL); + groupIdConfig = cbsConfigurationForAutoCommitDisabledMode.getMessageRouterSubscribeRequest().consumerGroup(); + + System.setProperty("kafkaTopic", kafkaTopic); + System.setProperty("groupIdConfig", groupIdConfig); + + } @Override @KafkaListener(topics = "${kafkaTopic}", groupId = "${groupIdConfig}") public void onMessage(List<ConsumerRecord<String, String>> list, Acknowledgment acknowledgment) { - - if (list != null && !list.isEmpty()) { - - - list.stream().filter(consumerRecord -> consumerRecord.timestamp() >= epochDateTimeConversion.getStartDateOfTheDay() && consumerRecord.timestamp() <= epochDateTimeConversion.getEndDateOfTheDay()) - .map(ConsumerRecord::value) - .forEach(value -> { - jsonEvent.add(value); + + if (list != null && !list.isEmpty()) { + list.stream().filter( + consumerRecord -> consumerRecord.timestamp() >= epochDateTimeConversion.getStartDateOfTheDay() + && consumerRecord.timestamp() <= epochDateTimeConversion.getEndDateOfTheDay()) + .map(ConsumerRecord::value).forEach(value -> { + jsonEvent.add(value); }); - } - offset = acknowledgment; } @@ -86,14 +106,25 @@ public class KafkaConsumerTaskImpl implements KafkaConsumerTask, BatchAcknowledg return dmaapConsumerJsonParser.getConsumerDmaapModelFromKafkaConsumerRecord(jsonEvent); } + public void setJsonEvent(List<String> jsonEvent) { + this.jsonEvent = jsonEvent; + } + @Override public void commitOffset() { - if(!jsonEvent.isEmpty()){ + if (!jsonEvent.isEmpty()) { jsonEvent.clear(); } - if(offset != null){ + if (offset != null) { offset.acknowledge(); } } + public String getTopicFromTopicUrl(String topicUrl) { + if (topicUrl.endsWith("/")) { + return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length(), topicUrl.lastIndexOf("/")); + } + return topicUrl.substring(topicUrl.indexOf(commonInURL) + commonInURL.length()); + } + } diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java index 64d7798e..91cdd122 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksRunnerWithCommit.java @@ -29,7 +29,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.Marker; import org.slf4j.MarkerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.event.ApplicationStartedEvent; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; @@ -52,7 +51,6 @@ public class ScheduledTasksRunnerWithCommit { private final TaskScheduler taskScheduler; private final PrhProperties prhProperties; - @Autowired private ScheduledTasksWithCommit scheduledTasksWithCommit; public ScheduledTasksRunnerWithCommit(TaskScheduler taskScheduler, ScheduledTasksWithCommit scheduledTasksWithCommit, @@ -64,7 +62,8 @@ public class ScheduledTasksRunnerWithCommit { @EventListener public void onApplicationStartedEvent(ApplicationStartedEvent applicationStartedEvent) { - tryToStartTaskWithCommit(); + LOGGER.info(ENTRY,"### in onApplicationStartedEvent"); + LOGGER.info(ENTRY,"###tryToStartTaskWithCommit="+tryToStartTaskWithCommit()); } /** @@ -72,6 +71,7 @@ public class ScheduledTasksRunnerWithCommit { */ @PreDestroy public synchronized void cancelTasks() { + LOGGER.info(ENTRY,"###In cancelTasks"); scheduledPrhTaskFutureList.forEach(x -> x.cancel(false)); scheduledPrhTaskFutureList.clear(); } @@ -96,4 +96,3 @@ public class ScheduledTasksRunnerWithCommit { } } - diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java index b0eae949..352c0bbc 100644 --- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java +++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/commit/ScheduledTasksWithCommit.java @@ -33,8 +33,6 @@ import org.onap.dcaegen2.services.prh.tasks.DmaapPublisherTask; import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel; import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables; -import org.slf4j.Marker; -import org.slf4j.MarkerFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; @@ -48,7 +46,8 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** - * @author <a href="mailto:sangeeta.bellara@t-systems.com">Sangeeta Bellara</a> on 3/13/23 + * @author <a href="mailto:sangeeta.bellara@t-systems.com">Sangeeta Bellara</a> + * on 3/13/23 */ @Profile("autoCommitDisabled") @Component @@ -59,7 +58,7 @@ public class ScheduledTasksWithCommit { private KafkaConsumerTask kafkaConsumerTask; private DmaapPublisherTask dmaapReadyProducerTask; private DmaapPublisherTask dmaapUpdateProducerTask; - private AaiQueryTask aaiQueryTask; + public AaiQueryTask aaiQueryTask; private AaiProducerTask aaiProducerTask; private BbsActionsTask bbsActionsTask; private Map<String, String> mdcContextMap; @@ -73,17 +72,16 @@ public class ScheduledTasksWithCommit { * @param aaiPublisherTask - second task */ @Autowired - public ScheduledTasksWithCommit( - final KafkaConsumerTask kafkaConsumerTask, - @Qualifier("ReadyPublisherTask") final DmaapPublisherTask dmaapReadyPublisherTask, - @Qualifier("UpdatePublisherTask") final DmaapPublisherTask dmaapUpdatePublisherTask, - final AaiQueryTask aaiQueryTask, - final AaiProducerTask aaiPublisherTask, - final BbsActionsTask bbsActionsTask, - final Map<String, String> mdcContextMap) { + public ScheduledTasksWithCommit(final KafkaConsumerTask kafkaConsumerTask, + @Qualifier("ReadyPublisherTask") final DmaapPublisherTask dmaapReadyPublisherTask, + @Qualifier("UpdatePublisherTask") final DmaapPublisherTask dmaapUpdatePublisherTask, + final AaiQueryTask aaiQueryTask, final AaiProducerTask aaiPublisherTask, + final BbsActionsTask bbsActionsTask, final Map<String, String> mdcContextMap) + + { this.dmaapReadyProducerTask = dmaapReadyPublisherTask; this.dmaapUpdateProducerTask = dmaapUpdatePublisherTask; - this.kafkaConsumerTask=kafkaConsumerTask; + this.kafkaConsumerTask = kafkaConsumerTask; this.aaiQueryTask = aaiQueryTask; this.aaiProducerTask = aaiPublisherTask; this.bbsActionsTask = bbsActionsTask; @@ -92,7 +90,7 @@ public class ScheduledTasksWithCommit { static class State { public ConsumerDmaapModel dmaapModel; - public Boolean activationStatus; + public Boolean activationStatus; public State(ConsumerDmaapModel dmaapModel, final Boolean activationStatus) { this.dmaapModel = dmaapModel; @@ -103,50 +101,47 @@ public class ScheduledTasksWithCommit { public void scheduleKafkaPrhEventTask() { MdcVariables.setMdcContextMap(mdcContextMap); try { + LOGGER.info("Execution of tasks was registered with commit"); CountDownLatch mainCountDownLatch = new CountDownLatch(1); consumeFromKafkaMessage() - .flatMap(model->queryAaiForPnf(model) - .doOnError(e -> { LOGGER.info("PNF Not Found in AAI --> {}" + e); - disableCommit(); - }) - .onErrorResume(e -> Mono.empty()) - - ) - .flatMap(this::queryAaiForConfiguration) - .flatMap(this::publishToAaiConfiguration) - .flatMap(this::processAdditionalFields) - .flatMap(this::publishToDmaapConfiguration) + .flatMap(model -> queryAaiForPnf(model).doOnError(e -> { + LOGGER.info("PNF Not Found in AAI --> {}" + e); + LOGGER.info("PNF Not Found in AAI With description of exception --> {}" + e.getMessage()); + disableCommit(); + }).onErrorResume(e -> Mono.empty()) + + ) + .flatMap(this::queryAaiForConfiguration) + .flatMap(this::publishToAaiConfiguration) + .flatMap(this::processAdditionalFields).flatMap(this::publishToDmaapConfiguration) + .onErrorResume(e -> Mono.empty()) - - .doOnTerminate(mainCountDownLatch::countDown) - .subscribe(this::onSuccess, this::onError, this::onCompleteKafka); + + .doOnTerminate(mainCountDownLatch::countDown) + .subscribe(this::onSuccess, this::onError, this::onCompleteKafka); mainCountDownLatch.await(); - } catch (InterruptedException | JSONException e ) { + } catch (InterruptedException | JSONException e) { LOGGER.warn("Interruption problem on countDownLatch {}", e); Thread.currentThread().interrupt(); } } - private static void disableCommit() - { - pnfFound=false; + private static void disableCommit() { + pnfFound = false; } private void onCompleteKafka() { LOGGER.info("PRH tasks have been completed"); - if(pnfFound){ + if (pnfFound) { kafkaConsumerTask.commitOffset(); LOGGER.info("Committed the Offset"); - } - else - { + } else { LOGGER.info("Offset not Committed"); - pnfFound=true; + pnfFound = true; } } - private void onSuccess(MessageRouterPublishResponse response) { if (response.successful()) { String statusCodeOk = HttpStatus.OK.name(); @@ -167,23 +162,18 @@ public class ScheduledTasksWithCommit { } private Mono<State> queryAaiForConfiguration(final ConsumerDmaapModel monoDMaaPModel) { - return aaiQueryTask - .execute(monoDMaaPModel) - .map(x -> new State(monoDMaaPModel, x)); + return aaiQueryTask.execute(monoDMaaPModel).map(x -> new State(monoDMaaPModel, x)); } private Mono<ConsumerDmaapModel> queryAaiForPnf(final ConsumerDmaapModel monoDMaaPModel) { - LOGGER.info("Find PNF --> "+monoDMaaPModel.getCorrelationId()); + LOGGER.info("Find PNF --> " + monoDMaaPModel.getCorrelationId()); return aaiQueryTask.findPnfinAAI(monoDMaaPModel); } - private Mono<State> publishToAaiConfiguration(final State state) { try { - return aaiProducerTask - .execute(state.dmaapModel) - .map(x -> state); + return aaiProducerTask.execute(state.dmaapModel).map(x -> state); } catch (PrhTaskException e) { LOGGER.warn("AAIProducerTask exception has been registered: {}", e); return Mono.error(e); |