Thread safety issues

The TrustManager is now loaded and initialized once
in a thread safe way (instead of each time it is used).

Removed some unneeded wrappers.

Using AutoCloseable for FTP clients to make
sure they are closed in case of exceptions.

Made AppConfig thread safe.

Change-Id: Ia6a2c8a76bf960013180fdd7c53ae0ff17b26505
Issue-ID: DCAEGEN2-1118
Signed-off-by: PatrikBuhr <patrik.buhr@est.tech>

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);
+
     }
 }