Merge "Thread safety issues"

21 files changed, 436 insertions, 782 deletions

diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java
index 40de33dd..82c390f7 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java
@@ -2,39 +2,44 @@
  * ============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
+ * 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
+ * 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.
+ * 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.collectors.datafile.configuration;
 
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonSyntaxException;
+import com.google.gson.TypeAdapterFactory;
+
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ServiceLoader;
+
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+
 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 java.io.*;
-import java.util.ServiceLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.stereotype.Component;
-import javax.validation.constraints.NotEmpty;
-import javax.validation.constraints.NotNull;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import com.google.gson.JsonSyntaxException;
-import com.google.gson.TypeAdapterFactory;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
@@ -55,49 +60,53 @@ public class AppConfig {
     private static final String SECURITY = "security";
     private static final Logger logger = LoggerFactory.getLogger(AppConfig.class);
 
-    DmaapConsumerConfiguration dmaapConsumerConfiguration;
-
-    DmaapPublisherConfiguration dmaapPublisherConfiguration;
-
-    FtpesConfig ftpesConfig;
+    private DmaapConsumerConfiguration dmaapConsumerConfiguration;
+    private DmaapPublisherConfiguration dmaapPublisherConfiguration;
+    private FtpesConfig ftpesConfiguration;
 
     @NotEmpty
     private String filepath;
 
-    public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
+    public synchronized DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
         return dmaapConsumerConfiguration;
     }
 
-    public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
+    public synchronized DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
         return dmaapPublisherConfiguration;
     }
 
-    public FtpesConfig getFtpesConfiguration() {
-        return ftpesConfig;
+    public synchronized FtpesConfig getFtpesConfiguration() {
+        return ftpesConfiguration;
     }
 
-    public void initFileStreamReader() {
+    public void loadConfigurationFromFile() {
 
         GsonBuilder gsonBuilder = new GsonBuilder();
         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
         JsonParser parser = new JsonParser();
         JsonObject jsonObject;
-        try (InputStream inputStream = getInputStream(filepath)) {
+        try (InputStream inputStream = createInputStream(filepath)) {
             JsonElement rootElement = getJsonElement(parser, inputStream);
             if (rootElement.isJsonObject()) {
                 jsonObject = rootElement.getAsJsonObject();
-                ftpesConfig = deserializeType(gsonBuilder,
+                FtpesConfig ftpesConfig = deserializeType(gsonBuilder,
                         jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(FTP).getAsJsonObject(FTPES_CONFIGURATION),
                         FtpesConfig.class);
-                dmaapConsumerConfiguration = deserializeType(gsonBuilder, concatenateJsonObjects(
-                        jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_CONSUMER),
-                        rootElement.getAsJsonObject().getAsJsonObject(CONFIG).getAsJsonObject(SECURITY)),
+                DmaapConsumerConfiguration consumerConfiguration = deserializeType(gsonBuilder,
+                        concatenateJsonObjects(
+                                jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP)
+                                        .getAsJsonObject(DMAAP_CONSUMER),
+                                rootElement.getAsJsonObject().getAsJsonObject(CONFIG).getAsJsonObject(SECURITY)),
                         DmaapConsumerConfiguration.class);
 
-                dmaapPublisherConfiguration = deserializeType(gsonBuilder, concatenateJsonObjects(
-                        jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP).getAsJsonObject(DMAAP_PRODUCER),
-                        rootElement.getAsJsonObject().getAsJsonObject(CONFIG).getAsJsonObject(SECURITY)),
+                DmaapPublisherConfiguration publisherConfiguration = deserializeType(gsonBuilder,
+                        concatenateJsonObjects(
+                                jsonObject.getAsJsonObject(CONFIG).getAsJsonObject(DMAAP)
+                                        .getAsJsonObject(DMAAP_PRODUCER),
+                                rootElement.getAsJsonObject().getAsJsonObject(CONFIG).getAsJsonObject(SECURITY)),
                         DmaapPublisherConfiguration.class);
+
+                setConfiguration(consumerConfiguration, publisherConfiguration, ftpesConfig);
             }
         } catch (IOException e) {
             logger.error("Problem with file loading, file: {}", filepath, e);
@@ -106,30 +115,36 @@ public class AppConfig {
         }
     }
 
+    synchronized void setConfiguration(DmaapConsumerConfiguration consumerConfiguration,
+            DmaapPublisherConfiguration publisherConfiguration, FtpesConfig ftpesConfig) {
+        this.dmaapConsumerConfiguration = consumerConfiguration;
+        this.dmaapPublisherConfiguration = publisherConfiguration;
+        this.ftpesConfiguration = ftpesConfig;
+    }
+
     JsonElement getJsonElement(JsonParser parser, InputStream inputStream) {
         return parser.parse(new InputStreamReader(inputStream));
     }
 
     private <T> T deserializeType(@NotNull GsonBuilder gsonBuilder, @NotNull JsonObject jsonObject,
-                                  @NotNull Class<T> type) {
+            @NotNull Class<T> type) {
         return gsonBuilder.create().fromJson(jsonObject, type);
     }
 
-    InputStream getInputStream(@NotNull String filepath) throws IOException {
+    InputStream createInputStream(@NotNull String filepath) throws IOException {
         return new BufferedInputStream(new FileInputStream(filepath));
     }
 
-    String getFilepath() {
+    synchronized String getFilepath() {
         return this.filepath;
     }
 
-    public void setFilepath(String filepath) {
+    public synchronized void setFilepath(String filepath) {
         this.filepath = filepath;
     }
 
     private JsonObject concatenateJsonObjects(JsonObject target, JsonObject source) {
-        source.entrySet()
-                .forEach(entry -> target.add(entry.getKey(), entry.getValue()));
+        source.entrySet().forEach(entry -> target.add(entry.getKey(), entry.getValue()));
         return target;
     }
 
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 208e8a43..0254597e 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
@@ -56,7 +56,7 @@ public class CloudConfiguration extends AppConfig {
     private Properties systemEnvironment;
 
     @Autowired
-    public void setThreadPoolTaskScheduler(ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
+    public synchronized void setThreadPoolTaskScheduler(ReactiveCloudConfigurationProvider reactiveCloudConfigurationProvider) {
         this.reactiveCloudConfigurationProvider = reactiveCloudConfigurationProvider;
     }
 
@@ -80,7 +80,7 @@ public class CloudConfiguration extends AppConfig {
                 .subscribe(this::parseCloudConfig, this::cloudConfigError);
     }
 
-    private void parseCloudConfig(JsonObject jsonObject) {
+    private synchronized void parseCloudConfig(JsonObject jsonObject) {
         logger.info("Received application configuration: {}", jsonObject);
         CloudConfigParser cloudConfigParser = new CloudConfigParser(jsonObject);
         dmaapPublisherCloudConfiguration = cloudConfigParser.getDmaapPublisherConfig();
@@ -89,17 +89,17 @@ public class CloudConfiguration extends AppConfig {
     }
 
     @Override
-    public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
+    public synchronized DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
         return Optional.ofNullable(dmaapPublisherCloudConfiguration).orElse(super.getDmaapPublisherConfiguration());
     }
 
     @Override
-    public DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
+    public synchronized DmaapConsumerConfiguration getDmaapConsumerConfiguration() {
         return Optional.ofNullable(dmaapConsumerCloudConfiguration).orElse(super.getDmaapConsumerConfiguration());
     }
 
     @Override
-    public FtpesConfig getFtpesConfiguration() {
+    public synchronized FtpesConfig getFtpesConfiguration() {
         return Optional.ofNullable(ftpesCloudConfiguration).orElse(super.getFtpesConfiguration());
     }
 }
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 a0020318..af4670e3 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
@@ -17,6 +17,7 @@
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.Map;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
@@ -40,22 +41,16 @@ import reactor.core.publisher.Mono;
 public class FileCollector {
 
     private static final Logger logger = LoggerFactory.getLogger(FileCollector.class);
-    private AppConfig datafileAppConfig;
-    private final FtpsClient ftpsClient;
-    private final SftpClient sftpClient;
+    private final AppConfig datafileAppConfig;
 
-
-    public FileCollector(AppConfig datafileAppConfig, FtpsClient ftpsClient, SftpClient sftpClient) {
+    public FileCollector(AppConfig datafileAppConfig) {
         this.datafileAppConfig = datafileAppConfig;
-        this.ftpsClient = ftpsClient;
-        this.sftpClient = sftpClient;
     }
 
     public Mono<ConsumerDmaapModel> execute(FileData fileData, MessageMetaData metaData, long maxNumberOfRetries,
             Duration firstBackoffTimeout, Map<String, String> contextMap) {
         MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Entering execute with {}", fileData);
-        resolveKeyStore();
 
         //@formatter:off
         return Mono.just(fileData)
@@ -65,18 +60,6 @@ public class FileCollector {
         //@formatter:on
     }
 
-    private FtpesConfig resolveConfiguration() {
-        return datafileAppConfig.getFtpesConfiguration();
-    }
-
-    private void resolveKeyStore() {
-        FtpesConfig ftpesConfig = resolveConfiguration();
-        ftpsClient.setKeyCertPath(ftpesConfig.keyCert());
-        ftpsClient.setKeyCertPassword(ftpesConfig.keyPassword());
-        ftpsClient.setTrustedCAPath(ftpesConfig.trustedCA());
-        ftpsClient.setTrustedCAPassword(ftpesConfig.trustedCAPassword());
-    }
-
     private Mono<ConsumerDmaapModel> collectFile(FileData fileData, MessageMetaData metaData,
             Map<String, String> contextMap) {
         MdcVariables.setMdcContextMap(contextMap);
@@ -85,11 +68,8 @@ public class FileCollector {
         final String remoteFile = fileData.remoteFilePath();
         final Path localFile = fileData.getLocalFileName();
 
-        try {
+        try (FileCollectClient currentClient = createClient(fileData)) {
             localFile.getParent().toFile().mkdir(); // Create parent directories
-
-            FileCollectClient currentClient = selectClient(fileData);
-
             currentClient.collectFile(remoteFile, localFile);
             return Mono.just(getConsumerDmaapModel(fileData, metaData, localFile));
         } catch (Exception throwable) {
@@ -98,12 +78,12 @@ public class FileCollector {
         }
     }
 
-    private FileCollectClient selectClient(FileData fileData) throws DatafileTaskException {
+    private FileCollectClient createClient(FileData fileData) throws DatafileTaskException {
         switch (fileData.scheme()) {
             case SFTP:
-                return sftpClient;
+                return createSftpClient(fileData);
             case FTPS:
-                return ftpsClient;
+                return createFtpsClient(fileData);
             default:
                 throw new DatafileTaskException("Unhandeled protocol: " + fileData.scheme());
         }
@@ -129,4 +109,17 @@ public class FileCollector {
                 .build();
         // @formatter:on
     }
+
+    SftpClient createSftpClient(FileData fileData) throws DatafileTaskException {
+        SftpClient client = new SftpClient(fileData.fileServerData());
+        client.open();
+        return client;
+    }
+
+    FtpsClient createFtpsClient(FileData fileData) throws DatafileTaskException {
+        FtpesConfig config = datafileAppConfig.getFtpesConfiguration();
+        FtpsClient client = new FtpsClient(fileData.fileServerData());
+        client.open(config.keyCert(), config.keyPassword(), Paths.get(config.trustedCA()), config.trustedCAPassword());
+        return client;
+    }
 }
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 89ebde8f..28963377 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
@@ -26,8 +26,6 @@ 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;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
@@ -57,12 +55,10 @@ public class ScheduledTasks {
     /** Data needed for fetching of one file */
     private class FileCollectionData {
         final FileData fileData;
-        final FileCollector collectorTask;
         final MessageMetaData metaData;
 
-        FileCollectionData(FileData fd, FileCollector collectorTask, MessageMetaData metaData) {
+        FileCollectionData(FileData fd, MessageMetaData metaData) {
             this.fileData = fd;
-            this.collectorTask = collectorTask;
             this.metaData = metaData;
         }
     }
@@ -90,7 +86,7 @@ public class ScheduledTasks {
     public void scheduleMainDatafileEventTask(Map<String, String> contextMap) {
         MdcVariables.setMdcContextMap(contextMap);
         logger.trace("Execution of tasks was registered");
-        applicationConfiguration.initFileStreamReader();
+        applicationConfiguration.loadConfigurationFromFile();
         createMainTask(contextMap).subscribe(model -> onSuccess(model, contextMap), thr -> onError(thr, contextMap),
                 () -> onComplete(contextMap));
     }
@@ -143,8 +139,7 @@ public class ScheduledTasks {
         List<FileCollectionData> fileCollects = new ArrayList<>();
 
         for (FileData fileData : availableFiles.files()) {
-            fileCollects.add(
-                    new FileCollectionData(fileData, createFileCollector(fileData), availableFiles.messageMetaData()));
+            fileCollects.add(new FileCollectionData(fileData, availableFiles.messageMetaData()));
         }
         return Flux.fromIterable(fileCollects);
     }
@@ -159,7 +154,7 @@ public class ScheduledTasks {
         final Duration initialRetryTimeout = Duration.ofSeconds(5);
 
         MdcVariables.setMdcContextMap(contextMap);
-        return fileCollect.collectorTask
+        return createFileCollector()
                 .execute(fileCollect.fileData, fileCollect.metaData, maxNUmberOfRetries, initialRetryTimeout,
                         contextMap)
                 .onErrorResume(exception -> handleCollectFailure(fileCollect.fileData, contextMap));
@@ -236,9 +231,8 @@ public class ScheduledTasks {
         return new DMaaPMessageConsumerTask(this.applicationConfiguration);
     }
 
-    FileCollector createFileCollector(FileData fileData) {
-        return new FileCollector(applicationConfiguration, new FtpsClient(fileData.fileServerData()),
-                new SftpClient(fileData.fileServerData()));
+    FileCollector createFileCollector() {
+        return new FileCollector(applicationConfiguration);
     }
 
     DataRouterPublisher createDataRouterPublisher() {
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
index 443ddae7..2c136304 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
@@ -47,16 +47,15 @@ import org.onap.dcaegen2.collectors.datafile.integration.junit5.mockito.MockitoE
  */
 @ExtendWith({MockitoExtension.class})
 class AppConfigTest {
-    
+
     private static final String DATAFILE_ENDPOINTS = "datafile_endpoints.json";
     private static final boolean CORRECT_JSON = true;
     private static final boolean INCORRECT_JSON = false;
 
     private static AppConfig appConfigUnderTest;
 
-
-    private static String filePath = Objects
-            .requireNonNull(AppConfigTest.class.getClassLoader().getResource(DATAFILE_ENDPOINTS)).getFile();
+    private static String filePath =
+            Objects.requireNonNull(AppConfigTest.class.getClassLoader().getResource(DATAFILE_ENDPOINTS)).getFile();
 
     @BeforeEach
     public void setUp() {
@@ -70,7 +69,7 @@ class AppConfigTest {
 
         // Then
         verify(appConfigUnderTest, times(1)).setFilepath(anyString());
-        verify(appConfigUnderTest, times(0)).initFileStreamReader();
+        verify(appConfigUnderTest, times(0)).loadConfigurationFromFile();
         Assertions.assertEquals(filePath, appConfigUnderTest.getFilepath());
     }
 
@@ -82,15 +81,12 @@ class AppConfigTest {
 
         // When
         appConfigUnderTest.setFilepath(filePath);
-        doReturn(inputStream).when(appConfigUnderTest).getInputStream(any());
-        appConfigUnderTest.initFileStreamReader();
-        appConfigUnderTest.dmaapConsumerConfiguration = appConfigUnderTest.getDmaapConsumerConfiguration();
-        appConfigUnderTest.dmaapPublisherConfiguration = appConfigUnderTest.getDmaapPublisherConfiguration();
-        appConfigUnderTest.ftpesConfig = appConfigUnderTest.getFtpesConfiguration();
+        doReturn(inputStream).when(appConfigUnderTest).createInputStream(any());
+        appConfigUnderTest.loadConfigurationFromFile();
 
         // Then
         verify(appConfigUnderTest, times(1)).setFilepath(anyString());
-        verify(appConfigUnderTest, times(1)).initFileStreamReader();
+        verify(appConfigUnderTest, times(1)).loadConfigurationFromFile();
         Assertions.assertNotNull(appConfigUnderTest.getDmaapConsumerConfiguration());
         Assertions.assertNotNull(appConfigUnderTest.getDmaapPublisherConfiguration());
         Assertions.assertEquals(appConfigUnderTest.getDmaapPublisherConfiguration(),
@@ -98,7 +94,6 @@ class AppConfigTest {
         Assertions.assertEquals(appConfigUnderTest.getDmaapConsumerConfiguration(),
                 appConfigUnderTest.getDmaapConsumerConfiguration());
         Assertions.assertEquals(appConfigUnderTest.getFtpesConfiguration(), appConfigUnderTest.getFtpesConfiguration());
-
     }
 
     @Test
@@ -108,11 +103,11 @@ class AppConfigTest {
         appConfigUnderTest.setFilepath(filePath);
 
         // When
-        appConfigUnderTest.initFileStreamReader();
+        appConfigUnderTest.loadConfigurationFromFile();
 
         // Then
         verify(appConfigUnderTest, times(1)).setFilepath(anyString());
-        verify(appConfigUnderTest, times(1)).initFileStreamReader();
+        verify(appConfigUnderTest, times(1)).loadConfigurationFromFile();
         Assertions.assertNull(appConfigUnderTest.getDmaapConsumerConfiguration());
         Assertions.assertNull(appConfigUnderTest.getDmaapPublisherConfiguration());
         Assertions.assertNull(appConfigUnderTest.getFtpesConfiguration());
@@ -127,15 +122,15 @@ class AppConfigTest {
 
         // When
         appConfigUnderTest.setFilepath(filePath);
-        doReturn(inputStream).when(appConfigUnderTest).getInputStream(any());
-        appConfigUnderTest.initFileStreamReader();
+        doReturn(inputStream).when(appConfigUnderTest).createInputStream(any());
+        appConfigUnderTest.loadConfigurationFromFile();
 
         // Then
         verify(appConfigUnderTest, times(1)).setFilepath(anyString());
-        verify(appConfigUnderTest, times(1)).initFileStreamReader();
-        Assertions.assertNotNull(appConfigUnderTest.getDmaapConsumerConfiguration());
+        verify(appConfigUnderTest, times(1)).loadConfigurationFromFile();
+        Assertions.assertNull(appConfigUnderTest.getDmaapConsumerConfiguration());
         Assertions.assertNull(appConfigUnderTest.getDmaapPublisherConfiguration());
-        Assertions.assertNotNull(appConfigUnderTest.getFtpesConfiguration());
+        Assertions.assertNull(appConfigUnderTest.getFtpesConfiguration());
 
     }
 
@@ -147,18 +142,15 @@ class AppConfigTest {
                 new ByteArrayInputStream((getJsonConfig(CORRECT_JSON).getBytes(StandardCharsets.UTF_8)));
         // When
         appConfigUnderTest.setFilepath(filePath);
-        doReturn(inputStream).when(appConfigUnderTest).getInputStream(any());
+        doReturn(inputStream).when(appConfigUnderTest).createInputStream(any());
         JsonElement jsonElement = mock(JsonElement.class);
         when(jsonElement.isJsonObject()).thenReturn(false);
         doReturn(jsonElement).when(appConfigUnderTest).getJsonElement(any(JsonParser.class), any(InputStream.class));
-        appConfigUnderTest.initFileStreamReader();
-        appConfigUnderTest.dmaapConsumerConfiguration = appConfigUnderTest.getDmaapConsumerConfiguration();
-        appConfigUnderTest.dmaapPublisherConfiguration = appConfigUnderTest.getDmaapPublisherConfiguration();
-        appConfigUnderTest.ftpesConfig = appConfigUnderTest.getFtpesConfiguration();
+        appConfigUnderTest.loadConfigurationFromFile();
 
         // Then
         verify(appConfigUnderTest, times(1)).setFilepath(anyString());
-        verify(appConfigUnderTest, times(1)).initFileStreamReader();
+        verify(appConfigUnderTest, times(1)).loadConfigurationFromFile();
         Assertions.assertNull(appConfigUnderTest.getDmaapConsumerConfiguration());
         Assertions.assertNull(appConfigUnderTest.getDmaapPublisherConfiguration());
         Assertions.assertNull(appConfigUnderTest.getFtpesConfiguration());
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/XnfCollectorTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
index b5d3c159..c266d50e 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/XnfCollectorTaskImplTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
@@ -16,16 +16,21 @@
 
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
+
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
+
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
@@ -40,13 +45,14 @@ import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
+
 import reactor.test.StepVerifier;
 
 /**
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  *
  */
-public class XnfCollectorTaskImplTest {
+public class FileCollectorTest {
     private static final String PRODUCT_NAME = "NrRadio";
     private static final String VENDOR_NAME = "Ericsson";
     private static final String LAST_EPOCH_MICROSEC = "8745745764578";
@@ -70,7 +76,7 @@ public class XnfCollectorTaskImplTest {
     private static final String FTPES_LOCATION_NO_PORT =
             FTPES_SCHEME + USER + ":" + PWD + "@" + SERVER_ADDRESS + REMOTE_FILE_LOCATION;
     private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
-    private static final String SFTP_LOCATION_NO_PORT = SFTP_SCHEME + SERVER_ADDRESS +  REMOTE_FILE_LOCATION;
+    private static final String SFTP_LOCATION_NO_PORT = SFTP_SCHEME + SERVER_ADDRESS + REMOTE_FILE_LOCATION;
 
     private static final String GZIP_COMPRESSION = "gzip";
     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
@@ -104,7 +110,7 @@ public class XnfCollectorTaskImplTest {
         // @formatter:on
     }
 
-    private FileData createFileData(String location) {
+    private FileData createFileData(String location, Scheme scheme) {
         // @formatter:off
         return  ImmutableFileData.builder()
             .name(PM_FILE_NAME)
@@ -112,7 +118,7 @@ public class XnfCollectorTaskImplTest {
             .compression(GZIP_COMPRESSION)
             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
             .fileFormatVersion(FILE_FORMAT_VERSION)
-            .scheme(Scheme.FTPS)
+            .scheme(scheme)
             .build();
         // @formatter:on
     }
@@ -147,10 +153,10 @@ public class XnfCollectorTaskImplTest {
 
     @Test
     public void whenFtpesFile_returnCorrectResponse() throws Exception {
-        FileCollector collectorUndetTest =
-                new FileCollector(appConfigMock, ftpsClientMock, sftpClientMock);
+        FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+        doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
 
-        FileData fileData = createFileData(FTPES_LOCATION_NO_PORT);
+        FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS);
 
         ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT);
 
@@ -159,56 +165,43 @@ public class XnfCollectorTaskImplTest {
                 .expectNext(expectedConsumerDmaapModel).verifyComplete();
 
         verify(ftpsClientMock, times(1)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
-        verify(ftpsClientMock).setKeyCertPath(FTP_KEY_PATH);
-        verify(ftpsClientMock).setKeyCertPassword(FTP_KEY_PASSWORD);
-        verify(ftpsClientMock).setTrustedCAPath(TRUSTED_CA_PATH);
-        verify(ftpsClientMock).setTrustedCAPassword(TRUSTED_CA_PASSWORD);
+        verify(ftpsClientMock, times(1)).close();
+
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
     public void whenSftpFile_returnCorrectResponse() throws Exception {
-        FileCollector collectorUndetTest =
-                new FileCollector(appConfigMock, ftpsClientMock, sftpClientMock);
-        // @formatter:off
-        FileData fileData = ImmutableFileData.builder()
-                .name(PM_FILE_NAME)
-                .location(SFTP_LOCATION_NO_PORT)
-                .compression(GZIP_COMPRESSION)
-                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
-                .fileFormatVersion(FILE_FORMAT_VERSION)
-                .scheme(Scheme.SFTP)
-                .build();
+        FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+        doReturn(sftpClientMock).when(collectorUndetTest).createSftpClient(any());
 
-        ConsumerDmaapModel expectedConsumerDmaapModel = ImmutableConsumerDmaapModel.builder()
-                .productName(PRODUCT_NAME)
-                .vendorName(VENDOR_NAME)
-                .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
-                .sourceName(SOURCE_NAME)
-                .startEpochMicrosec(START_EPOCH_MICROSEC)
-                .timeZoneOffset(TIME_ZONE_OFFSET)
-                .name(PM_FILE_NAME)
-                .location(SFTP_LOCATION_NO_PORT)
-                .internalLocation(LOCAL_FILE_LOCATION.toString())
-                .compression(GZIP_COMPRESSION)
-                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
-                .fileFormatVersion(FILE_FORMAT_VERSION)
-                .build();
-        // @formatter:on
+        FileData fileData = createFileData(SFTP_LOCATION_NO_PORT, Scheme.SFTP);
+        ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION_NO_PORT);
 
         Map<String, String> contextMap = new HashMap<>();
+         StepVerifier.create(collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap))
+                .expectNext(expectedConsumerDmaapModel) //
+                .verifyComplete();
+
+        // The same again, but with port
+        fileData = createFileData(SFTP_LOCATION, Scheme.SFTP);
+        expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION);
+
         StepVerifier.create(collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap))
-                .expectNext(expectedConsumerDmaapModel).verifyComplete();
+                .expectNext(expectedConsumerDmaapModel) //
+                .verifyComplete();
 
-        verify(sftpClientMock, times(1)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+        verify(sftpClientMock, times(2)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+        verify(sftpClientMock, times(2)).close();
         verifyNoMoreInteractions(sftpClientMock);
     }
 
     @Test
     public void whenFtpesFileAlwaysFail_retryAndFail() throws Exception {
-        FileCollector collectorUndetTest =
-                new FileCollector(appConfigMock, ftpsClientMock, sftpClientMock);
-        FileData fileData = createFileData(FTPES_LOCATION);
+        FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+        doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
+
+        FileData fileData = createFileData(FTPES_LOCATION, Scheme.FTPS);
         doThrow(new DatafileTaskException("Unable to collect file.")).when(ftpsClientMock)
                 .collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
 
@@ -221,14 +214,14 @@ public class XnfCollectorTaskImplTest {
 
     @Test
     public void whenFtpesFileFailOnce_retryAndReturnCorrectResponse() throws Exception {
-        FileCollector collectorUndetTest =
-                new FileCollector(appConfigMock, ftpsClientMock, sftpClientMock);
+        FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+        doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
         doThrow(new DatafileTaskException("Unable to collect file.")).doNothing().when(ftpsClientMock)
                 .collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
 
         ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT);
 
-        FileData fileData = createFileData(FTPES_LOCATION_NO_PORT);
+        FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS);
         Map<String, String> contextMap = new HashMap<>();
         StepVerifier.create(collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap))
                 .expectNext(expectedConsumerDmaapModel).verifyComplete();
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
index 8a572be4..8c4b3891 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
@@ -82,7 +82,7 @@ public class ScheduledTasksTest {
 
         doReturn(dmaapPublisherConfiguration).when(appConfig).getDmaapPublisherConfiguration();
         doReturn(consumerMock).when(testedObject).createConsumerTask();
-        doReturn(fileCollectorMock).when(testedObject).createFileCollector(notNull());
+        doReturn(fileCollectorMock).when(testedObject).createFileCollector();
         doReturn(dataRouterMock).when(testedObject).createDataRouterPublisher();
     }
 
diff --git a/datafile-commons/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java b/datafile-commons/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java
index 442b766b..0eaa7a17 100644
--- a/datafile-commons/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java
+++ b/datafile-commons/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java
@@ -25,10 +25,6 @@ public class DatafileTaskException extends Exception {
 
     private static final long serialVersionUID = 1L;
 
-    public DatafileTaskException(Exception e) {
-        super(e);
-    }
-
     public DatafileTaskException(String message) {
         super(message);
     }
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java
index bedae43a..bca7dfd2 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java
@@ -23,7 +23,6 @@ import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 /**
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
-@FunctionalInterface
-public interface FileCollectClient {
+public interface FileCollectClient extends AutoCloseable  {
     public void collectFile(String remoteFile, Path localFile) throws DatafileTaskException;
 }
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java
index c3b7990f..1bf3ec5a 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java
@@ -22,26 +22,25 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Path;
-import java.nio.file.Paths;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
 import java.util.Optional;
 
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+
 import org.apache.commons.net.ftp.FTP;
 import org.apache.commons.net.ftp.FTPReply;
 import org.apache.commons.net.ftp.FTPSClient;
+import org.apache.commons.net.util.KeyManagerUtils;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
-import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
-import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
-import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
-import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils.KeyManagerException;
-import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
-import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
-import org.onap.dcaegen2.collectors.datafile.ssl.KeyManagerUtilsWrapper;
-import org.onap.dcaegen2.collectors.datafile.ssl.KeyStoreWrapper;
-import org.onap.dcaegen2.collectors.datafile.ssl.TrustManagerFactoryWrapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.core.io.FileSystemResource;
 
 /**
  * Gets file from xNF with FTPS protocol.
@@ -50,139 +49,40 @@ import org.slf4j.LoggerFactory;
  */
 public class FtpsClient implements FileCollectClient {
     private static final Logger logger = LoggerFactory.getLogger(FtpsClient.class);
-    private String keyCertPath;
-    private String keyCertPassword;
-    private Path trustedCAPath;
-    private String trustedCAPassword;
-
-    private FTPSClient realFtpsClient = new FTPSClient();
-    private IKeyManagerUtils keyManagerUtils = new KeyManagerUtilsWrapper();
-    private IKeyStore keyStore;
-    private ITrustManagerFactory trustManagerFactory;
-    private IFileSystemResource fileSystemResource = new FileSystemResourceWrapper();
-    private boolean keyManagerSet = false;
-    private boolean trustManagerSet = false;
+    FTPSClient realFtpsClient = new FTPSClient();
     private final FileServerData fileServerData;
-
+    private static TrustManager theTrustManager = null;
 
     public FtpsClient(FileServerData fileServerData) {
         this.fileServerData = fileServerData;
     }
 
-    @Override
-    public void collectFile(String remoteFile, Path localFile) throws DatafileTaskException {
-        logger.trace("collectFile called");
-
+    public void open(String keyCertPath, String keyCertPassword, Path trustedCAPath, String trustedCAPassword)
+            throws DatafileTaskException {
         try {
             realFtpsClient.setNeedClientAuth(true);
-            setUpKeyManager(realFtpsClient);
-            setUpTrustedCA(realFtpsClient);
-            setUpConnection(realFtpsClient);
-            getFileFromxNF(realFtpsClient, remoteFile, localFile);
-        } catch (IOException e) {
-            logger.trace("", e);
+            realFtpsClient.setKeyManager(createKeyManager(keyCertPath, keyCertPassword));
+            realFtpsClient.setTrustManager(getTrustManager(trustedCAPath, trustedCAPassword));
+            setUpConnection();
+        } catch (DatafileTaskException e) {
+            throw e;
+        } catch (Exception e) {
             throw new DatafileTaskException("Could not open connection: ", e);
-        } catch (KeyManagerException e) {
-            logger.trace("", e);
-            throw new DatafileTaskException(e);
-        } finally {
-            closeDownConnection(realFtpsClient);
-        }
-        logger.trace("collectFile fetched: {}", localFile);
-    }
-
-    private void setUpKeyManager(FTPSClient ftps) throws KeyManagerException {
-        if (keyManagerSet) {
-            logger.trace("keyManager already set!");
-        } else {
-            keyManagerUtils.setCredentials(keyCertPath, keyCertPassword);
-            ftps.setKeyManager(keyManagerUtils.getClientKeyManager());
-            keyManagerSet = true;
-        }
-        logger.trace("complete setUpKeyManager");
-    }
-
-    private void setUpTrustedCA(FTPSClient ftps) throws DatafileTaskException {
-        if (trustManagerSet) {
-            logger.trace("trustManager already set!");
-        } else {
-            try {
-                fileSystemResource.setPath(trustedCAPath);
-                InputStream fis = fileSystemResource.getInputStream();
-                IKeyStore ks = getKeyStore();
-                ks.load(fis, trustedCAPassword.toCharArray());
-                fis.close();
-                ITrustManagerFactory tmf = getTrustManagerFactory();
-                tmf.init(ks.getKeyStore());
-                ftps.setTrustManager(tmf.getTrustManagers()[0]);
-                trustManagerSet = true;
-            } catch (Exception e) {
-                throw new DatafileTaskException("Unable to trust xNF's CA, " + trustedCAPath + " " + e);
-            }
         }
-        logger.trace("complete setUpTrustedCA");
     }
 
-    private int getPort(Optional<Integer> port) {
-        final int FTPS_DEFAULT_PORT = 21;
-        return port.isPresent() ? port.get() : FTPS_DEFAULT_PORT;
-    }
-
-    private void setUpConnection(FTPSClient ftps) throws DatafileTaskException, IOException {
-        if (!ftps.isConnected()) {
-            ftps.connect(fileServerData.serverAddress(), getPort(fileServerData.port()));
-            logger.trace("after ftp connect");
-
-            if (!ftps.login(fileServerData.userId(), fileServerData.password())) {
-                throw new DatafileTaskException("Unable to log in to xNF. " + fileServerData.serverAddress());
-            }
-
-            if (FTPReply.isPositiveCompletion(ftps.getReplyCode())) {
-                ftps.enterLocalPassiveMode();
-                ftps.setFileType(FTP.BINARY_FILE_TYPE);
-                // Set protection buffer size
-                ftps.execPBSZ(0);
-                // Set data channel protection to private
-                ftps.execPROT("P");
-                ftps.setBufferSize(1024 * 1024);
-            } else {
-                throw new DatafileTaskException("Unable to connect to xNF. " + fileServerData.serverAddress()
-                        + " xNF reply code: " + ftps.getReplyCode());
-            }
-        }
-        logger.trace("setUpConnection successfully!");
-    }
-
-    private void getFileFromxNF(FTPSClient ftps, String remoteFileName, Path localFileName)
-            throws IOException {
-        logger.trace("starting to getFile");
-
-        File localFile = localFileName.toFile();
-        if (localFile.createNewFile()) {
-            logger.warn("Local file {} already created", localFileName);
-        }
-        OutputStream output = new FileOutputStream(localFile);
-        logger.trace("begin to retrieve from xNF.");
-        if (!ftps.retrieveFile(remoteFileName, output)) {
-            throw new IOException("Could not retrieve file");
-        }
-        logger.trace("end retrieve from xNF.");
-        output.close();
-        logger.debug("File {} Download Successfull from xNF", localFileName);
-    }
-
-
-    private void closeDownConnection(FTPSClient ftps) {
+    @Override
+    public void close() {
         logger.trace("starting to closeDownConnection");
-        if (ftps != null && ftps.isConnected()) {
+        if (realFtpsClient.isConnected()) {
             try {
-                boolean logOut = ftps.logout();
+                boolean logOut = realFtpsClient.logout();
                 logger.trace("logOut: {}", logOut);
             } catch (Exception e) {
                 logger.trace("Unable to logout connection.", e);
             }
             try {
-                ftps.disconnect();
+                realFtpsClient.disconnect();
                 logger.trace("disconnected!");
             } catch (Exception e) {
                 logger.trace("Unable to disconnect connection.", e);
@@ -190,54 +90,89 @@ public class FtpsClient implements FileCollectClient {
         }
     }
 
-    public void setKeyCertPath(String keyCertPath) {
-        this.keyCertPath = keyCertPath;
-    }
+    @Override
+    public void collectFile(String remoteFileName, Path localFileName) throws DatafileTaskException {
+        logger.trace("collectFile called");
 
-    public void setKeyCertPassword(String keyCertPassword) {
-        this.keyCertPassword = keyCertPassword;
+        try (OutputStream output = createOutputStream(localFileName)) {
+            logger.trace("begin to retrieve from xNF.");
+            if (!realFtpsClient.retrieveFile(remoteFileName, output)) {
+                throw new DatafileTaskException("Could not retrieve file " + remoteFileName);
+            }
+        } catch (IOException e) {
+            throw new DatafileTaskException("Could not fetch file: ", e);
+        }
+        logger.trace("collectFile fetched: {}", localFileName);
     }
 
-    public void setTrustedCAPath(String trustedCAPath) {
-        this.trustedCAPath = Paths.get(trustedCAPath);
+    private int getPort(Optional<Integer> port) {
+        final int FTPS_DEFAULT_PORT = 21;
+        return port.isPresent() ? port.get() : FTPS_DEFAULT_PORT;
     }
 
-    public void setTrustedCAPassword(String trustedCAPassword) {
-        this.trustedCAPassword = trustedCAPassword;
-    }
+    private void setUpConnection() throws DatafileTaskException, IOException {
 
-    private ITrustManagerFactory getTrustManagerFactory() throws NoSuchAlgorithmException {
-        if (trustManagerFactory == null) {
-            trustManagerFactory = new TrustManagerFactoryWrapper();
-        }
-        return trustManagerFactory;
-    }
+        realFtpsClient.connect(fileServerData.serverAddress(), getPort(fileServerData.port()));
+        logger.trace("after ftp connect");
 
-    private IKeyStore getKeyStore() throws KeyStoreException {
-        if (keyStore == null) {
-            keyStore = new KeyStoreWrapper();
+        if (!realFtpsClient.login(fileServerData.userId(), fileServerData.password())) {
+            throw new DatafileTaskException("Unable to log in to xNF. " + fileServerData.serverAddress());
         }
 
-        return keyStore;
-    }
+        if (FTPReply.isPositiveCompletion(realFtpsClient.getReplyCode())) {
+            realFtpsClient.enterLocalPassiveMode();
+            realFtpsClient.setFileType(FTP.BINARY_FILE_TYPE);
+            // Set protection buffer size
+            realFtpsClient.execPBSZ(0);
+            // Set data channel protection to private
+            realFtpsClient.execPROT("P");
+            realFtpsClient.setBufferSize(1024 * 1024);
+        } else {
+            throw new DatafileTaskException("Unable to connect to xNF. " + fileServerData.serverAddress()
+                    + " xNF reply code: " + realFtpsClient.getReplyCode());
+        }
 
-    void setFtpsClient(FTPSClient ftpsClient) {
-        this.realFtpsClient = ftpsClient;
+        logger.trace("setUpConnection successfully!");
     }
 
-    void setKeyManagerUtils(IKeyManagerUtils keyManagerUtils) {
-        this.keyManagerUtils = keyManagerUtils;
+    InputStream createInputStream(Path localFileName) throws IOException {
+        FileSystemResource realResource = new FileSystemResource(localFileName);
+        return realResource.getInputStream();
     }
 
-    void setKeyStore(IKeyStore keyStore) {
-        this.keyStore = keyStore;
+    OutputStream createOutputStream(Path localFileName) throws IOException {
+        File localFile = localFileName.toFile();
+        if (localFile.createNewFile()) {
+            logger.warn("Local file {} already created", localFileName);
+        }
+        OutputStream output = new FileOutputStream(localFile);
+        logger.debug("File {} opened xNF", localFileName);
+        return output;
+    }
+
+    private TrustManager createTrustManager(Path trustedCAPath, String trustedCAPassword)
+            throws IOException, KeyStoreException, NoSuchAlgorithmException, CertificateException {
+        try (InputStream fis = createInputStream(trustedCAPath)) {
+            KeyStore keyStore = KeyStore.getInstance("JKS");
+            keyStore.load(fis, trustedCAPassword.toCharArray());
+            TrustManagerFactory factory = TrustManagerFactory.getInstance("SunX509");
+            factory.init(keyStore);
+            return factory.getTrustManagers()[0];
+        }
     }
 
-    void setTrustManagerFactory(ITrustManagerFactory tmf) {
-        trustManagerFactory = tmf;
+    TrustManager getTrustManager(Path trustedCAPath, String trustedCAPassword)
+            throws KeyStoreException, NoSuchAlgorithmException, IOException, CertificateException {
+        synchronized (FtpsClient.class) {
+            if (theTrustManager == null) {
+                theTrustManager = createTrustManager(trustedCAPath, trustedCAPassword);
+            }
+            return theTrustManager;
+        }
     }
 
-    void setFileSystemResource(IFileSystemResource fileSystemResource) {
-        this.fileSystemResource = fileSystemResource;
+    KeyManager createKeyManager(String keyCertPath, String keyCertPassword)
+            throws IOException, GeneralSecurityException {
+        return KeyManagerUtils.createClientKeyManager(new File(keyCertPath), keyCertPassword);
     }
 }
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java
index 0c6491b8..2f489166 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java
@@ -38,6 +38,8 @@ import org.slf4j.LoggerFactory;
 public class SftpClient implements FileCollectClient {
     private static final Logger logger = LoggerFactory.getLogger(SftpClient.class);
     private final FileServerData fileServerData;
+    private Session session = null;
+    private ChannelSftp sftpChannel = null;
 
     public SftpClient(FileServerData fileServerData) {
         this.fileServerData = fileServerData;
@@ -48,18 +50,37 @@ public class SftpClient implements FileCollectClient {
         logger.trace("collectFile called");
 
         try {
-            Session session = setUpSession(fileServerData);
-            ChannelSftp sftpChannel = getChannel(session);
             sftpChannel.get(remoteFile, localFile.toString());
             logger.debug("File {} Download Successfull from xNF", localFile.getFileName());
-            sftpChannel.exit();
-            session.disconnect();
         } catch (Exception e) {
-            throw new DatafileTaskException("Unable to get file from xNF. Data: " + fileServerData + e);
+            throw new DatafileTaskException("Unable to get file from xNF. Data: " + fileServerData, e);
         }
 
         logger.trace("collectFile OK");
+    }
+
+    @Override
+    public void close() {
+        logger.trace("close");
+        if (sftpChannel != null) {
+            sftpChannel.exit();
+            sftpChannel = null;
+        }
+        if (session != null) {
+            session.disconnect();
+            session = null;
+        }
+    }
 
+    public void open() throws DatafileTaskException {
+        try {
+            if (session == null) {
+                session = setUpSession(fileServerData);
+                sftpChannel = getChannel(session);
+            }
+        } catch (JSchException e) {
+            throw new DatafileTaskException("Could not open Sftp client", e);
+        }
     }
 
     private int getPort(Optional<Integer> port) {
@@ -70,12 +91,12 @@ public class SftpClient implements FileCollectClient {
     private Session setUpSession(FileServerData fileServerData) throws JSchException {
         JSch jsch = new JSch();
 
-        Session session =
-                jsch.getSession(fileServerData.userId(), fileServerData.serverAddress(), getPort(fileServerData.port()));
-        session.setConfig("StrictHostKeyChecking", "no");
-        session.setPassword(fileServerData.password());
-        session.connect();
-        return session;
+        Session newSession = jsch.getSession(fileServerData.userId(), fileServerData.serverAddress(),
+                getPort(fileServerData.port()));
+        newSession.setConfig("StrictHostKeyChecking", "no");
+        newSession.setPassword(fileServerData.password());
+        newSession.connect();
+        return newSession;
     }
 
     private ChannelSftp getChannel(Session session) throws JSchException {
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
index f80fcd0f..9304688f 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
@@ -19,8 +19,10 @@ package org.onap.dcaegen2.collectors.datafile.service.producer;
 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 com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
@@ -33,7 +35,9 @@ import java.security.NoSuchAlgorithmException;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Future;
+
 import javax.net.ssl.SSLContext;
+
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
@@ -58,6 +62,7 @@ import org.slf4j.MarkerFactory;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.util.DefaultUriBuilderFactory;
+
 import reactor.core.publisher.Mono;
 
 /**
@@ -85,7 +90,6 @@ public class DmaapProducerReactiveHttpClient {
     private final String pwd;
 
     private IFileSystemResource fileResource = new FileSystemResourceWrapper();
-    private CloseableHttpAsyncClient webClient;
 
     /**
      * Constructor DmaapProducerReactiveHttpClient.
@@ -111,10 +115,7 @@ public class DmaapProducerReactiveHttpClient {
     public Mono<HttpStatus> getDmaapProducerResponse(ConsumerDmaapModel consumerDmaapModel,
             Map<String, String> contextMap) {
         MdcVariables.setMdcContextMap(contextMap);
-        logger.trace("Entering getDmaapProducerResponse with {}", consumerDmaapModel);
-        try {
-            webClient = getWebClient();
-            webClient.start();
+       try (CloseableHttpAsyncClient webClient = createWebClient()) {
 
             HttpPut put = new HttpPut();
             prepareHead(consumerDmaapModel, put);
@@ -124,8 +125,7 @@ public class DmaapProducerReactiveHttpClient {
             logger.trace(INVOKE, "Starting to publish to DR {}", consumerDmaapModel.getInternalLocation());
             Future<HttpResponse> future = webClient.execute(put, null);
             HttpResponse response = future.get();
-            logger.trace(INVOKE_RETURN, "Response from DR {}", response.toString());
-            webClient.close();
+            logger.trace(INVOKE_RETURN, "Response from DR {}", response);
             return Mono.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode()));
         } catch (Exception e) {
             logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel.getInternalLocation(), e);
@@ -175,25 +175,20 @@ public class DmaapProducerReactiveHttpClient {
         fileResource = fileSystemResource;
     }
 
-    protected CloseableHttpAsyncClient getWebClient()
+    protected CloseableHttpAsyncClient createWebClient()
             throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
-        if (webClient != null) {
-            return webClient;
-        }
-        SSLContext sslContext = null;
 
-        sslContext = new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build();
+        SSLContext sslContext = new SSLContextBuilder() //
+                .loadTrustMaterial(null, (certificate, authType) -> true) //
+                .build();
 
-        //@formatter:off
-        return HttpAsyncClients.custom()
-                .setSSLContext(sslContext)
-                .setSSLHostnameVerifier(new NoopHostnameVerifier())
-                .setRedirectStrategy(PublishRedirectStrategy.INSTANCE)
+        CloseableHttpAsyncClient webClient =  HttpAsyncClients.custom() //
+                .setSSLContext(sslContext) //
+                .setSSLHostnameVerifier(new NoopHostnameVerifier()) //
+                .setRedirectStrategy(PublishRedirectStrategy.INSTANCE) //
                 .build();
-        //@formatter:on
+        webClient.start();
+        return webClient;
     }
 
-    protected void setWebClient(CloseableHttpAsyncClient client) {
-        this.webClient = client;
-    }
 }
\ No newline at end of file
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java
deleted file mode 100644
index 8c4525e7..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- *      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.collectors.datafile.ssl;
-
-import javax.net.ssl.KeyManager;
-
-public interface IKeyManagerUtils {
-    public void setCredentials(String keyStorePath, String keyStorePass) throws KeyManagerException;
-
-    public KeyManager getClientKeyManager();
-
-    public static class KeyManagerException extends Exception {
-        private static final long serialVersionUID = 1L;
-
-        public KeyManagerException(Exception e) {
-            super(e);
-        }
-    }
-}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java
deleted file mode 100644
index 0e54cecf..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- * 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.collectors.datafile.ssl;
-
-import java.io.InputStream;
-import java.security.KeyStore;
-
-public interface IKeyStore {
-    public void load(InputStream arg0, char[] arg1) throws KeyStoreLoadException;
-
-    public KeyStore getKeyStore();
-
-    public static class KeyStoreLoadException extends Exception {
-        private static final long serialVersionUID = 1L;
-
-        public KeyStoreLoadException(Exception e) {
-            super(e);
-        }
-    }
-}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java
deleted file mode 100644
index 99e3de1f..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- *      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.collectors.datafile.ssl;
-
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-
-import javax.net.ssl.TrustManager;
-
-public interface ITrustManagerFactory {
-    public void init(KeyStore ks) throws KeyStoreException;
-
-    public TrustManager[] getTrustManagers();
-}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java
deleted file mode 100644
index 93a7a2fb..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- *      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.collectors.datafile.ssl;
-
-import java.io.File;
-import java.io.IOException;
-import java.security.GeneralSecurityException;
-
-import javax.net.ssl.KeyManager;
-
-import org.apache.commons.net.util.KeyManagerUtils;
-
-public class KeyManagerUtilsWrapper implements IKeyManagerUtils {
-    private KeyManager keyManager;
-
-    @Override
-    public void setCredentials(String keyStorePath, String keyStorePass) throws KeyManagerException {
-        try {
-            keyManager = KeyManagerUtils.createClientKeyManager(new File(keyStorePath), keyStorePass);
-        } catch (IOException | GeneralSecurityException e) {
-            throw new KeyManagerException(e);
-        }
-    }
-
-    @Override
-    public KeyManager getClientKeyManager() {
-        return keyManager;
-    }
-}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java
deleted file mode 100644
index a8eebea7..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- * 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.collectors.datafile.ssl;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
-
-public class KeyStoreWrapper implements IKeyStore {
-    private KeyStore keyStore;
-
-    public KeyStoreWrapper() throws KeyStoreException {
-        keyStore = KeyStore.getInstance("JKS");
-    }
-
-    @Override
-    public void load(InputStream stream, char[] password)
-            throws KeyStoreLoadException {
-        try {
-            keyStore.load(stream, password);
-        } catch (NoSuchAlgorithmException | CertificateException | IOException e) {
-            throw new KeyStoreLoadException(e);
-        }
-    }
-
-    @Override
-    public KeyStore getKeyStore() {
-        return keyStore;
-    }
-
-}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java
deleted file mode 100644
index f539634d..00000000
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018 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
- *
- *      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.collectors.datafile.ssl;
-
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-import java.security.NoSuchAlgorithmException;
-
-import javax.net.ssl.TrustManager;
-import javax.net.ssl.TrustManagerFactory;
-
-public class TrustManagerFactoryWrapper implements ITrustManagerFactory {
-    TrustManagerFactory tmf;
-
-    public TrustManagerFactoryWrapper() throws NoSuchAlgorithmException {
-        tmf = TrustManagerFactory.getInstance("SunX509");
-    }
-
-    @Override
-    public void init(KeyStore ks) throws KeyStoreException {
-        tmf.init(ks);
-    }
-
-    @Override
-    public TrustManager[] getTrustManagers() {
-        return tmf.getTrustManagers();
-    }
-
-}
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
index 670b1bdc..e2882606 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
@@ -17,8 +17,11 @@
 package org.onap.dcaegen2.collectors.datafile.ftp;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -29,23 +32,15 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.security.GeneralSecurityException;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
 
 import javax.net.ssl.KeyManager;
 import javax.net.ssl.TrustManager;
 
 import org.apache.commons.net.ftp.FTP;
-import org.apache.commons.net.ftp.FTPReply;
 import org.apache.commons.net.ftp.FTPSClient;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
-import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
-import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
-import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
-import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
 import org.springframework.http.HttpStatus;
 
 public class FtpsClientTest {
@@ -62,18 +57,14 @@ public class FtpsClientTest {
     private static final String USERNAME = "bob";
     private static final String PASSWORD = "123";
 
+
     private FTPSClient ftpsClientMock = mock(FTPSClient.class);
-    private IKeyManagerUtils keyManagerUtilsMock = mock(IKeyManagerUtils.class);
     private KeyManager keyManagerMock = mock(KeyManager.class);
-    private IKeyStore keyStoreWrapperMock = mock(IKeyStore.class);
-    private KeyStore keyStoreMock = mock(KeyStore.class);
-    private ITrustManagerFactory trustManagerFactoryMock = mock(ITrustManagerFactory.class);
     private TrustManager trustManagerMock = mock(TrustManager.class);
-    private IFileSystemResource fileResourceMock = mock(IFileSystemResource.class);
     private InputStream inputStreamMock = mock(InputStream.class);
+    private OutputStream outputStreamMock = mock(OutputStream.class);
 
-    FtpsClient clientUnderTest = new FtpsClient(createFileServerData());
-
+    FtpsClient clientUnderTestSpy;
 
     private ImmutableFileServerData createFileServerData() {
         return ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS).userId(USERNAME).password(PASSWORD)
@@ -82,40 +73,17 @@ public class FtpsClientTest {
 
     @BeforeEach
     protected void setUp() throws Exception {
-        clientUnderTest.setFtpsClient(ftpsClientMock);
-        clientUnderTest.setKeyManagerUtils(keyManagerUtilsMock);
-        clientUnderTest.setKeyStore(keyStoreWrapperMock);
-        clientUnderTest.setTrustManagerFactory(trustManagerFactoryMock);
-        clientUnderTest.setFileSystemResource(fileResourceMock);
-
-        clientUnderTest.setKeyCertPath(FTP_KEY_PATH);
-        clientUnderTest.setKeyCertPassword(FTP_KEY_PASSWORD);
-        clientUnderTest.setTrustedCAPath(TRUSTED_CA_PATH.toString());
-        clientUnderTest.setTrustedCAPassword(TRUSTED_CA_PASSWORD);
+        clientUnderTestSpy = spy(new FtpsClient(createFileServerData()));
+        clientUnderTestSpy.realFtpsClient = ftpsClientMock;
     }
 
-    @Test
-    public void collectFile_allOk() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
-        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
-        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
-
-        when(ftpsClientMock.isConnected()).thenReturn(false, true);
+    private void verifyFtpsClientMock_openOK() throws Exception {
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+
         when(ftpsClientMock.retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
                 ArgumentMatchers.any(OutputStream.class))).thenReturn(true);
-
-        clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH);
-
         verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
         verify(ftpsClientMock).setKeyManager(keyManagerMock);
-        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
-        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
-        verify(inputStreamMock, times(1)).close();
-        verify(trustManagerFactoryMock).init(keyStoreMock);
         verify(ftpsClientMock).setTrustManager(trustManagerMock);
         verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
         verify(ftpsClientMock).login(USERNAME, PASSWORD);
@@ -125,59 +93,85 @@ public class FtpsClientTest {
         verify(ftpsClientMock).execPROT("P");
         verify(ftpsClientMock).setFileType(FTP.BINARY_FILE_TYPE);
         verify(ftpsClientMock).setBufferSize(1024 * 1024);
-        verify(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
+    }
+
+    @Test
+    public void collectFile_allOk() throws Exception {
+
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+
+        doReturn(true).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
+        clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        doReturn(true).when(ftpsClientMock).isConnected();
+        clientUnderTestSpy.close();
+
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).isConnected();
         verify(ftpsClientMock, times(1)).logout();
         verify(ftpsClientMock, times(1)).disconnect();
-        verify(ftpsClientMock, times(2)).isConnected();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
     public void collectFileFaultyOwnKey_shouldFail() throws Exception {
-        doThrow(new IKeyManagerUtils.KeyManagerException(new GeneralSecurityException())).when(keyManagerUtilsMock)
-                .setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
-        when(ftpsClientMock.isConnected()).thenReturn(false);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH)).hasMessage(
-                "org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils$KeyManagerException: java.security.GeneralSecurityException");
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        assertThatThrownBy(() -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH,
+                TRUSTED_CA_PASSWORD)).hasMessageContaining(
+                        "Could not open connection: java.io.FileNotFoundException:");
 
         verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+
+        doReturn(false).when(ftpsClientMock).isConnected();
+        clientUnderTestSpy.close();
         verify(ftpsClientMock).isConnected();
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
-    public void collectFileFaultTrustedCA_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+    public void collectFileFaultTrustedCA_shouldFail_no_trustedCA_file() throws Exception {
+
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doThrow(new IOException("problem")).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH);
+
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Could not open connection: java.io.IOException: problem");
+    }
+
+    @Test
+    public void collectFileFaultTrustedCA_shouldFail_empty_trustedCA_file() throws Exception {
 
-        doThrow(new KeyStoreException()).when(trustManagerFactoryMock).init(keyStoreMock);
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(inputStreamMock).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to trust xNF's CA, trustedCAPath java.security.KeyStoreException");
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Could not open connection: java.io.EOFException");
     }
 
     @Test
     public void collectFileFaultyLogin_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
-        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(false);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to log in to xNF. 127.0.0.1");
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(false).when(ftpsClientMock).login(USERNAME, PASSWORD);
+
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Unable to log in to xNF. 127.0.0.1");
 
         verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
         verify(ftpsClientMock).setKeyManager(keyManagerMock);
-        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
-        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
-        verify(inputStreamMock, times(1)).close();
-        verify(trustManagerFactoryMock).init(keyStoreMock);
         verify(ftpsClientMock).setTrustManager(trustManagerMock);
         verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
         verify(ftpsClientMock).login(USERNAME, PASSWORD);
@@ -185,105 +179,61 @@ public class FtpsClientTest {
 
     @Test
     public void collectFileBadRequestResponse_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
-        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
-        when(ftpsClientMock.getReplyCode()).thenReturn(FTPReply.BAD_COMMAND_SEQUENCE);
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(503).when(ftpsClientMock).getReplyCode();
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503");
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503");
 
         verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
         verify(ftpsClientMock).setKeyManager(keyManagerMock);
-        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
-        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
-        verify(inputStreamMock, times(1)).close();
-        verify(trustManagerFactoryMock).init(keyStoreMock);
         verify(ftpsClientMock).setTrustManager(trustManagerMock);
         verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
         verify(ftpsClientMock).login(USERNAME, PASSWORD);
         verify(ftpsClientMock, times(2)).getReplyCode();
-        verify(ftpsClientMock, times(2)).isConnected();
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
-    public void collectFileFaultyConnection_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+    public void collectFile_shouldFail() throws Exception {
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
 
-        doThrow(new IOException()).when(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+        doReturn(false).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Could not open connection: java.io.IOException");
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+                .hasMessage("Could not retrieve file /dir/sample.txt");
 
-        verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
-        verify(ftpsClientMock).setKeyManager(keyManagerMock);
-        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
-        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
-        verify(inputStreamMock, times(1)).close();
-        verify(trustManagerFactoryMock).init(keyStoreMock);
-        verify(ftpsClientMock).setTrustManager(trustManagerMock);
-        verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
-        verify(ftpsClientMock, times(2)).isConnected();
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
-    public void collectFileFailingFileCollect_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
-        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
-        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
-
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, Paths.get("")))
-                .hasMessage("Could not open connection: java.io.IOException: No such file or directory");
-
-    }
-
-    @Test
-    public void collectFileFailingFileRetrieve_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
-        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
-        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
-        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
+    public void collectFile_shouldFail_ioexception() throws Exception {
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
         when(ftpsClientMock.isConnected()).thenReturn(false);
 
-        doThrow(new IOException("problemas")).when(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
+        doThrow(new IOException("problem")).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Could not open connection: java.io.IOException: problemas");
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+                .hasMessage("Could not fetch file: java.io.IOException: problem");
 
-        verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
-        verify(ftpsClientMock).setKeyManager(keyManagerMock);
-        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
-        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
-        verify(inputStreamMock, times(1)).close();
-        verify(trustManagerFactoryMock).init(keyStoreMock);
-        verify(ftpsClientMock).setTrustManager(trustManagerMock);
-        verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
-        verify(ftpsClientMock).login(USERNAME, PASSWORD);
-        verify(ftpsClientMock).getReplyCode();
-        verify(ftpsClientMock, times(1)).enterLocalPassiveMode();
-        verify(ftpsClientMock).execPBSZ(0);
-        verify(ftpsClientMock).execPROT("P");
-        verify(ftpsClientMock).setFileType(FTP.BINARY_FILE_TYPE);
-        verify(ftpsClientMock).setBufferSize(1024 * 1024);
-        verify(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
-        verify(ftpsClientMock, times(2)).isConnected();
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 }
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
index 90fb9336..85a0c090 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
@@ -20,7 +20,6 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.commons.io.IOUtils.toByteArray;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.github.stefanbirkner.fakesftpserver.rule.FakeSftpServerRule;
 import com.jcraft.jsch.ChannelSftp;
@@ -56,45 +55,45 @@ public class SftpClientTest {
             throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
         FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
                 .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
-        byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
-
-        sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE);
-        byte[] localFile = Files.readAllBytes(LOCAL_DUMMY_FILE.toFile().toPath());
-        assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
-        assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+            sftpClient.open();
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+            byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
+
+            sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE);
+            byte[] localFile = Files.readAllBytes(LOCAL_DUMMY_FILE.toFile().toPath());
+            assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
+            assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        }
     }
 
     @Test
-    public void collectFile_withWrongUserName_shouldFail() throws IOException, JSchException, SftpException {
-        FileServerData expectedFileServerData =
-                ImmutableFileServerData.builder().serverAddress("127.0.0.1").userId("Wrong").password(PASSWORD).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+    public void collectFile_withWrongUserName_shouldFail() throws DatafileTaskException, IOException {
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
+                .userId("wrong").password(PASSWORD).port(sftpServer.getPort()).build();
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
 
 
-        assertThatThrownBy(() -> sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE))
-                .hasMessageContaining("Unable to get file from xNF");
+            assertThatThrownBy(() -> sftpClient.open())
+                    .hasMessageContaining("Could not open Sftp clientcom.jcraft.jsch.JSchException: Auth fail");
+        }
     }
 
     @Test
-    public void collectFile_withWrongFileName_shouldFail() throws IOException, JSchException, SftpException {
+    public void collectFile_withWrongFileName_shouldFail()
+            throws IOException, JSchException, SftpException, DatafileTaskException {
         FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
                 .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+            sftpClient.open();
 
-        String errorMessage = "";
-        try {
-            sftpClient.collectFile("wrong", LOCAL_DUMMY_FILE);
-        } catch (Exception e) {
-            errorMessage = e.getMessage();
+            assertThatThrownBy(() -> sftpClient.collectFile("wrong", LOCAL_DUMMY_FILE)).hasMessageStartingWith(
+                    "Unable to get file from xNF. Data: FileServerData{serverAddress=127.0.0.1, "
+                            + "userId=bob, password=123, port=");
         }
-
-        String expectedErrorMessage = "Unable to get file from xNF. Data: FileServerData{serverAddress=127.0.0.1, "
-                + "userId=bob, password=123, port=";
-        assertTrue(errorMessage.startsWith(expectedErrorMessage));
     }
 
     private static Session connectToServer(FakeSftpServerRule sftpServer) throws JSchException {
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
index 2bbe8e1d..06ff570c 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
@@ -17,21 +17,30 @@
 package org.onap.dcaegen2.collectors.datafile.service.producer;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
@@ -50,6 +59,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPub
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.util.DefaultUriBuilderFactory;
+
 import reactor.test.StepVerifier;
 
 /**
@@ -80,12 +90,13 @@ class DmaapProducerReactiveHttpClientTest {
     private IFileSystemResource fileSystemResourceMock = mock(IFileSystemResource.class);
     private CloseableHttpAsyncClient clientMock;
     private HttpResponse responseMock = mock(HttpResponse.class);
+    @SuppressWarnings("unchecked")
     private Future<HttpResponse> futureMock = mock(Future.class);
     private StatusLine statusLine = mock(StatusLine.class);
     private InputStream fileStream;
 
     @BeforeEach
-    void setUp() {
+    void setUp() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
         when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
         when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
         when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
@@ -111,15 +122,15 @@ class DmaapProducerReactiveHttpClientTest {
                 .build();
         //formatter:on
 
-        dmaapProducerReactiveHttpClient = new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
+        dmaapProducerReactiveHttpClient = spy(new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock));
         dmaapProducerReactiveHttpClient.setFileSystemResource(fileSystemResourceMock);
         clientMock = mock(CloseableHttpAsyncClient.class);
-        dmaapProducerReactiveHttpClient.setWebClient(clientMock);
+        doReturn(clientMock).when(dmaapProducerReactiveHttpClient).createWebClient();
     }
 
     @Test
     void getHttpResponse_Success() throws Exception {
-        mockWebClientDependantObject(true);
+        mockWebClientDependantObject();
 
         HttpPut httpPut = new HttpPut();
         httpPut.addHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_OCTET_STREAM_CONTENT_TYPE);
@@ -151,24 +162,22 @@ class DmaapProducerReactiveHttpClientTest {
 
     @Test
     void getHttpResponse_Fail() throws Exception {
-        mockWebClientDependantObject(false);
-        StepVerifier.create(dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel, any()))
-        .expectError()
-        .verify();
+        Map<String, String> contextMap = new HashMap<>();
+        doReturn(futureMock).when(clientMock).execute(any(), any());
+        doThrow(new InterruptedException()).when(futureMock).get();
+        StepVerifier.create(dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel, contextMap)) //
+           .expectError() //
+           .verify(); //
     }
 
-    private void mockWebClientDependantObject(boolean success)
+    private void mockWebClientDependantObject()
             throws IOException, InterruptedException, ExecutionException {
         fileStream = new ByteArrayInputStream(FILE_CONTENT.getBytes());
         when(fileSystemResourceMock.getInputStream()).thenReturn(fileStream);
-        if (success) {
-            when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
-            when(futureMock.get()).thenReturn(responseMock);
-            when(responseMock.getStatusLine()).thenReturn(statusLine);
-            when(statusLine.getStatusCode()).thenReturn(HttpUtils.SC_OK);
-        } else {
-            when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
-            when(futureMock.get()).thenThrow(new InterruptedException());
-        }
+        when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
+        when(futureMock.get()).thenReturn(responseMock);
+        when(responseMock.getStatusLine()).thenReturn(statusLine);
+        when(statusLine.getStatusCode()).thenReturn(HttpUtils.SC_OK);
+
     }
 }