[DCAEGEN2] Pass autoCommitDisabled mode configuration via CBSContentParser

Pass autoCommitDisabled mode configuration via CBSContentParser instead
of environment variables.

Issue-ID: DCAEGEN2-3365
Change-Id: I3c56153d66a7291966d298874f12a01839f3ee3f
Signed-off-by: pranitk1905 <pranit.kapdule@t-systems.com>

[DCAEGEN2] Pass autoCommitDisabled mode configuration via CBSContentParser

Mentioned Copyrights and removed tabbed indentations

Issue-ID: DCAEGEN2-3365
Change-Id: I3c56153d66a7291966d298874f12a01839f3ee3f
Signed-off-by: pranitk1905 <pranit.kapdule@t-systems.com>

[DCAEGEN2] Pass autoCommitDisabled mode configuration via CBSContentParser

Added missing copyrights

Issue-ID: DCAEGEN2-3365
Change-Id: I3c56153d66a7291966d298874f12a01839f3ee3f
Signed-off-by: pranitk1905 <pranit.kapdule@t-systems.com>

[DCAEGEN2] Pass autoCommitDisabled mode configuration via CBSContentParser

Added missing copyrights

Issue-ID: DCAEGEN2-3365
Change-Id: I3c56153d66a7291966d298874f12a01839f3ee3f
Signed-off-by: pranitk1905 <pranit.kapdule@t-systems.com>

[DCAEGEN2] Pass autoCommitDisabled mode configuration via CBSContentParser

Removed Docs folder from the project as it is maintained in different
place

Issue-ID: DCAEGEN2-3365
Change-Id: I3c56153d66a7291966d298874f12a01839f3ee3f
Signed-off-by: pranitk1905 <pranit.kapdule@t-systems.com>

34 files changed, 1482 insertions, 347 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);
diff --git a/prh-app-server/src/main/resources/logback-spring.xml b/prh-app-server/src/main/resources/logback-spring.xml
index 060cf6c5..c4d31e52 100644
--- a/prh-app-server/src/main/resources/logback-spring.xml
+++ b/prh-app-server/src/main/resources/logback-spring.xml
@@ -1,8 +1,27 @@
 <?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ ============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=========================================================
+-->
 <configuration debug="false" scan="false">
   <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
   <property name="outputFilename" value="prh-app-server_output"/>
-  <property name="logPath" value="/var/log/ONAP/prh/prh-app-server"/>
+  <property name="logPath" value="/var/log/ONAP/prh/prh-app-server"/> 
   <property name="archivePath" value="${logPath}/archive"/>
   <property name="maxFileSize" value="50MB"/>
   <property name="maxHistory" value="30"/>
@@ -17,7 +36,7 @@
     |%thread
     |%n"/>
 
-  <springProfile name="prod">
+  <springProfile name="prod,autoCommitDisabled">
     <appender class="ch.qos.logback.core.ConsoleAppender" name="CONSOLE" target="SYSTEM_OUT">
       <encoder>
         <pattern>${defaultPattern}</pattern>
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsClientConfigurationResolverTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsClientConfigurationResolverTest.java
index 87dd18ca..ca1413a0 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsClientConfigurationResolverTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsClientConfigurationResolverTest.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.
@@ -24,7 +25,6 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
-
 import static org.assertj.core.api.Assertions.assertThat;
 
 class CbsClientConfigurationResolverTest {
@@ -34,18 +34,15 @@ class CbsClientConfigurationResolverTest {
     @BeforeEach
     void setUp() {
         cbsProperties = new CbsProperties();
-        cbsProperties.setHostname("some-cbs-host");
-        cbsProperties.setPort(123);
         cbsProperties.setAppName("client-app-name");
     }
 
     @Test
     @DisabledIfEnvironmentVariable(named = "CONFIG_BINDING_SERVICE", matches = ".+")
     void whenCbsEnvPropertiesAreNotePresentInEnvironment_ShouldFallbackToLoadingDefaultsFromCbsProperties() {
-        CbsClientConfiguration config = new CbsClientConfigurationResolver(cbsProperties).resolveCbsClientConfiguration();
+        CbsClientConfiguration config = new CbsClientConfigurationResolver(cbsProperties)
+                .resolveCbsClientConfiguration();
 
-        assertThat(config.hostname()).isEqualTo(cbsProperties.getHostname());
-        assertThat(config.port()).isEqualTo(cbsProperties.getPort());
         assertThat(config.appName()).isEqualTo(cbsProperties.getAppName());
     }
-}
\ No newline at end of file
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabledTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabledTest.java
new file mode 100644
index 00000000..f5863ac5
--- /dev/null
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorForAutoCommitDisabledTest.java
@@ -0,0 +1,112 @@
+/*
+ * ============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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import java.util.Map;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.RequestPath;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
+import org.springframework.core.env.Environment;
+import org.springframework.test.context.ActiveProfiles;
+import com.google.common.collect.ImmutableMap;
+import com.google.gson.JsonObject;
+import reactor.core.publisher.Mono;
+import reactor.test.scheduler.VirtualTimeScheduler;
+
+/**
+ *  * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ *   *        24/08/23
+ *    */
+
+@ExtendWith(MockitoExtension.class)
+@ActiveProfiles(value = "autoCommitDisabled")
+public class CbsPropertySourceLocatorForAutoCommitDisabledTest {
+    private static final RequestPath GET_ALL_REQUEST_PATH = CbsRequests.getAll(RequestDiagnosticContext.create())
+            .requestPath();
+
+    private CbsProperties cbsProperties = new CbsProperties();
+    @Mock
+    private CbsJsonToPropertyMapConverter cbsJsonToPropertyMapConverter;
+    @Mock
+    private CbsClientConfigurationResolver cbsClientConfigurationResolver;
+    @Mock
+    private CbsClientFactoryFacade cbsClientFactoryFacade;
+    @Mock
+    private CbsConfiguration cbsConfiguration;
+    @Mock
+    private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode;
+    @Mock
+    private Environment environment;
+    @Mock
+    private CbsClient cbsClient;
+    @Mock
+    private JsonObject cbsConfigJsonObject;
+
+    private Map<String, Object> cbsConfigMap = ImmutableMap.of("foo", "bar");
+
+    private VirtualTimeScheduler virtualTimeScheduler;
+
+    private CbsPropertySourceLocatorForAutoCommitDisabled cbsPropertySourceLocatorACDM;
+
+    @BeforeEach
+    void setup() {
+        virtualTimeScheduler = VirtualTimeScheduler.getOrSet(true);
+
+        cbsPropertySourceLocatorACDM = new CbsPropertySourceLocatorForAutoCommitDisabled(cbsProperties,
+                cbsJsonToPropertyMapConverter, cbsClientConfigurationResolver, cbsClientFactoryFacade,
+                cbsConfigurationForAutoCommitDisabledMode);
+
+    }
+
+    @AfterEach
+    void cleanup() {
+        virtualTimeScheduler.dispose();
+    }
+
+    @Test
+    void cbsProperySourceLocatorForAutoCommitDisabledTest() throws Exception {
+       
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
+        when(cbsClient.get(argThat(request -> request.requestPath().equals(GET_ALL_REQUEST_PATH))))
+                .thenReturn(Mono.just(cbsConfigJsonObject));
+        when(cbsJsonToPropertyMapConverter.convertToMap(cbsConfigJsonObject)).thenReturn(cbsConfigMap);
+
+        cbsPropertySourceLocatorACDM.locate(environment);
+       
+        verify(cbsConfigurationForAutoCommitDisabledMode).parseCBSConfig(cbsConfigJsonObject);
+       
+        
+       }
+
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorTest.java
index 22a11ed6..c1e938e5 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/bootstrap/CbsPropertySourceLocatorTest.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.
@@ -29,6 +30,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
@@ -39,20 +41,21 @@ import org.springframework.core.env.Environment;
 import org.springframework.core.env.PropertySource;
 import reactor.core.publisher.Mono;
 import reactor.test.scheduler.VirtualTimeScheduler;
-
 import java.util.Map;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assumptions.assumeThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
-import static org.mockito.Mockito.*;
-
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.doThrow;
 
 @ExtendWith(MockitoExtension.class)
 class CbsPropertySourceLocatorTest {
 
-    private static final RequestPath GET_ALL_REQUEST_PATH = CbsRequests.getAll(RequestDiagnosticContext.create()).requestPath();
+    private static final RequestPath GET_ALL_REQUEST_PATH = CbsRequests.getAll(RequestDiagnosticContext.create())
+            .requestPath();
 
     private CbsProperties cbsProperties = new CbsProperties();
     @Mock
@@ -66,28 +69,24 @@ class CbsPropertySourceLocatorTest {
     @Mock
     private CbsConfiguration cbsConfiguration;
     @Mock
+    private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode;
+    @Mock
     private Environment environment;
     @Mock
     private CbsClient cbsClient;
     @Mock
     private JsonObject cbsConfigJsonObject;
     private Map<String, Object> cbsConfigMap = ImmutableMap.of("foo", "bar");
-
     private VirtualTimeScheduler virtualTimeScheduler;
-
     private CbsPropertySourceLocator cbsPropertySourceLocator;
 
-
     @BeforeEach
     void setup() {
         virtualTimeScheduler = VirtualTimeScheduler.getOrSet(true);
 
-        when(cbsClientConfigurationResolver.resolveCbsClientConfiguration()).thenReturn(cbsClientConfiguration);
-        when(cbsClientFactoryFacade.createCbsClient(cbsClientConfiguration)).thenReturn(Mono.just(cbsClient));
+        cbsPropertySourceLocator = new CbsPropertySourceLocator(cbsProperties, cbsJsonToPropertyMapConverter,
+                cbsClientConfigurationResolver, cbsClientFactoryFacade, cbsConfiguration);
 
-        cbsPropertySourceLocator = new CbsPropertySourceLocator(
-                cbsProperties, cbsJsonToPropertyMapConverter, cbsClientConfigurationResolver,
-                cbsClientFactoryFacade, cbsConfiguration);
     }
 
     @AfterEach
@@ -95,9 +94,10 @@ class CbsPropertySourceLocatorTest {
         virtualTimeScheduler.dispose();
     }
 
-
     @Test
     void shouldBuildCbsPropertySourceBasedOnDataFetchedUsingCbsClient() {
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
         when(cbsClient.get(argThat(request -> request.requestPath().equals(GET_ALL_REQUEST_PATH))))
                 .thenReturn(Mono.just(cbsConfigJsonObject));
         when(cbsJsonToPropertyMapConverter.convertToMap(cbsConfigJsonObject)).thenReturn(cbsConfigMap);
@@ -108,9 +108,10 @@ class CbsPropertySourceLocatorTest {
         assertThat(propertySource).extracting(s -> s.getProperty("foo")).isEqualTo("bar");
     }
 
-
     @Test
     void shouldUpdateCbsConfigurationStateBasedOnDataFetchedUsingCbsClient() {
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
         when(cbsClient.get(argThat(request -> request.requestPath().equals(GET_ALL_REQUEST_PATH))))
                 .thenReturn(Mono.just(cbsConfigJsonObject));
         when(cbsJsonToPropertyMapConverter.convertToMap(cbsConfigJsonObject)).thenReturn(cbsConfigMap);
@@ -120,9 +121,10 @@ class CbsPropertySourceLocatorTest {
         verify(cbsConfiguration).parseCBSConfig(cbsConfigJsonObject);
     }
 
-
     @Test
     void shouldPropagateExceptionWhenCbsConfigurationParsingFails() {
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
         when(cbsClient.get(any(CbsRequest.class))).thenReturn(Mono.just(cbsConfigJsonObject));
 
         RuntimeException someCbsConfigParsingException = new RuntimeException("boom!");
@@ -134,13 +136,13 @@ class CbsPropertySourceLocatorTest {
 
     @Test
     void shouldRetryFetchingConfigFromCbsInCaseOfFailure() {
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
         assumeThat(cbsProperties.getFetchRetries().getMaxAttempts()).isGreaterThan(1);
-        when(cbsClient.get(any(CbsRequest.class)))
-                .thenReturn(Mono.defer(() -> {
-                        virtualTimeScheduler.advanceTimeBy(cbsProperties.getFetchRetries().getMaxBackoff());
-                        return Mono.error(new RuntimeException("some connection failure"));
-                }))
-                .thenReturn(Mono.just(cbsConfigJsonObject));
+        when(cbsClient.get(any(CbsRequest.class))).thenReturn(Mono.defer(() -> {
+            virtualTimeScheduler.advanceTimeBy(cbsProperties.getFetchRetries().getMaxBackoff());
+            return Mono.error(new RuntimeException("some connection failure"));
+        })).thenReturn(Mono.just(cbsConfigJsonObject));
         when(cbsJsonToPropertyMapConverter.convertToMap(cbsConfigJsonObject)).thenReturn(cbsConfigMap);
 
         PropertySource<?> propertySource = cbsPropertySourceLocator.locate(environment);
@@ -150,15 +152,16 @@ class CbsPropertySourceLocatorTest {
 
     @Test
     void shouldFailAfterExhaustingAllOfConfiguredRetryAttempts() {
+        Mono<CbsClient> just = Mono.just(cbsClient);
+        when(cbsClientFactoryFacade.createCbsClient(any())).thenReturn(just);
         assumeThat(cbsProperties.getFetchRetries().getMaxAttempts()).isGreaterThan(1);
-        when(cbsClient.get(any(CbsRequest.class)))
-                .thenReturn(Mono.defer(() -> {
-                    virtualTimeScheduler.advanceTimeBy(cbsProperties.getFetchRetries().getMaxBackoff());
-                    return Mono.error(new RuntimeException("some connection failure"));
-                }));
+        when(cbsClient.get(any(CbsRequest.class))).thenReturn(Mono.defer(() -> {
+            virtualTimeScheduler.advanceTimeBy(cbsProperties.getFetchRetries().getMaxBackoff());
+            return Mono.error(new RuntimeException("some connection failure"));
+        }));
 
-        assertThatThrownBy(() -> cbsPropertySourceLocator.locate(environment))
-                .hasMessageContaining("Retries exhausted")
+        assertThatThrownBy(() -> cbsPropertySourceLocator.locate(environment)).hasMessageContaining("Retries exhausted")
                 .hasMessageContaining(cbsProperties.getFetchRetries().getMaxAttempts().toString());
     }
+
 }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledModeTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledModeTest.java
new file mode 100644
index 00000000..89f1a043
--- /dev/null
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationForAutoCommitDisabledModeTest.java
@@ -0,0 +1,109 @@
+/*
+ * ============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 static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable;
+import static java.lang.ClassLoader.getSystemResource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import org.junit.jupiter.api.Test;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+
+/**
+ *  * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ *   *        24/08/23
+ *    */
+
+public class CbsConfigurationForAutoCommitDisabledModeTest {
+
+    /**
+     * Testcase is used to check correctness of values provided by
+     * autoCommitDisabledConfigurationFromCbs2.json
+     */
+
+    @Test
+    void beforecbsConfigurationForAutoCommitDisabledMode() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            this.cbsConfigurationForAutoCommitDisabledMode();
+        });
+    }
+
+    void cbsConfigurationForAutoCommitDisabledMode() throws Exception {
+
+        JsonObject cbsConfigJsonForAutoCommitDisabled = new Gson().fromJson(
+                new String(Files.readAllBytes(
+                        Paths.get(getSystemResource("autoCommitDisabledConfigurationFromCbs2.json").toURI()))),
+                JsonObject.class);
+        CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabled = new CbsConfigurationForAutoCommitDisabledMode();
+
+        cbsConfigurationForAutoCommitDisabled.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+
+        String expectedKafKaBoostrapServerConfig = "onap-strimzi-kafka-bootstrap:9092";
+        String actualKafkaBoostrapServerConfig = (cbsConfigurationForAutoCommitDisabled.getKafkaConfig()
+                .kafkaBoostrapServerConfig());
+
+        String expectedGroupIdConfig = "OpenDCAE-c12";
+        String actualGroupIdConfig = (cbsConfigurationForAutoCommitDisabled.getKafkaConfig().groupIdConfig());
+
+        String expectedKafkaSecurityProtocol = "SASL_PLAINTEXT";
+        String actualKafkaSecurityProtocol = (cbsConfigurationForAutoCommitDisabled.getKafkaConfig()
+                .kafkaSecurityProtocol());
+
+        String expectedKafkaSaslMechanism = "SCRAM-SHA-512";
+        String actualKafkaSaslMechanism = (cbsConfigurationForAutoCommitDisabled.getKafkaConfig().kafkaSaslMechanism());
+
+        String expectedKafkaJaasConfig = "jaas_config";
+        String actualKafkaJaasConfig = (cbsConfigurationForAutoCommitDisabled.getKafkaConfig().kafkaJaasConfig());
+
+        String expectedAaiUserName = "AAI";
+        String actualAaiUserName = (cbsConfigurationForAutoCommitDisabled.getAaiClientConfiguration().aaiUserName());
+
+        String expectedConsumerGroup = "OpenDCAE-c12";
+        String actualConsumerGroup = (cbsConfigurationForAutoCommitDisabled.getMessageRouterSubscribeRequest()
+                .consumerGroup());
+
+        assertEquals(expectedKafKaBoostrapServerConfig, actualKafkaBoostrapServerConfig,
+                "Expected value of KafKaBoostrapServerConfig is not matching with actual value");
+        assertEquals(expectedGroupIdConfig, actualGroupIdConfig,
+                "Expected value of GroupIdConfig is not matching with actual value");
+        assertEquals(expectedKafkaSecurityProtocol, actualKafkaSecurityProtocol,
+                "Expected value of KafkaSecurityProtocol is not matching with actual value");
+        assertEquals(expectedKafkaSaslMechanism, actualKafkaSaslMechanism,
+                "Expected value of KafkaSaslMechanism is not matching with actual value");
+        assertEquals(expectedKafkaJaasConfig, actualKafkaJaasConfig,
+                "Expected value of KafkaJaasConfig is not matching with actual value");
+        assertEquals(expectedAaiUserName, actualAaiUserName,
+                "Expected value of AaiUserName is not matching with actual value");
+        assertEquals(expectedConsumerGroup, actualConsumerGroup,
+                "Expected value of ConsumerGroup is not matching with actual value");
+
+        assertThat((cbsConfigurationForAutoCommitDisabled).getAaiClientConfiguration()).isNotNull();
+        assertThat((cbsConfigurationForAutoCommitDisabled).getMessageRouterPublisher()).isNotNull();
+        assertThat((cbsConfigurationForAutoCommitDisabled).getMessageRouterSubscriber()).isNotNull();
+        assertThat((cbsConfigurationForAutoCommitDisabled).getMessageRouterPublishRequest()).isNotNull();
+        assertThat((cbsConfigurationForAutoCommitDisabled).getMessageRouterSubscribeRequest()).isNotNull();
+        assertThat((cbsConfigurationForAutoCommitDisabled).getMessageRouterUpdatePublishRequest()).isNotNull();
+
+    }
+
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationTest.java
index 7f5d26fc..4f3cd864 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/CbsConfigurationTest.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.
@@ -23,10 +24,8 @@ package org.onap.dcaegen2.services.prh.configuration;
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 import org.junit.jupiter.api.Test;
-
 import java.nio.file.Files;
 import java.nio.file.Paths;
-
 import static java.lang.ClassLoader.getSystemResource;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -51,11 +50,11 @@ class CbsConfigurationTest {
                 .hasMessage(EXPECTED_ERROR_MESSAGE_WHEN_CBS_CONFIG_IS_NOT_INITIALIZED);
     }
 
-
     @Test
     void cbsConfigurationShouldExposeDataReceivedAsJsonFromCbs() throws Exception {
-        JsonObject cbsConfigJson = new Gson().fromJson(new String(Files.readAllBytes(Paths.get(
-                getSystemResource("configurationFromCbs.json").toURI()))), JsonObject.class);
+        JsonObject cbsConfigJson = new Gson().fromJson(
+                new String(Files.readAllBytes(Paths.get(getSystemResource("configurationFromCbs.json").toURI()))),
+                JsonObject.class);
         CbsConfiguration cbsConfiguration = new CbsConfiguration();
 
         cbsConfiguration.parseCBSConfig(cbsConfigJson);
@@ -66,5 +65,8 @@ class CbsConfigurationTest {
         assertThat(cbsConfiguration.getMessageRouterPublishRequest()).isNotNull();
         assertThat(cbsConfiguration.getMessageRouterSubscribeRequest()).isNotNull();
         assertThat(cbsConfiguration.getMessageRouterUpdatePublishRequest()).isNotNull();
+
     }
-}
\ No newline at end of file
+
+   
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfigTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfigTest.java
index 22b82e3d..bd7d7779 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfigTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/configuration/KafkaConfigTest.java
@@ -19,38 +19,118 @@
  */
 package org.onap.dcaegen2.services.prh.configuration;
 
-import org.junit.jupiter.api.BeforeEach;
+import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable;
+import static java.lang.ClassLoader.getSystemResource;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
+import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.kafka.core.ConsumerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
 
 @ExtendWith(MockitoExtension.class)
 public class KafkaConfigTest {
 
-    @InjectMocks
-    KafkaConfig kafkaConfig;
-
-    @BeforeEach
-    void setUp() {
-        kafkaConfig.kafkaBoostrapServerConfig = "0.0.0.0";
-        kafkaConfig.groupIdConfig = "consumer-test";
-        kafkaConfig.kafkaSecurityProtocol = "test";
-        kafkaConfig.kafkaSaslMechanism = "test";
-        kafkaConfig.kafkaUsername = "test";
-        kafkaConfig.kafkaPassword = "test";
-        kafkaConfig.kafkaJaasConfig = null;
-        kafkaConfig.kafkaLoginModuleClassConfig = "test";
-        kafkaConfig.kafkaJaasConfig = "test";
-    }
+    KafkaConfig kafkaConfig = new KafkaConfig();
+
+     CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode = new
+     CbsConfigurationForAutoCommitDisabledMode();
+
+   
+//    @BeforeEach
+//    void setUp() {
+//        kafkaConfig.kafkaBoostrapServerConfig = "0.0.0.0";
+//        kafkaConfig.groupIdConfig = "consumer-test";
+//        kafkaConfig.kafkaSecurityProtocol = "test";
+//        kafkaConfig.kafkaSaslMechanism = "test";
+//        kafkaConfig.kafkaUsername = "test";
+//        kafkaConfig.kafkaPassword = "test";
+//        kafkaConfig.kafkaJaasConfig = null;
+//        kafkaConfig.kafkaLoginModuleClassConfig = "test";
+//        kafkaConfig.kafkaJaasConfig = "test";
+//    }
 
     @Test
-    public void consumerFactoryTest(){
-        kafkaConfig.consumerFactory();
+    void beforecbsConfigurationForAutoCommitDisabledMode() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            this.consumerFactoryTest();
+        });
+    }
+
+    void consumerFactoryTest() throws Exception {
+        JsonObject cbsConfigJsonForAutoCommitDisabled = new Gson().fromJson(
+                new String(Files.readAllBytes(
+                        Paths.get(getSystemResource("autoCommitDisabledConfigurationFromCbs2.json").toURI()))),
+                JsonObject.class);
+        cbsConfigurationForAutoCommitDisabledMode.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+        ConsumerFactory<String, String> consumerFactory = kafkaConfig
+                .consumerFactory(cbsConfigurationForAutoCommitDisabledMode);
+
+        String expectedKafkaBoostrapServerConfig = "onap-strimzi-kafka-bootstrap:9092";
+        String actualKafkaBoostrapServerConfig = consumerFactory.getConfigurationProperties().get("bootstrap.servers")
+                .toString();
+
+        String expectedGroupIdConfig = "OpenDCAE-c12";
+        String actualGroupIdConfig = consumerFactory.getConfigurationProperties().get("group.id").toString();
+
+        String expectedKafkaSecurityProtocol = "SASL_PLAINTEXT";
+        String actualKafkaSecurityProtocol = consumerFactory.getConfigurationProperties().get("security.protocol")
+                .toString();
+
+        String expectedKafkaSaslMechanism = "SCRAM-SHA-512";
+        String actualKafkaSaslMechanism = consumerFactory.getConfigurationProperties().get("sasl.mechanism").toString();
+
+        String expectedKafkaJaasConfig = "jaas_config";
+        String actualKafkaJaasConfig = consumerFactory.getConfigurationProperties().get("sasl.jaas.config").toString();
+
+        String expectedKeyDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
+        String actualKeyDeserializer = consumerFactory.getConfigurationProperties().get("key.deserializer").toString();
+
+        String expectedValueDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
+        String actualValueDeserializer = consumerFactory.getConfigurationProperties().get("value.deserializer")
+                .toString();
+
+        String expectedEnableAutoCommit = "false";
+        String actualEnableAutoCommit = consumerFactory.getConfigurationProperties().get("enable.auto.commit")
+                .toString();
+
+        assertEquals(expectedKafkaBoostrapServerConfig, actualKafkaBoostrapServerConfig,
+                "Expected value of KafKaBoostrapServerConfig is not matching with actual value");
+        assertEquals(expectedGroupIdConfig, actualGroupIdConfig,
+                "Expected value of GroupIdConfig is not matching with actual value");
+        assertEquals(expectedKafkaSecurityProtocol, actualKafkaSecurityProtocol,
+                "Expected value of KafkaSecurityProtocol is not matching with actual value");
+        assertEquals(expectedKafkaSaslMechanism, actualKafkaSaslMechanism,
+                "Expected value of KafkaSaslMechanism is not matching with actual value");
+        assertEquals(expectedKafkaJaasConfig, actualKafkaJaasConfig,
+                "Expected value of KafkaJaasConfig is not matching with actual value");
+        assertEquals(expectedKeyDeserializer, actualKeyDeserializer,
+                "Expected value of KeyDeserializer is not matching with actual value");
+        assertEquals(expectedValueDeserializer, actualValueDeserializer,
+                "Expected value of ValueDeserializer is not matching with actual value");
+        assertEquals(expectedEnableAutoCommit, actualEnableAutoCommit,
+                "Expected value of EnableAutoCommit is not matching with actual value");
+
     }
 
     @Test
-    public void kafkaListenerContainerFactoryTest(){
-        kafkaConfig.kafkaListenerContainerFactory();
+    void beforeKafkaListenerContainerFactoryTest() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            this.kafkaListenerContainerFactoryTest();
+        });
+    }
+
+    public void kafkaListenerContainerFactoryTest() throws Exception {
+        JsonObject cbsConfigJsonForAutoCommitDisabled = new Gson().fromJson(
+                new String(Files.readAllBytes(
+                        Paths.get(getSystemResource("autoCommitDisabledConfigurationFromCbs2.json").toURI()))),
+                JsonObject.class);
+        cbsConfigurationForAutoCommitDisabledMode.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+        kafkaConfig.kafkaListenerContainerFactory(cbsConfigurationForAutoCommitDisabledMode);
     }
 }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/AppInfoControllerTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/AppInfoControllerTest.java
index 1dba66a1..40b42480 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/AppInfoControllerTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/AppInfoControllerTest.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.
@@ -21,19 +22,28 @@
 package org.onap.dcaegen2.services.prh.controllers;
 
 import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
+import org.onap.dcaegen2.services.prh.configuration.KafkaConfig;
 import org.onap.dcaegen2.services.prh.configuration.PrhAppConfig;
+import org.onap.dcaegen2.services.prh.tasks.DmaapConsumerTaskImpl;
+import org.onap.dcaegen2.services.prh.tasks.ScheduledTasks;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Profile;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.http.MediaType;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.IfProfileValue;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.web.reactive.server.WebTestClient;
-
 import static org.mockito.Mockito.when;
 
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @DirtiesContext
+@ActiveProfiles(value = "prod")
 class AppInfoControllerTest {
 
     private static final String SAMPLE_GIT_INFO_CONTENT = "{ \"git.commit.id\" : \"37444e\" }";
@@ -44,17 +54,15 @@ class AppInfoControllerTest {
     @Autowired
     private WebTestClient webTestClient;
 
+    @MockBean
+    private ScheduledTasks scheduledTasks;
+
     @Test
     void shouldProvideHeartbeatResponse() {
-        webTestClient
-                .get().uri("/heartbeat")
-                .accept(MediaType.TEXT_PLAIN)
-                .exchange()
-                .expectStatus().isOk()
+        webTestClient.get().uri("/heartbeat").accept(MediaType.TEXT_PLAIN).exchange().expectStatus().isOk()
                 .expectBody(String.class).isEqualTo("alive");
     }
 
-
     @Test
     void shouldProvideVersionInfo() {
         when(prhAppConfig.getGitInfo()).thenReturn(new ByteArrayResource(SAMPLE_GIT_INFO_CONTENT.getBytes()));
@@ -66,4 +74,4 @@ class AppInfoControllerTest {
                 .expectStatus().isOk()
                 .expectBody(String.class).isEqualTo(SAMPLE_GIT_INFO_CONTENT);
     }
-}
\ No newline at end of file
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabledTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabledTest.java
new file mode 100644
index 00000000..43dcadf9
--- /dev/null
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerForAutoCommitDisabledTest.java
@@ -0,0 +1,96 @@
+/*
+ * ============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 org.junit.Ignore;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.prh.configuration.KafkaConfig;
+import org.onap.dcaegen2.services.prh.tasks.commit.KafkaConsumerTaskImpl;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.HttpStatus;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.web.reactive.server.WebTestClient;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@DirtiesContext(classMode = ClassMode.AFTER_CLASS)
+@ActiveProfiles(value = "autoCommitDisabled")
+class ScheduleControllerForAutoCommitDisabledTest {
+
+    @MockBean
+    private ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit;
+    
+    @MockBean
+    private KafkaConfig kafkaConfig;
+    
+    @MockBean
+    private ConsumerFactory<String, String> consumerFactory;
+    
+    @MockBean
+    private ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory;
+    
+    @MockBean
+    private KafkaConsumerTaskImpl kafkaConsumerTaskImpl;
+    
+    @Autowired
+    private WebTestClient webTestClient;
+
+    
+   @Test
+   void startEndpointShouldAllowStartingPrhTasks() {
+        when(scheduledTasksRunnerWithCommit.tryToStartTaskWithCommit()).thenReturn(true);
+        webTestClient
+                .get().uri("/start")
+                .exchange()
+                .expectStatus().isCreated()
+                .expectBody(String.class).isEqualTo("PRH Service has been started!");
+    }
+
+    @Test
+    void whenPrhTasksAreAlreadyStarted_shouldRespondThatRequestWasNotAccepted() {
+        when(scheduledTasksRunnerWithCommit.tryToStartTaskWithCommit()).thenReturn(false);
+        webTestClient
+                .get().uri("/start")
+                .exchange()
+                .expectStatus().isEqualTo(HttpStatus.NOT_ACCEPTABLE)
+                .expectBody(String.class).isEqualTo("PRH Service is already running!");
+    }
+
+    @Test
+    void stopEndpointShouldAllowStoppingPrhTasks() {
+        webTestClient
+                .get().uri("/stopPrh")
+                .exchange()
+                .expectStatus().isOk()
+                .expectBody(String.class).isEqualTo("PRH Service has been stopped!");
+
+        verify(scheduledTasksRunnerWithCommit).cancelTasks();
+    }
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerTest.java
index bbc6b968..92a527b7 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/controllers/ScheduleControllerTest.java
@@ -23,18 +23,21 @@ package org.onap.dcaegen2.services.prh.controllers;
 
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.services.prh.tasks.ScheduledTasksRunner;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.http.HttpStatus;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.IfProfileValue;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.web.reactive.server.WebTestClient;
-
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @DirtiesContext
+@ActiveProfiles("prod")
 class ScheduleControllerTest {
 
     @MockBean
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationForAutoCommitDisabledTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationForAutoCommitDisabledTest.java
new file mode 100644
index 00000000..b10c1ad8
--- /dev/null
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationForAutoCommitDisabledTest.java
@@ -0,0 +1,193 @@
+/*
+ * ============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.integration;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import com.jayway.jsonpath.JsonPath;
+import reactor.core.publisher.Flux;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.prh.MainApp;
+import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.adapter.kafka.ImmutableKafkaConfiguration;
+import org.onap.dcaegen2.services.prh.adapter.kafka.KafkaConfiguration;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
+import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
+import org.onap.dcaegen2.services.prh.tasks.commit.KafkaConsumerTaskImpl;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksWithCommit;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.configurationprocessor.json.JSONException;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.cloud.contract.wiremock.AutoConfigureWireMock;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
+import static com.github.tomakehurst.wiremock.client.WireMock.ok;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.anyRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.matchingJsonPath;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static com.github.tomakehurst.wiremock.client.WireMock.patch;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import static java.lang.ClassLoader.getSystemResource;
+import static org.mockito.Mockito.when;
+
+/**
+ *  * @author <a href="mailto:PRANIT.KAPDULE@t-systems.com">Pranit Kapdule</a> on
+ *   *        24/08/23
+ *    */
+
+@SpringBootTest
+@AutoConfigureWireMock(port = 0)
+@ActiveProfiles(value = "autoCommitDisabled")
+class PrhWorkflowIntegrationForAutoCommitDisabledTest {
+
+    @Autowired
+    private ScheduledTasksWithCommit scheduledTasksWithCommit;
+
+    @MockBean
+    private ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit; // just to disable scheduling - some
+                                                                           // configurability in ScheduledTaskRunner not
+                                                                           // to start tasks at app startup would be
+                                                                           // welcome
+
+    @MockBean
+    private KafkaConsumerTaskImpl kafkaConsumerTaskImpl;
+
+    @Autowired
+    private DmaapConsumerJsonParser dmaapConsumerJsonParser;
+
+    @Configuration
+    @Import(MainApp.class)
+    static class CbsConfigTestConfig {
+
+        @Value("http://localhost:${wiremock.server.port}")
+        private String wiremockServerAddress;
+
+        protected KafkaConfiguration kafkaConfiguration;
+
+        @Bean
+        public CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode() {
+
+            JsonObject cbsConfigJson = new Gson()
+                    .fromJson(getResourceContent("autoCommitDisabledConfigurationFromCbs2.json")
+                            .replaceAll("https?://dmaap-mr[\\w.]*:\\d+", wiremockServerAddress)
+                            .replaceAll("https?://aai[\\w.]*:\\d+", wiremockServerAddress), JsonObject.class);
+
+            CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabledMode = new CbsConfigurationForAutoCommitDisabledMode();
+
+            try {
+                cbsConfigurationForAutoCommitDisabledMode.parseCBSConfig(cbsConfigJson);
+            } catch (Exception e) {
+               //Exception is expected as environment variable for JAAS_CONFIG is not available
+                if (e.getMessage() == "kafkaJaasConfig") {
+                    kafkaConfiguration = new ImmutableKafkaConfiguration.Builder().kafkaBoostrapServerConfig("0.0.0.0")
+                            .groupIdConfig("CG1").kafkaSaslMechanism("SASL_MECHANISM")
+                            .kafkaSecurityProtocol("SEC-PROTOCOL").kafkaJaasConfig("JAAS_CONFIG").build();
+                    cbsConfigurationForAutoCommitDisabledMode.setKafkaConfiguration(kafkaConfiguration);
+
+                }
+
+            }
+            return cbsConfigurationForAutoCommitDisabledMode;
+        };
+
+    }
+
+    @BeforeEach
+    void resetWireMock() {
+        WireMock.reset();
+    }
+
+    @Test
+    void beforeCbsConfigurationForAutoCommitDisabledMode() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            this.whenThereAreNoEventsInDmaap_WorkflowShouldFinish();
+        });
+    }
+
+    void whenThereAreNoEventsInDmaap_WorkflowShouldFinish() throws JSONException {
+
+        when(kafkaConsumerTaskImpl.execute()).thenReturn(Flux.empty());
+       
+        scheduledTasksWithCommit.scheduleKafkaPrhEventTask();
+
+        verify(0, anyRequestedFor(urlPathMatching("/aai.*")));
+        verify(0, postRequestedFor(urlPathMatching("/events.*")));
+    }
+
+    @Test
+    void beforeWhenThereIsAnEventsInDmaap_ShouldSendPnfReadyNotification() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            this.whenThereIsAnEventsInDmaap_ShouldSendPnfReadyNotification();
+        });
+    }
+
+    void whenThereIsAnEventsInDmaap_ShouldSendPnfReadyNotification()
+            throws JSONException, JsonMappingException, JsonProcessingException {
+
+        String event = getResourceContent("integration/event.json");
+        String pnfName = JsonPath.read(event, "$.event.commonEventHeader.sourceName");
+
+        java.util.List<String> eventList = new ArrayList<>();
+        eventList.add(event);
+
+        Flux<ConsumerDmaapModel> fluxList = dmaapConsumerJsonParser
+                .getConsumerDmaapModelFromKafkaConsumerRecord(eventList);
+
+        stubFor(get(urlEqualTo("/aai/v23/network/pnfs/pnf/" + pnfName)).willReturn(ok().withBody("{}")));
+        stubFor(patch(urlEqualTo("/aai/v23/network/pnfs/pnf/" + pnfName)));
+        stubFor(post(urlEqualTo("/events/unauthenticated.PNF_READY")));
+
+        when(kafkaConsumerTaskImpl.execute()).thenReturn(fluxList);
+
+        scheduledTasksWithCommit.scheduleKafkaPrhEventTask();
+
+        verify(1, postRequestedFor(urlEqualTo("/events/unauthenticated.PNF_READY"))
+                .withRequestBody(matchingJsonPath("$[0].correlationId", equalTo(pnfName))));
+
+    }
+
+    private static String getResourceContent(String resourceName) {
+        try {
+            return new String(Files.readAllBytes(Paths.get(getSystemResource(resourceName).toURI())));
+        } catch (Exception e) {
+            throw new RuntimeException("failed loading content of '" + resourceName + "'", e);
+        }
+    }
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationTest.java
index 1a6c76c4..f5033ca2 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowIntegrationTest.java
@@ -39,6 +39,7 @@ import org.springframework.cloud.contract.wiremock.AutoConfigureWireMock;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
@@ -52,22 +53,15 @@ import static com.github.tomakehurst.wiremock.client.WireMock.matchingJsonPath;
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.verify;
 import static com.github.tomakehurst.wiremock.client.WireMock.patch;
-
-
-
-
-
-
 import java.nio.file.Files;
 import java.nio.file.Paths;
-
-
 import static java.lang.ClassLoader.getSystemResource;
 import static java.util.Collections.singletonList;
 
 
 @SpringBootTest
 @AutoConfigureWireMock(port = 0)
+@ActiveProfiles(value = "prod")
 class PrhWorkflowIntegrationTest {
 
     @Autowired
@@ -95,6 +89,8 @@ class PrhWorkflowIntegrationTest {
             cbsConfiguration.parseCBSConfig(cbsConfigJson);
             return cbsConfiguration;
         }
+        
+
     }
 
     @BeforeEach
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationForAutoCommitDisabledTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationForAutoCommitDisabledTest.java
new file mode 100644
index 00000000..a623e24c
--- /dev/null
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationForAutoCommitDisabledTest.java
@@ -0,0 +1,100 @@
+/*
+ * ============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.integration;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.mockito.stubbing.Answer;
+import org.onap.dcaegen2.services.prh.configuration.KafkaConfig;
+import org.onap.dcaegen2.services.prh.tasks.commit.KafkaConsumerTask;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksRunnerWithCommit;
+import org.onap.dcaegen2.services.prh.tasks.commit.ScheduledTasksWithCommit;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.TestPropertySource;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.doAnswer;
+
+
+@SpringBootTest
+@TestPropertySource (properties = {"prh.workflow-scheduling-interval=20ms","spring.main.allow-bean-definition-overriding=true"})
+@DirtiesContext
+@ActiveProfiles(value = "autoCommitDisabled")
+@Disabled
+class PrhWorkflowSchedulingIntegrationForAutoCommitDisabledTest {
+
+    private static final int EXPECTED_INVOCATIONS_NUMBER = 1;
+    private static final int REMAINING_INVOCATIONS_NUMBER = 0;
+    
+    @MockBean
+    private ScheduledTasksRunnerWithCommit scheduledTasksRunnerWithCommit;
+    
+    @MockBean
+    private ScheduledTasksWithCommit scheduledTasksWithCommit;
+    
+    @MockBean
+    private KafkaConsumerTask kafkaConsumerTask;
+   
+    
+    @MockBean
+    private KafkaConfig kafkaConfig;
+    
+    @MockBean
+    private ConsumerFactory<String, String> consumerFactory;
+    
+    @MockBean
+    private ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory;
+    
+    private CountDownLatch invocationLatch;
+    
+    @Test
+    void prhWorkflowShouldBeExecutedRightAfterApplicationStart() {
+        try {
+            
+            invocationLatch = new CountDownLatch(EXPECTED_INVOCATIONS_NUMBER);
+            doAnswer(registerInvocation(invocationLatch)).when(scheduledTasksWithCommit).scheduleKafkaPrhEventTask();
+            assertThatMethodWasInvokedOnce();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void assertThatMethodWasInvokedOnce() throws InterruptedException {
+        boolean awaitResult = invocationLatch.await(1, TimeUnit.SECONDS);
+        System.out.println("###awaitResult="+awaitResult);
+        assertEquals(REMAINING_INVOCATIONS_NUMBER, invocationLatch.getCount());
+    }
+
+    private static Answer registerInvocation(CountDownLatch invocationLatch) {
+        return invocation -> {
+            System.out.println("###before countDown:"+invocationLatch.getCount());
+            invocationLatch.countDown();
+            System.out.println("###after countDown:"+invocationLatch.getCount());
+            return null;
+        };
+    }
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationTest.java
index 939dd2a3..8fb952d0 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/integration/PrhWorkflowSchedulingIntegrationTest.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.
@@ -27,14 +28,17 @@ import org.mockito.stubbing.Answer;
 import org.onap.dcaegen2.services.prh.tasks.ScheduledTasks;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.TestPropertySource;
-
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.doAnswer;
 
 
 @SpringBootTest
 @TestPropertySource (properties = {"prh.workflow-scheduling-interval=20ms"})
+@ActiveProfiles(value = "prod")
+@DirtiesContext
 class PrhWorkflowSchedulingIntegrationTest {
 
     private static final int EXPECTED_INVOCATIONS_NUMBER = 1;
@@ -61,4 +65,4 @@ class PrhWorkflowSchedulingIntegrationTest {
             return null;
         };
     }
-}
\ No newline at end of file
+}
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImplTest.java
index c23a1886..f33ff43e 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/commit/KafkaConsumerTaskImplTest.java
@@ -18,30 +18,35 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.dcaegen2.services.prh.tasks;
+package org.onap.dcaegen2.services.prh.tasks.commit;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.record.TimestampType;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.onap.dcaegen2.services.prh.adapter.aai.api.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.adapter.aai.api.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfigurationForAutoCommitDisabledMode;
 import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.prh.tasks.commit.EpochDateTimeConversion;
-import org.onap.dcaegen2.services.prh.tasks.commit.KafkaConsumerTaskImpl;
-import org.springframework.boot.configurationprocessor.json.JSONException;
 import org.springframework.kafka.support.Acknowledgment;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonSyntaxException;
 import reactor.core.publisher.Flux;
-
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
-
-import static org.mockito.Mockito.when;
+import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable;
+import static java.lang.ClassLoader.getSystemResource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 @ExtendWith(MockitoExtension.class)
 public class KafkaConsumerTaskImplTest {
@@ -49,38 +54,92 @@ public class KafkaConsumerTaskImplTest {
     @Mock
     private Acknowledgment acknowledgment;
 
-    @Mock
+    private KafkaConsumerTaskImpl kafkaConsumerTask;
+
     private DmaapConsumerJsonParser dmaapConsumerJsonParser;
 
-    @Mock
     private EpochDateTimeConversion epochDateTimeConversion;
 
-    @InjectMocks
-    private KafkaConsumerTaskImpl kafkaConsumerTask;
+    private CbsConfigurationForAutoCommitDisabledMode cbsConfigurationForAutoCommitDisabled;
+
+    private JsonObject cbsConfigJsonForAutoCommitDisabled;
+
+    @BeforeEach
+    void beforeEach() throws JsonSyntaxException, IOException, URISyntaxException {
+        cbsConfigJsonForAutoCommitDisabled = new Gson().fromJson(
+                new String(Files.readAllBytes(
+                        Paths.get(getSystemResource("autoCommitDisabledConfigurationFromCbs2.json").toURI()))),
+                JsonObject.class);
+        cbsConfigurationForAutoCommitDisabled = new CbsConfigurationForAutoCommitDisabledMode();
+        dmaapConsumerJsonParser = new DmaapConsumerJsonParser();
+        epochDateTimeConversion = new EpochDateTimeConversion();
+
+    }
 
     @Test
-    public void onMessageTest(){
-        List<ConsumerRecord<String, String>> list = new ArrayList<>();
-        TimestampType timestampType = null;
-        Headers headers = new RecordHeaders();
-        epochDateTimeConversion.setDaysForRecords("3");
-        ConsumerRecord<String, String> records = new ConsumerRecord<>
-                ("test-topic", 1, 1l, 0l, timestampType, 1, 1, "test-key", "test-value", headers
-        , null);
-        list.add(records);
-        kafkaConsumerTask.onMessage(list, acknowledgment);
+    void beforeOnMessageTest() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            cbsConfigurationForAutoCommitDisabled.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+            kafkaConsumerTask = new KafkaConsumerTaskImpl(cbsConfigurationForAutoCommitDisabled,
+                    dmaapConsumerJsonParser, epochDateTimeConversion);
+            List<ConsumerRecord<String, String>> list = new ArrayList<>();
+            TimestampType timestampType = null;
+            Headers headers = new RecordHeaders();
+            epochDateTimeConversion.setDaysForRecords("3");
+            ConsumerRecord<String, String> records = new ConsumerRecord<>("test-topic", 1, 1l, 0l, timestampType, 1, 1,
+                    "test-key", "test-value", headers, null);
+            list.add(records);
+            kafkaConsumerTask.onMessage(list, acknowledgment);
+            String actualTopicInList = list.get(0).topic();
+            String expectedTopicInList = "test-topic";
+            assertEquals(expectedTopicInList, actualTopicInList, "topic is not matching");
+            assertThat(kafkaConsumerTask.getOffset().equals(acknowledgment));
+            assertThat(kafkaConsumerTask.getJsonEvent().contains("test-topic"));
+        });
     }
 
     @Test
-    public void commitOffsetTest(){
-        kafkaConsumerTask.commitOffset();
+    void beforeCommitOffsetTest() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            cbsConfigurationForAutoCommitDisabled.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+            kafkaConsumerTask = new KafkaConsumerTaskImpl(cbsConfigurationForAutoCommitDisabled,
+                    dmaapConsumerJsonParser, epochDateTimeConversion);
+            kafkaConsumerTask.commitOffset();
+        });
     }
 
     @Test
-    public void executeTest() throws JSONException {
-        List<String> jsonEvent = new ArrayList<>();
-        ConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder().correlationId("123").build();
-        when(dmaapConsumerJsonParser.getConsumerDmaapModelFromKafkaConsumerRecord(jsonEvent)).thenReturn(Flux.just(consumerDmaapModel));
-        kafkaConsumerTask.execute();
+    void beforeExecuteTest() throws Exception {
+        withEnvironmentVariable("JAAS_CONFIG", "jaas_config").execute(() -> {
+            cbsConfigurationForAutoCommitDisabled.parseCBSConfig(cbsConfigJsonForAutoCommitDisabled);
+            kafkaConsumerTask = new KafkaConsumerTaskImpl(cbsConfigurationForAutoCommitDisabled,
+                    dmaapConsumerJsonParser, epochDateTimeConversion);
+            String event = getResourceContent("integration/event.json");
+            java.util.List<String> eventList = new ArrayList<>();
+            eventList.add(event);
+            kafkaConsumerTask.setJsonEvent(eventList);
+            Flux<ConsumerDmaapModel> flux = kafkaConsumerTask.execute();
+            String expectedSourceName = "NOK6061ZW8";
+            String actualSourceName = flux.blockFirst().getCorrelationId();
+
+            String expectedOamV4IpAddress = "val3";
+            String actualOamV4IpAddress = flux.blockFirst().getIpv4();
+
+            String expectedOamV6IpAddress = "val4";
+            String actualOamV6IpAddress = flux.blockFirst().getIpv6();
+
+            assertEquals(expectedSourceName, actualSourceName, "SourceName is not matching");
+            assertEquals(expectedOamV4IpAddress, actualOamV4IpAddress, "OamV4IpAddress is not matching");
+            assertEquals(expectedOamV6IpAddress, actualOamV6IpAddress, "OamV6IpAddress is not matching");
+        });
+    }
+
+    private static String getResourceContent(String resourceName) {
+        try {
+            return new String(Files.readAllBytes(Paths.get(getSystemResource(resourceName).toURI())));
+        } catch (Exception e) {
+            throw new RuntimeException("failed loading content of '" + resourceName + "'", e);
+        }
     }
+
 }
diff --git a/prh-app-server/src/test/resources/autoCommitDisabledConfigurationFromCbs2.json b/prh-app-server/src/test/resources/autoCommitDisabledConfigurationFromCbs2.json
new file mode 100644
index 00000000..6e58e6a5
--- /dev/null
+++ b/prh-app-server/src/test/resources/autoCommitDisabledConfigurationFromCbs2.json
@@ -0,0 +1,66 @@
+{
+    "config": {
+        "aai.aaiClientConfiguration.aaiBasePath": "/aai/v23",
+        "aai.aaiClientConfiguration.aaiHeaders": {
+            "Accept": "application/json",
+            "Authorization": "Basic QUFJOkFBSQ==",
+            "Real-Time": "true",
+            "X-FromAppId": "prh",
+            "X-TransactionId": "9999"
+        },
+        "aai.aaiClientConfiguration.aaiHost": "aai-internal.onap.svc.cluster.local",
+        "aai.aaiClientConfiguration.aaiHostPortNumber": 80,
+        "aai.aaiClientConfiguration.aaiIgnoreSslCertificateErrors": true,
+        "aai.aaiClientConfiguration.aaiPnfPath": "/network/pnfs/pnf",
+        "aai.aaiClientConfiguration.aaiProtocol": "http",
+        "aai.aaiClientConfiguration.aaiServiceInstancePath": "/business/customers/customer/{{customer}}/service-subscriptions/service-subscription/{{serviceType}}/service-instances/service-instance/{{serviceInstanceId}}",
+        "aai.aaiClientConfiguration.aaiUserName": "AAI",
+        "aai.aaiClientConfiguration.aaiUserPassword": "AAI",
+        "aai.aaiClientConfiguration.baseUrl": "https://aai.onap.svc.cluster.local:8443/aai/v23",
+        "aai.aaiClientConfiguration.pnfUrl": "http://aai.onap.svc.cluster.local:8443/aai/v23/network/pnfs/pnf",
+        "dmaap.dmaapConsumerConfiguration.consumerGroup": "OpenDCAE-c12",
+        "dmaap.dmaapConsumerConfiguration.consumerId": "c12",
+        "dmaap.dmaapConsumerConfiguration.dmaapContentType": "application/json",
+        "dmaap.dmaapConsumerConfiguration.timeoutMs": -1,
+        "dmaap.dmaapProducerConfiguration.dmaapContentType": "application/json",
+        "dmaap.dmaapUpdateProducerConfiguration.dmaapContentType": "application/json",
+        "security.enableAaiCertAuth": false,
+        "security.enableDmaapCertAuth": false,
+        "security.keyStorePasswordPath": "/opt/app/prh/etc/cert/jks.pass",
+        "security.keyStorePath": "/opt/app/prh/etc/cert/cert.jks",
+        "security.trustStorePasswordPath": "/opt/app/prh/etc/cert/trust.pass",
+        "security.trustStorePath": "/opt/app/prh/etc/cert/trust.jks",
+        "streams_publishes": {
+            "pnf-ready": {
+                "dmaap_info": {
+                    "topic_url": "http://dmaap-mr:2222/events/unauthenticated.PNF_READY"
+                },
+                "type": "message_router"
+            },
+            "pnf-update": {
+                "dmaap_info": {
+                    "topic_url": "http://dmaap-mr:2222/events/unauthenticated.PNF_UPDATE"
+                },
+                "type": "message_router"
+            }
+        },
+        "streams_subscribes": {
+            "ves-reg-output": {
+                "dmaap_info": {
+                    "topic_url": "http://dmaap-mr:2222/events/unauthenticated.VES_PNFREG_OUTPUT"
+                },
+                "type": "message_router"
+            }
+        },
+        "kafka-configurations": {
+            "kafkaBoostrapServerConfig": "onap-strimzi-kafka-bootstrap:9092",
+            "groupIdConfig": "OpenDCAE-c12",
+            "kafkaSecurityProtocol": "SASL_PLAINTEXT",
+            "kafkaSaslMechanism": "SCRAM-SHA-512",
+            "kafkaUsername": "strimzi-kafka-admin-username",
+            "kafkaPassword": "strimzi-kafka-admin-password",
+            "kafkaJaasConfig": "jaas_config",
+            "Login_Module_Class": "MODULE-CLASS"
+        }
+    }
+}