Merge "DFC logging according to ONAP specification"

7 files changed, 129 insertions, 59 deletions

diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfiguration.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfiguration.java
index 9838afb1..208e8a43 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfiguration.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfiguration.java
@@ -1,6 +1,6 @@
 /*
  * ============LICENSE_START======================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. 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
@@ -17,15 +17,13 @@
 package org.onap.dcaegen2.collectors.datafile.configuration;
 
 import com.google.gson.JsonObject;
-
+import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
-
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
-
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.ReactiveCloudConfigurationProvider;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -35,7 +33,6 @@ import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 import org.springframework.scheduling.annotation.EnableScheduling;
-
 import reactor.core.publisher.Flux;
 import reactor.core.scheduler.Schedulers;
 
@@ -64,8 +61,8 @@ public class CloudConfiguration extends AppConfig {
     }
 
 
-    protected void runTask() {
-        Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment)).subscribeOn(Schedulers.parallel())
+    protected void runTask(Map<String, String> contextMap) {
+        Flux.defer(() -> EnvironmentProcessor.evaluate(systemEnvironment, contextMap)).subscribeOn(Schedulers.parallel())
                 .subscribe(this::parsingConfigSuccess, this::parsingConfigError);
     }
 
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/EnvironmentProcessor.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/EnvironmentProcessor.java
index 106725ce..8f417261 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/EnvironmentProcessor.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/EnvironmentProcessor.java
@@ -1,7 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
+ * ============LICENSE_START========================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. 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
@@ -13,14 +13,16 @@
  * 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=========================================================
+ * ============LICENSE_END==========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.configuration;
 
+import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
 import org.onap.dcaegen2.collectors.datafile.exceptions.EnvironmentLoaderException;
+import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.ImmutableEnvProperties;
 import org.slf4j.Logger;
@@ -38,7 +40,8 @@ class EnvironmentProcessor {
     private EnvironmentProcessor() {
     }
 
-    static Mono<EnvProperties> evaluate(Properties systemEnvironment) {
+    static Mono<EnvProperties> evaluate(Properties systemEnvironment, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.info("Loading configuration from system environment variables");
         EnvProperties envProperties;
         try {
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java
index bc21f96c..b4dc6353 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java
@@ -16,22 +16,30 @@
 
 package org.onap.dcaegen2.collectors.datafile.configuration;
 
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.INVOCATION_ID;
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.REQUEST_ID;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ScheduledFuture;
-
 import javax.annotation.PostConstruct;
-
+import org.apache.commons.lang3.StringUtils;
+import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables;
 import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.annotation.EnableScheduling;
-
 import io.swagger.annotations.ApiOperation;
 import reactor.core.publisher.Mono;
 
@@ -45,7 +53,11 @@ public class SchedulerConfig {
     private static final Duration SCHEDULING_DELAY_FOR_DATAFILE_COLLECTOR_TASKS = Duration.ofSeconds(15);
     private static final Duration SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY = Duration.ofMinutes(5);
     private static final Duration SCHEDULING_DELAY_FOR_DATAFILE_PURGE_CACHE = Duration.ofHours(1);
+    private static final Logger logger = LoggerFactory.getLogger(SchedulerConfig.class);
+    private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY");
+    private static final Marker EXIT = MarkerFactory.getMarker("EXIT");
     private static volatile List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>();
+    private Map<String, String> contextMap;
 
     private final TaskScheduler taskScheduler;
     private final ScheduledTasks scheduledTask;
@@ -68,6 +80,9 @@ public class SchedulerConfig {
     public synchronized Mono<ResponseEntity<String>> getResponseFromCancellationOfTasks() {
         scheduledFutureList.forEach(x -> x.cancel(false));
         scheduledFutureList.clear();
+        MdcVariables.setMdcContextMap(contextMap);
+        logger.info(EXIT, "Stopped Datafile workflow");
+        MDC.clear();
         return Mono.defer(() -> Mono
             .just(new ResponseEntity<>("Datafile Service has already been stopped!", HttpStatus.CREATED)));
     }
@@ -80,10 +95,20 @@ public class SchedulerConfig {
     @PostConstruct
     @ApiOperation(value = "Start task if possible")
     public synchronized boolean tryToStartTask() {
+        String requestId = MDC.get(REQUEST_ID);
+        if (StringUtils.isBlank(requestId)) {
+            MDC.put(REQUEST_ID, UUID.randomUUID().toString());
+        }
+        String invocationId = MDC.get(INVOCATION_ID);
+        if (StringUtils.isBlank(invocationId)) {
+            MDC.put(INVOCATION_ID, UUID.randomUUID().toString());
+        }
+        contextMap = MDC.getCopyOfContextMap();
+        logger.info(ENTRY, "Start scheduling Datafile workflow");
         if (scheduledFutureList.isEmpty()) {
-            scheduledFutureList.add(taskScheduler.scheduleAtFixedRate(cloudConfiguration::runTask, Instant.now(),
+            scheduledFutureList.add(taskScheduler.scheduleAtFixedRate(() -> cloudConfiguration.runTask(contextMap), Instant.now(),
                     SCHEDULING_REQUEST_FOR_CONFIGURATION_DELAY));
-            scheduledFutureList.add(taskScheduler.scheduleWithFixedDelay(scheduledTask::scheduleMainDatafileEventTask,
+            scheduledFutureList.add(taskScheduler.scheduleWithFixedDelay(() -> scheduledTask.scheduleMainDatafileEventTask(contextMap),
                     SCHEDULING_DELAY_FOR_DATAFILE_COLLECTOR_TASKS));
             scheduledFutureList.add(taskScheduler.scheduleWithFixedDelay(() -> scheduledTask.purgeCachedInformation(Instant.now()),
                    SCHEDULING_DELAY_FOR_DATAFILE_PURGE_CACHE));
@@ -94,4 +119,4 @@ public class SchedulerConfig {
         }
 
     }
-}
+}
\ No newline at end of file
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleController.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleController.java
index 825308e7..a21ed041 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleController.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleController.java
@@ -18,16 +18,24 @@
 
 package org.onap.dcaegen2.collectors.datafile.controllers;
 
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.INVOCATION_ID;
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.REQUEST_ID;
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.X_INVOCATION_ID;
+import static org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables.X_ONAP_REQUEST_ID;
+import java.util.UUID;
+import org.apache.commons.lang3.StringUtils;
 import org.onap.dcaegen2.collectors.datafile.configuration.SchedulerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RequestHeader;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
-
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import reactor.core.publisher.Mono;
@@ -50,9 +58,24 @@ public class ScheduleController {
         this.schedulerConfig = schedulerConfig;
     }
 
+    public Mono<ResponseEntity<String>> startTasks() {
+        logger.trace("Start scheduling worker request");
+        return Mono.fromSupplier(schedulerConfig::tryToStartTask).map(this::createStartTaskResponse);
+    }
+
     @RequestMapping(value = "start", method = RequestMethod.GET)
     @ApiOperation(value = "Start scheduling worker request")
-    public Mono<ResponseEntity<String>> startTasks() {
+    public Mono<ResponseEntity<String>> startTasks(@RequestHeader HttpHeaders headers) {
+        String requestId = headers.getFirst(X_ONAP_REQUEST_ID);
+        if (StringUtils.isBlank(requestId)) {
+            requestId = UUID.randomUUID().toString();
+        }
+        String invocationId = headers.getFirst(X_INVOCATION_ID);
+        if (StringUtils.isBlank(invocationId)) {
+            invocationId = UUID.randomUUID().toString();
+        }
+        MDC.put(REQUEST_ID, requestId);
+        MDC.put(INVOCATION_ID, invocationId);
         logger.trace("Receiving start scheduling worker request");
         return Mono.fromSupplier(schedulerConfig::tryToStartTask).map(this::createStartTaskResponse);
     }
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
index 4c0dcce5..57edc364 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
@@ -17,16 +17,16 @@
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
 import java.time.Duration;
-
+import java.util.Map;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables;
 import org.onap.dcaegen2.collectors.datafile.service.HttpUtils;
 import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerReactiveHttpClient;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
-
 import reactor.core.publisher.Mono;
 
 /**
@@ -50,21 +50,24 @@ public class DataRouterPublisher {
      * @param firstBackoffTimeout the time to delay the first retry
      * @return the HTTP response status as a string
      */
-    public Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel model, long numRetries, Duration firstBackoff) {
+    public Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel model, long numRetries, Duration firstBackoff,
+            Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Method called with arg {}", model);
         DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient = resolveClient();
 
         //@formatter:off
         return Mono.just(model)
                 .cache()
-                .flatMap(dmaapProducerReactiveHttpClient::getDmaapProducerResponse)
-                .flatMap(httpStatus -> handleHttpResponse(httpStatus, model))
+                .flatMap(m -> dmaapProducerReactiveHttpClient.getDmaapProducerResponse(m, contextMap))
+                .flatMap(httpStatus -> handleHttpResponse(httpStatus, model, contextMap))
                 .retryBackoff(numRetries, firstBackoff);
         //@formatter:on
     }
 
-    private Mono<ConsumerDmaapModel> handleHttpResponse(HttpStatus response, ConsumerDmaapModel model) {
-
+    private Mono<ConsumerDmaapModel> handleHttpResponse(HttpStatus response, ConsumerDmaapModel model,
+            Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         if (HttpUtils.isSuccessfulResponseCode(response.value())) {
             logger.trace("Publish to DR successful!");
             return Mono.just(model);
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java
index 0b647bf5..a0020318 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java
@@ -18,7 +18,7 @@ package org.onap.dcaegen2.collectors.datafile.tasks;
 
 import java.nio.file.Path;
 import java.time.Duration;
-
+import java.util.Map;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
@@ -29,9 +29,9 @@ import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
+import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import reactor.core.publisher.Mono;
 
 /**
@@ -52,14 +52,15 @@ public class FileCollector {
     }
 
     public Mono<ConsumerDmaapModel> execute(FileData fileData, MessageMetaData metaData, long maxNumberOfRetries,
-            Duration firstBackoffTimeout) {
+            Duration firstBackoffTimeout, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Entering execute with {}", fileData);
         resolveKeyStore();
 
         //@formatter:off
         return Mono.just(fileData)
             .cache()
-            .flatMap(fd -> collectFile(fileData, metaData))
+            .flatMap(fd -> collectFile(fileData, metaData, contextMap))
             .retryBackoff(maxNumberOfRetries, firstBackoffTimeout);
         //@formatter:on
     }
@@ -76,7 +77,9 @@ public class FileCollector {
         ftpsClient.setTrustedCAPassword(ftpesConfig.trustedCAPassword());
     }
 
-    private Mono<ConsumerDmaapModel> collectFile(FileData fileData, MessageMetaData metaData) {
+    private Mono<ConsumerDmaapModel> collectFile(FileData fileData, MessageMetaData metaData,
+            Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.trace("starting to collectFile");
 
         final String remoteFile = fileData.remoteFilePath();
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
index 783c699c..89ebde8f 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
@@ -23,8 +23,8 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
-
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.ftp.FtpsClient;
 import org.onap.dcaegen2.collectors.datafile.ftp.SftpClient;
@@ -32,12 +32,13 @@ import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
+import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables;
 import org.onap.dcaegen2.collectors.datafile.service.PublishedFileCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
@@ -86,22 +87,24 @@ public class ScheduledTasks {
     /**
      * Main function for scheduling for the file collection Workflow.
      */
-    public void scheduleMainDatafileEventTask() {
+    public void scheduleMainDatafileEventTask(Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Execution of tasks was registered");
         applicationConfiguration.initFileStreamReader();
-        createMainTask().subscribe(this::onSuccess, this::onError, this::onComplete);
+        createMainTask(contextMap).subscribe(model -> onSuccess(model, contextMap), thr -> onError(thr, contextMap),
+                () -> onComplete(contextMap));
     }
 
-    Flux<ConsumerDmaapModel> createMainTask() {
+    Flux<ConsumerDmaapModel> createMainTask(Map<String, String> contextMap) {
         return fetchMoreFileReadyMessages() //
                 .parallel(getParallelism()) // Each FileReadyMessage in a separate thread
                 .runOn(scheduler) //
                 .flatMap(this::createFileCollectionTask) //
                 .filter(this::shouldBePublished) //
                 .doOnNext(fileData -> currentNumberOfTasks.incrementAndGet()) //
-                .flatMap(this::collectFileFromXnf) //
-                .flatMap(this::publishToDataRouter) //
-                .doOnNext(model -> deleteFile(Paths.get(model.getInternalLocation()))) //
+                .flatMap(fileData -> collectFileFromXnf(fileData, contextMap)) //
+                .flatMap(model -> publishToDataRouter(model, contextMap)) //
+                .doOnNext(model -> deleteFile(Paths.get(model.getInternalLocation()), contextMap)) //
                 .doOnNext(model -> currentNumberOfTasks.decrementAndGet()) //
                 .sequential();
     }
@@ -113,15 +116,18 @@ public class ScheduledTasks {
         alreadyPublishedFiles.purge(now);
     }
 
-    private void onComplete() {
+    private void onComplete(Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.info("Datafile tasks have been completed");
     }
 
-    private void onSuccess(ConsumerDmaapModel model) {
+    private void onSuccess(ConsumerDmaapModel model, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.info("Datafile consumed tasks {}", model.getInternalLocation());
     }
 
-    private void onError(Throwable throwable) {
+    private void onError(Throwable throwable, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.error("Chain of tasks have been aborted due to errors in Datafile workflow {}", throwable);
     }
 
@@ -147,38 +153,45 @@ public class ScheduledTasks {
         return alreadyPublishedFiles.put(task.fileData.getLocalFileName()) == null;
     }
 
-    private Mono<ConsumerDmaapModel> collectFileFromXnf(FileCollectionData fileCollect) {
+    private Mono<ConsumerDmaapModel> collectFileFromXnf(FileCollectionData fileCollect,
+            Map<String, String> contextMap) {
         final long maxNUmberOfRetries = 3;
         final Duration initialRetryTimeout = Duration.ofSeconds(5);
 
+        MdcVariables.setMdcContextMap(contextMap);
         return fileCollect.collectorTask
-                .execute(fileCollect.fileData, fileCollect.metaData, maxNUmberOfRetries, initialRetryTimeout)
-                .onErrorResume(exception -> handleCollectFailure(fileCollect.fileData));
+                .execute(fileCollect.fileData, fileCollect.metaData, maxNUmberOfRetries, initialRetryTimeout,
+                        contextMap)
+                .onErrorResume(exception -> handleCollectFailure(fileCollect.fileData, contextMap));
     }
 
-    private Mono<ConsumerDmaapModel> handleCollectFailure(FileData fileData) {
+    private Mono<ConsumerDmaapModel> handleCollectFailure(FileData fileData, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         Path localFileName = fileData.getLocalFileName();
         logger.error("File fetching failed: {}", localFileName);
-        deleteFile(localFileName);
+        deleteFile(localFileName, contextMap);
         alreadyPublishedFiles.remove(localFileName);
         currentNumberOfTasks.decrementAndGet();
         return Mono.empty();
     }
 
-    private Mono<ConsumerDmaapModel> publishToDataRouter(ConsumerDmaapModel model) {
+    private Mono<ConsumerDmaapModel> publishToDataRouter(ConsumerDmaapModel model, Map<String, String> contextMap) {
         final long maxNumberOfRetries = 3;
         final Duration initialRetryTimeout = Duration.ofSeconds(5);
 
         DataRouterPublisher publisherTask = createDataRouterPublisher();
 
-        return publisherTask.execute(model, maxNumberOfRetries, initialRetryTimeout)
-                .onErrorResume(exception -> handlePublishFailure(model, exception));
+        MdcVariables.setMdcContextMap(contextMap);
+        return publisherTask.execute(model, maxNumberOfRetries, initialRetryTimeout, contextMap)
+                .onErrorResume(exception -> handlePublishFailure(model, exception, contextMap));
     }
 
-    private Mono<ConsumerDmaapModel> handlePublishFailure(ConsumerDmaapModel model, Throwable exception) {
+    private Mono<ConsumerDmaapModel> handlePublishFailure(ConsumerDmaapModel model, Throwable exception,
+            Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.error("File publishing failed: {}, exception: {}", model.getName(), exception);
         Path internalFileName = Paths.get(model.getInternalLocation());
-        deleteFile(internalFileName);
+        deleteFile(internalFileName, contextMap);
         alreadyPublishedFiles.remove(internalFileName);
         currentNumberOfTasks.decrementAndGet();
         return Mono.empty();
@@ -193,17 +206,20 @@ public class ScheduledTasks {
             return Flux.empty();
         }
 
+        Map<String, String> contextMap = MDC.getCopyOfContextMap();
         return createConsumerTask() //
                 .execute() //
-                .onErrorResume(this::handleConsumeMessageFailure);
+                .onErrorResume(exception -> handleConsumeMessageFailure(exception, contextMap));
     }
 
-    private Flux<FileReadyMessage> handleConsumeMessageFailure(Throwable exception) {
-        logger.error("Polling for file ready message filed, exception: {}", exception);
+    private Flux<FileReadyMessage> handleConsumeMessageFailure(Throwable exception, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
+        logger.error("Polling for file ready message failed, exception: {}", exception);
         return Flux.empty();
     }
 
-    private void deleteFile(Path localFile) {
+    private void deleteFile(Path localFile, Map<String, String> contextMap) {
+        MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Deleting file: {}", localFile);
         try {
             Files.delete(localFile);