diff --git a/connectors/riot-db/riot-db.gradle b/connectors/riot-db/riot-db.gradle
index 898a9d91b..8540f8dfc 100644
--- a/connectors/riot-db/riot-db.gradle
+++ b/connectors/riot-db/riot-db.gradle
@@ -31,9 +31,6 @@ dependencies {
     implementation group: 'com.microsoft.sqlserver', name: 'mssql-jdbc', version: mssqlVersion
     implementation group: 'com.oracle.ojdbc', name: 'ojdbc8', version: oracleVersion
     implementation group: 'org.xerial', name: 'sqlite-jdbc', version: sqliteVersion
-    testImplementation group: 'com.redis', name: 'spring-batch-redis', version: springBatchRedisVersion, classifier: 'tests'
-    testImplementation group: 'org.testcontainers', name: 'postgresql', version: testcontainersVersion
-    testImplementation group: 'org.testcontainers', name: 'oracle-xe', version: testcontainersVersion
 }
 
 compileJava {
diff --git a/connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseExport.java b/connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseExport.java
index b851b8d72..f92137e8e 100644
--- a/connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseExport.java
+++ b/connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseExport.java
@@ -25,18 +25,18 @@
 
 public class DatabaseExport extends AbstractExport {
 
-    private final String sql;
-
     public static final boolean DEFAULT_ASSERT_UPDATES = true;
 
     public static final Pattern DEFAULT_KEY_PATTERN = Pattern.compile("\\w+:(?<id>.+)");
 
+    private final String sql;
+
+    private DataSourceOptions dataSourceOptions = new DataSourceOptions();
+
     private Pattern keyPattern = DEFAULT_KEY_PATTERN;
 
     private boolean assertUpdates = DEFAULT_ASSERT_UPDATES;
 
-    private DataSourceOptions dataSourceOptions = new DataSourceOptions();
-
     public DatabaseExport(AbstractRedisClient client, String sql) {
         super(client);
         this.sql = sql;
diff --git a/connectors/riot-faker/src/test/java/com/redis/riot/faker/FakerReaderTests.java b/connectors/riot-faker/src/test/java/com/redis/riot/faker/FakerReaderTests.java
index ce13c840d..cff4b5317 100644
--- a/connectors/riot-faker/src/test/java/com/redis/riot/faker/FakerReaderTests.java
+++ b/connectors/riot-faker/src/test/java/com/redis/riot/faker/FakerReaderTests.java
@@ -8,7 +8,6 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.batch.item.ExecutionContext;
 
-import com.redis.riot.faker.FakerItemReader;
 import com.redis.spring.batch.util.BatchUtils;
 
 class FakerReaderTests {
diff --git a/connectors/riot-file/riot-file.gradle b/connectors/riot-file/riot-file.gradle
index 906f74883..09bb608a0 100644
--- a/connectors/riot-file/riot-file.gradle
+++ b/connectors/riot-file/riot-file.gradle
@@ -30,6 +30,7 @@ dependencies {
         exclude group: 'javax.annotation', module: 'javax.annotation-api'
     }
     testImplementation group: 'com.redis', name: 'spring-batch-redis', version: springBatchRedisVersion, classifier: 'tests'
+    testImplementation 'commons-io:commons-io:2.13.0'
 }
 
 compileJava {
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/AbstractResourceItemWriter.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/AbstractResourceItemWriter.java
index 7bb5db103..acd4b85bb 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/AbstractResourceItemWriter.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/AbstractResourceItemWriter.java
@@ -40,10 +40,6 @@
 import org.springframework.core.io.WritableResource;
 import org.springframework.util.Assert;
 
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.CountingOutputStream;
-import com.redis.riot.file.resource.TransactionAwareBufferedWriter;
-
 /**
  * Base class for item writers that write data to a file or stream. This class provides common features like restart, force
  * sync, append etc. The location of the output file is defined by a {@link Resource} which must represent a writable file.<br>
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriter.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriter.java
index e5fd09d7f..00497435d 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriter.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriter.java
@@ -16,16 +16,13 @@
 
 package com.redis.riot.file.resource;
 
+import java.util.List;
+
 import org.springframework.batch.item.file.transform.LineAggregator;
 import org.springframework.core.io.Resource;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.FlatResourceItemWriter;
-
-import java.util.List;
-
 /**
  * This class is an item writer that writes data to a file or stream. The writer
  * also provides restart. The location of the output file is defined by a
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriterBuilder.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriterBuilder.java
index 32b5e220e..ec99ee2bd 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriterBuilder.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/FlatResourceItemWriterBuilder.java
@@ -1,22 +1,22 @@
 package com.redis.riot.file.resource;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+
 import org.springframework.batch.item.file.FlatFileFooterCallback;
 import org.springframework.batch.item.file.FlatFileHeaderCallback;
-import org.springframework.batch.item.file.transform.*;
+import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor;
+import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
+import org.springframework.batch.item.file.transform.FieldExtractor;
+import org.springframework.batch.item.file.transform.FormatterLineAggregator;
+import org.springframework.batch.item.file.transform.LineAggregator;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.WritableResource;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.FlatResourceItemWriter;
-import com.redis.riot.file.resource.FlatResourceItemWriterBuilder;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Locale;
-
 public class FlatResourceItemWriterBuilder<T> {
 
 	private WritableResource resource;
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriter.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriter.java
index 42bf35310..f1fb75a0c 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriter.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriter.java
@@ -16,6 +16,9 @@
 
 package com.redis.riot.file.resource;
 
+import java.util.Iterator;
+import java.util.List;
+
 import org.springframework.batch.item.json.GsonJsonObjectMarshaller;
 import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
 import org.springframework.batch.item.json.JsonObjectMarshaller;
@@ -24,12 +27,6 @@
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.JsonResourceItemWriter;
-
-import java.util.Iterator;
-import java.util.List;
-
 /**
  * Item writer that writes data in json format to an output file. The location
  * of the output file is defined by a {@link Resource} and must represent a
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriterBuilder.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriterBuilder.java
index a31614b7a..21545371d 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriterBuilder.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/JsonResourceItemWriterBuilder.java
@@ -23,10 +23,6 @@
 import org.springframework.core.io.WritableResource;
 import org.springframework.util.Assert;
 
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.JsonResourceItemWriter;
-import com.redis.riot.file.resource.JsonResourceItemWriterBuilder;
-
 /**
  * Builder for {@link JsonResourceItemWriter}.
  *
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReader.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReader.java
index efbd1cffc..fa05bcf97 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReader.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReader.java
@@ -9,9 +9,6 @@
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
-import com.redis.riot.file.resource.XmlItemReader;
-import com.redis.riot.file.resource.XmlObjectReader;
-
 /**
  * {@link ItemStreamReader} implementation that reads XML objects from a {@link Resource} having the following format:
  * 
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReaderBuilder.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReaderBuilder.java
index 080a98d0a..55f128209 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReaderBuilder.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlItemReaderBuilder.java
@@ -4,10 +4,6 @@
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import com.redis.riot.file.resource.XmlItemReader;
-import com.redis.riot.file.resource.XmlItemReaderBuilder;
-import com.redis.riot.file.resource.XmlObjectReader;
-
 public class XmlItemReaderBuilder<T> {
 
 	private XmlObjectReader<T> xmlObjectReader;
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriter.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriter.java
index bac9c2a95..bf509c936 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriter.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriter.java
@@ -26,8 +26,6 @@
 import org.springframework.util.ClassUtils;
 
 import com.fasterxml.jackson.dataformat.xml.XmlMapper;
-import com.redis.riot.file.resource.AbstractResourceItemWriter;
-import com.redis.riot.file.resource.XmlResourceItemWriter;
 
 /**
  * Item writer that writes data in XML format to an output file. The location of the output file is defined by a
diff --git a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriterBuilder.java b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriterBuilder.java
index fcdeff0c6..2c67e13bf 100644
--- a/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriterBuilder.java
+++ b/connectors/riot-file/src/main/java/com/redis/riot/file/resource/XmlResourceItemWriterBuilder.java
@@ -23,9 +23,6 @@
 import org.springframework.core.io.WritableResource;
 import org.springframework.util.Assert;
 
-import com.redis.riot.file.resource.XmlResourceItemWriter;
-import com.redis.riot.file.resource.XmlResourceItemWriterBuilder;
-
 /**
  * Builder for {@link XmlResourceItemWriter}.
  *
diff --git a/connectors/riot-file/src/test/java/com/redis/riot/file/FileTests.java b/connectors/riot-file/src/test/java/com/redis/riot/file/FileTests.java
index 5fc5ec7fb..57e5f8f76 100644
--- a/connectors/riot-file/src/test/java/com/redis/riot/file/FileTests.java
+++ b/connectors/riot-file/src/test/java/com/redis/riot/file/FileTests.java
@@ -16,7 +16,6 @@
 import org.springframework.batch.item.UnexpectedInputException;
 
 import com.redis.riot.core.operation.HsetBuilder;
-import com.redis.riot.file.FileImport;
 import com.redis.spring.batch.test.AbstractTestBase;
 
 abstract class FileTests extends AbstractTestBase {
@@ -27,6 +26,7 @@ abstract class FileTests extends AbstractTestBase {
 
     private static final String keyspace = "beer";
 
+    @SuppressWarnings("unchecked")
     @Test
     void fileImportJSON() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {
         FileImport executable = new FileImport(client, BEERS_JSON_URL);
@@ -43,6 +43,7 @@ void fileImportJSON() throws UnexpectedInputException, ParseException, NonTransi
         Assertions.assertEquals("Hocus Pocus", beer1.get("name"));
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     void fileApiImportCSV() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {
         FileImport executable = new FileImport(client, "https://storage.googleapis.com/jrx/beers.csv");
@@ -59,6 +60,7 @@ void fileApiImportCSV() throws UnexpectedInputException, ParseException, NonTran
 
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     void fileApiFileExpansion() throws IOException {
         Path temp = Files.createTempDirectory("fileExpansion");
diff --git a/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonStatusCountItemWriter.java b/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonStatusCountItemWriter.java
index 6c1dea16d..05789ce2d 100644
--- a/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonStatusCountItemWriter.java
+++ b/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonStatusCountItemWriter.java
@@ -20,11 +20,27 @@ private long incrementAndGet(Status status) {
         return counts.get(status).incrementAndGet();
     }
 
+    public long getMissing() {
+        return getCount(Status.MISSING);
+    }
+
+    public long getType() {
+        return getCount(Status.TYPE);
+    }
+
+    public long getTtl() {
+        return getCount(Status.TTL);
+    }
+
+    public long getValue() {
+        return getCount(Status.VALUE);
+    }
+
     public long getCount(Status status) {
         return counts.get(status).get();
     }
 
-    public long getTotalCount() {
+    public long getTotal() {
         return counts.values().stream().collect(Collectors.summingLong(AtomicLong::get));
     }
 
diff --git a/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonSummaryLogger.java b/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonSummaryLogger.java
index 15dd31466..a00ad605a 100644
--- a/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonSummaryLogger.java
+++ b/core/riot-core/src/main/java/com/redis/riot/core/KeyComparisonSummaryLogger.java
@@ -26,7 +26,7 @@ public ExitStatus afterStep(StepExecution stepExecution) {
         if (stepExecution.getStatus().isUnsuccessful()) {
             return null;
         }
-        if (writer.getTotalCount() == writer.getCount(Status.OK)) {
+        if (writer.getTotal() == writer.getCount(Status.OK)) {
             out.println("Verification completed: all OK");
             return ExitStatus.COMPLETED;
         }
diff --git a/core/riot-core/src/main/java/com/redis/riot/core/Replication.java b/core/riot-core/src/main/java/com/redis/riot/core/Replication.java
index cadce848b..87f088333 100644
--- a/core/riot-core/src/main/java/com/redis/riot/core/Replication.java
+++ b/core/riot-core/src/main/java/com/redis/riot/core/Replication.java
@@ -48,7 +48,7 @@ public class Replication extends AbstractExport {
 
     private final AbstractRedisClient targetClient;
 
-    private PrintWriter out = new PrintWriter(System.out);
+    private final PrintWriter out;
 
     private ReplicationMode mode = ReplicationMode.SNAPSHOT;
 
@@ -66,6 +66,12 @@ public class Replication extends AbstractExport {
 
     private RedisWriterOptions targetWriterOptions = new RedisWriterOptions();
 
+    public Replication(AbstractRedisClient client, AbstractRedisClient targetClient, PrintWriter out) {
+        super(client);
+        this.targetClient = targetClient;
+        this.out = out;
+    }
+
     public void setTargetReaderOptions(RedisReaderOptions targetReaderOptions) {
         this.targetReaderOptions = targetReaderOptions;
     }
@@ -98,15 +104,6 @@ public void setShowDiff(boolean showDiff) {
         this.showDiff = showDiff;
     }
 
-    public Replication(AbstractRedisClient client, AbstractRedisClient targetClient) {
-        super(client);
-        this.targetClient = targetClient;
-    }
-
-    public void setOut(PrintWriter out) {
-        this.out = out;
-    }
-
     @Override
     protected ValueType getValueType() {
         return valueType;
diff --git a/core/riot-core/src/main/java/com/redis/riot/core/StepBuilder.java b/core/riot-core/src/main/java/com/redis/riot/core/StepBuilder.java
index e58ce26c7..f6470ddca 100644
--- a/core/riot-core/src/main/java/com/redis/riot/core/StepBuilder.java
+++ b/core/riot-core/src/main/java/com/redis/riot/core/StepBuilder.java
@@ -79,6 +79,10 @@ public ItemReader<I> getReader() {
         return reader;
     }
 
+    public ItemWriter<O> getWriter() {
+        return writer;
+    }
+
     @SuppressWarnings("unchecked")
     public StepBuilder<I, O> skippableExceptions(Class<? extends Throwable>... exceptions) {
         this.skippableExceptions = Arrays.asList(exceptions);
diff --git a/core/riot-core/src/test/java/com/redis/riot/core/ReplicationTests.java b/core/riot-core/src/test/java/com/redis/riot/core/ReplicationTests.java
index c57607def..fab06642d 100644
--- a/core/riot-core/src/test/java/com/redis/riot/core/ReplicationTests.java
+++ b/core/riot-core/src/test/java/com/redis/riot/core/ReplicationTests.java
@@ -2,6 +2,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.io.PrintWriter;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -41,6 +42,8 @@ protected static double abv(Map<String, String> beer) {
         return Double.parseDouble(beer.get("abv"));
     }
 
+    private PrintWriter printWriter = new PrintWriter(System.out);
+
     protected void execute(AbstractJobExecutable executable, TestInfo info) {
         executable.setName(name(info));
         executable.execute();
@@ -94,7 +97,7 @@ void testMapProcessorFilter(TestInfo info) throws JobExecutionException {
     void replicate(TestInfo info) throws Throwable {
         generate(info);
         Assertions.assertTrue(commands.dbsize() > 0);
-        Replication replicate = new Replication(client, targetClient);
+        Replication replicate = new Replication(client, targetClient, printWriter);
         replicate.execute();
         Assertions.assertTrue(compare(info));
     }
@@ -104,7 +107,7 @@ void keyProcessor(TestInfo info) throws Throwable {
         String key1 = "key1";
         String value1 = "value1";
         commands.set(key1, value1);
-        Replication replication = new Replication(client, targetClient);
+        Replication replication = new Replication(client, targetClient, printWriter);
         replication.setProcessorOptions(operatorOptions("#{type}:#{key}"));
         execute(replication, info);
         Assertions.assertEquals(value1, targetCommands.get("string:" + key1));
@@ -121,7 +124,7 @@ void keyProcessorWithDate(TestInfo info) throws Throwable {
         String key1 = "key1";
         String value1 = "value1";
         commands.set(key1, value1);
-        Replication replication = new Replication(client, targetClient);
+        Replication replication = new Replication(client, targetClient, printWriter);
         replication.setProcessorOptions(
                 operatorOptions(String.format("#{#date.parse('%s').getTime()}:#{key}", "2010-05-10T00:00:00.000+0000")));
         execute(replication, info);
diff --git a/gradle.properties b/gradle.properties
index 318007ac9..60cdedb74 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -41,7 +41,7 @@ oracleVersion               = 19.3.0.0
 picocliVersion              = 4.7.5
 plexusVersion               = 4.0.0
 progressbarVersion          = 0.10.0
-springBatchRedisVersion     = 3.6.6
+springBatchRedisVersion     = 3.6.7-SNAPSHOT
 sqliteVersion               = 3.43.0.0
 testcontainersRedisVersion  = 1.6.4
 testcontainersVersion       = 1.19.0
diff --git a/plugins/riot/riot.gradle b/plugins/riot/riot.gradle
index a1594d4d1..312fd0570 100644
--- a/plugins/riot/riot.gradle
+++ b/plugins/riot/riot.gradle
@@ -66,6 +66,10 @@ dependencies {
     implementation 'org.slf4j:slf4j-simple'
     testImplementation group: 'com.redis', name: 'spring-batch-redis', version: springBatchRedisVersion, classifier: 'tests'
     testImplementation group: 'org.awaitility', name: 'awaitility', version: awaitilityVersion
+    testImplementation 'org.springframework.boot:spring-boot-autoconfigure'
+    testImplementation 'org.springframework:spring-jdbc'
+    testImplementation group: 'org.testcontainers', name: 'postgresql', version: testcontainersVersion
+    testImplementation group: 'org.testcontainers', name: 'oracle-xe', version: testcontainersVersion
 }
 
 bootStartScripts.enabled = false
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractExportCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractExportCommand.java
index d167eb7a2..4d753180f 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractExportCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractExportCommand.java
@@ -1,5 +1,7 @@
 package com.redis.riot.cli;
 
+import java.util.function.Supplier;
+
 import com.redis.riot.core.AbstractExport;
 import com.redis.riot.core.AbstractJobExecutable;
 import com.redis.riot.core.RedisReaderOptions;
@@ -38,4 +40,9 @@ protected String taskName(StepBuilder<?, ?> step) {
         return "Exporting";
     }
 
+    @Override
+    protected Supplier<String> extraMessage(StepBuilder<?, ?> step) {
+        return null;
+    }
+
 }
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractImportCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractImportCommand.java
index b0c6d0499..27dc8941e 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractImportCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractImportCommand.java
@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import com.redis.riot.cli.operation.DelCommand;
@@ -67,4 +68,9 @@ protected String taskName(StepBuilder<?, ?> step) {
         return "Importing";
     }
 
+    @Override
+    protected Supplier<String> extraMessage(StepBuilder<?, ?> step) {
+        return null;
+    }
+
 }
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractJobCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractJobCommand.java
index 61ed4b66b..49014fd10 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractJobCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractJobCommand.java
@@ -1,5 +1,7 @@
 package com.redis.riot.cli;
 
+import java.util.function.Supplier;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -17,9 +19,6 @@ abstract class AbstractJobCommand extends AbstractCommand {
     @ArgGroup(exclusive = false, heading = "Execution options%n")
     StepArgs stepArgs = new StepArgs();
 
-    @ArgGroup(exclusive = false)
-    ProgressArgs progressArgs = new ProgressArgs();
-
     @Override
     protected Executable getExecutable() {
         AbstractJobExecutable executable = getJobExecutable();
@@ -29,25 +28,26 @@ protected Executable getExecutable() {
     }
 
     protected <I, O> void configure(StepBuilder<I, O> step) {
-        ProgressBarBuilder progressBar = progressBar();
+        ProgressBarBuilder progressBar = new ProgressBarBuilder();
+        progressBar.setStyle(stepArgs.style());
+        progressBar.setUpdateIntervalMillis(stepArgs.updateInterval);
+        progressBar.showSpeed();
+        if (stepArgs.isLog()) {
+            Logger logger = LoggerFactory.getLogger(getClass());
+            progressBar.setConsumer(new DelegatingProgressBarConsumer(logger::info));
+        }
         progressBar.setInitialMax(size(step));
         progressBar.setTaskName(taskName(step));
         ProgressStepListener<O> listener = new ProgressStepListener<>(progressBar);
+        Supplier<String> extraMessage = extraMessage(step);
+        if (extraMessage != null) {
+            listener = listener.extraMessage(extraMessage);
+        }
         step.addExecutionListener(listener);
         step.addWriteListener(listener);
     }
 
-    private ProgressBarBuilder progressBar() {
-        ProgressBarBuilder progressBar = new ProgressBarBuilder();
-        progressBar.setStyle(progressArgs.style());
-        progressBar.setUpdateIntervalMillis(progressArgs.updateInterval);
-        progressBar.showSpeed();
-        if (progressArgs.isLog()) {
-            Logger logger = LoggerFactory.getLogger(getClass());
-            progressBar.setConsumer(new DelegatingProgressBarConsumer(logger::info));
-        }
-        return progressBar;
-    }
+    protected abstract Supplier<String> extraMessage(StepBuilder<?, ?> step);
 
     protected abstract String taskName(StepBuilder<?, ?> step);
 
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractKeyValueImportCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractKeyValueImportCommand.java
index d041b8818..63148129b 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/AbstractKeyValueImportCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/AbstractKeyValueImportCommand.java
@@ -1,6 +1,9 @@
 package com.redis.riot.cli;
 
+import java.util.function.Supplier;
+
 import com.redis.riot.core.AbstractKeyValueImport;
+import com.redis.riot.core.StepBuilder;
 
 import picocli.CommandLine.ArgGroup;
 
@@ -18,4 +21,9 @@ protected AbstractKeyValueImport getJobExecutable() {
 
     protected abstract AbstractKeyValueImport getKeyValueImportExecutable();
 
+    @Override
+    protected Supplier<String> extraMessage(StepBuilder<?, ?> step) {
+        return null;
+    }
+
 }
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/DatabaseExportCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/DatabaseExportCommand.java
index 4a346614a..f9db7916a 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/DatabaseExportCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/DatabaseExportCommand.java
@@ -13,14 +13,14 @@ public class DatabaseExportCommand extends AbstractExportCommand {
     String sql;
 
     @ArgGroup(exclusive = false)
-    DatabaseExportArgs databaseExportArgs = new DatabaseExportArgs();
+    DatabaseExportArgs args = new DatabaseExportArgs();
 
     @Override
     protected DatabaseExport getExportExecutable() {
         DatabaseExport executable = new DatabaseExport(redisClient(), sql);
-        executable.setAssertUpdates(databaseExportArgs.assertUpdates);
-        executable.setDataSourceOptions(databaseExportArgs.dataSourceOptions());
-        executable.setKeyPattern(databaseExportArgs.keyRegex);
+        executable.setAssertUpdates(args.assertUpdates);
+        executable.setDataSourceOptions(args.dataSourceOptions());
+        executable.setKeyPattern(args.keyRegex);
         return executable;
     }
 
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/ProgressArgs.java b/plugins/riot/src/main/java/com/redis/riot/cli/ProgressArgs.java
deleted file mode 100644
index 145733463..000000000
--- a/plugins/riot/src/main/java/com/redis/riot/cli/ProgressArgs.java
+++ /dev/null
@@ -1,33 +0,0 @@
-package com.redis.riot.cli;
-
-import me.tongfei.progressbar.ProgressBarStyle;
-import picocli.CommandLine.Option;
-
-public class ProgressArgs {
-
-    public enum Style {
-        BLOCK, BAR, ASCII, LOG, NONE
-    }
-
-    @Option(names = "--progress", description = "Progress style: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<style>")
-    Style style = Style.ASCII;
-
-    @Option(names = "--progress-interval", description = "Progress update interval in millis (default: ${DEFAULT-VALUE}).", paramLabel = "<ms>", hidden = true)
-    int updateInterval = 300;
-
-    public ProgressBarStyle style() {
-        switch (style) {
-            case BAR:
-                return ProgressBarStyle.COLORFUL_UNICODE_BAR;
-            case BLOCK:
-                return ProgressBarStyle.COLORFUL_UNICODE_BLOCK;
-            default:
-                return ProgressBarStyle.ASCII;
-        }
-    }
-
-    public boolean isLog() {
-        return style == Style.LOG;
-    }
-
-}
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/ReplicationCommand.java b/plugins/riot/src/main/java/com/redis/riot/cli/ReplicationCommand.java
index bc0085a47..070fd7c24 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/ReplicationCommand.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/ReplicationCommand.java
@@ -1,14 +1,18 @@
 package com.redis.riot.cli;
 
 import java.time.Duration;
+import java.util.function.Supplier;
 
 import com.redis.riot.core.AbstractExport;
 import com.redis.riot.core.EvaluationContextOptions;
+import com.redis.riot.core.KeyComparisonStatusCountItemWriter;
 import com.redis.riot.core.KeyValueProcessorOptions;
 import com.redis.riot.core.Replication;
 import com.redis.riot.core.ReplicationMode;
 import com.redis.riot.core.StepBuilder;
+import com.redis.spring.batch.RedisItemReader;
 import com.redis.spring.batch.ValueType;
+import com.redis.spring.batch.reader.KeyspaceNotificationItemReader;
 import com.redis.spring.batch.util.BatchUtils;
 import com.redis.spring.batch.util.KeyComparisonItemReader;
 
@@ -19,9 +23,9 @@
 @Command(name = "replicate", description = "Replicate a Redis database into another Redis database.")
 public class ReplicationCommand extends AbstractExportCommand {
 
-    // private static final String COMPARE_MESSAGE = " | {0} missing | {1} type | {2} value | {3} ttl";
-    //
-    // private static final String QUEUE_MESSAGE = " | {0} queued";
+    private static final String COMPARE_MESSAGE = " | %,d missing | %,d type | %,d value | %,d ttl";
+
+    private static final String QUEUE_MESSAGE = " | %,d queue capacity";
 
     private static final String VAR_SOURCE_REDIS_URI = "src";
 
@@ -56,13 +60,12 @@ public class ReplicationCommand extends AbstractExportCommand {
 
     @Override
     protected AbstractExport getExportExecutable() {
-        Replication executable = new Replication(redisClient(), targetRedisArgs.client());
+        Replication executable = new Replication(redisClient(), targetRedisArgs.client(), parent.out);
         executable.setProcessorOptions(processorOptions());
         executable.setMode(mode);
         executable.setNoVerify(noVerify);
         executable.setProcessorOptions(processorOptions());
         executable.setShowDiff(showDiffs);
-        executable.setOut(parent.out);
         executable.setTargetReaderOptions(targetReaderArgs.readerOptions());
         executable.setTargetWriterOptions(targetWriterArgs.writerOptions());
         executable.setTtlTolerance(Duration.ofMillis(ttlTolerance));
@@ -95,10 +98,41 @@ protected String taskName(StepBuilder<?, ?> step) {
 
     @Override
     protected long size(StepBuilder<?, ?> step) {
-        if (Replication.STEP_COMPARE.equals(step.getName())) {
+        if (Replication.STEP_LIVE.equals(step.getName())) {
             return BatchUtils.SIZE_UNKNOWN;
         }
         return super.size(step);
     }
 
+    @Override
+    protected Supplier<String> extraMessage(StepBuilder<?, ?> step) {
+        switch (step.getName()) {
+            case Replication.STEP_COMPARE:
+                return compareMessage(step);
+            case Replication.STEP_LIVE:
+                return liveMessage(step);
+            default:
+                return super.extraMessage(step);
+        }
+    }
+
+    private Supplier<String> liveMessage(StepBuilder<?, ?> step) {
+        RedisItemReader<?, ?> reader = (RedisItemReader<?, ?>) step.getReader();
+        return () -> liveMessage(reader);
+    }
+
+    private String liveMessage(RedisItemReader<?, ?> reader) {
+        KeyspaceNotificationItemReader<?, ?> keyReader = (KeyspaceNotificationItemReader<?, ?>) reader.getKeyReader();
+        if (keyReader == null) {
+            return "";
+        }
+        return String.format(QUEUE_MESSAGE, keyReader.getQueue().remainingCapacity());
+
+    }
+
+    private Supplier<String> compareMessage(StepBuilder<?, ?> step) {
+        KeyComparisonStatusCountItemWriter writer = (KeyComparisonStatusCountItemWriter) step.getWriter();
+        return () -> String.format(COMPARE_MESSAGE, writer.getMissing(), writer.getType(), writer.getValue(), writer.getTtl());
+    }
+
 }
diff --git a/plugins/riot/src/main/java/com/redis/riot/cli/StepArgs.java b/plugins/riot/src/main/java/com/redis/riot/cli/StepArgs.java
index beabf2c88..208f3d16f 100644
--- a/plugins/riot/src/main/java/com/redis/riot/cli/StepArgs.java
+++ b/plugins/riot/src/main/java/com/redis/riot/cli/StepArgs.java
@@ -5,35 +5,61 @@
 import com.redis.riot.core.RiotSkipPolicy;
 import com.redis.riot.core.StepOptions;
 
+import me.tongfei.progressbar.ProgressBarStyle;
 import picocli.CommandLine.Option;
 
 public class StepArgs {
 
+    public enum ProgressStyle {
+        BLOCK, BAR, ASCII, LOG, NONE
+    }
+
     @Option(names = "--sleep", description = "Duration in ms to sleep after each batch write (default: ${DEFAULT-VALUE}).", paramLabel = "<ms>")
-    private long sleep;
+    long sleep;
 
     @Option(names = "--threads", description = "Thread count (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")
-    private int threads = StepOptions.DEFAULT_THREADS;
+    int threads = StepOptions.DEFAULT_THREADS;
 
     @Option(names = { "-b",
             "--batch" }, description = "Number of items in each batch (default: ${DEFAULT-VALUE}).", paramLabel = "<size>")
-    private int chunkSize = StepOptions.DEFAULT_CHUNK_SIZE;
+    int chunkSize = StepOptions.DEFAULT_CHUNK_SIZE;
 
     @Option(names = "--dry-run", description = "Enable dummy writes.")
-    private boolean dryRun;
+    boolean dryRun;
 
     @Option(names = "--ft", description = "Enable step fault-tolerance. Use in conjunction with retry and skip limit/policy.")
-    private boolean faultTolerance;
+    boolean faultTolerance;
 
     @Option(names = "--skip-policy", description = "Policy to determine if some processing should be skipped: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<name>")
-    private RiotSkipPolicy skipPolicy = StepOptions.DEFAULT_SKIP_POLICY;
+    RiotSkipPolicy skipPolicy = StepOptions.DEFAULT_SKIP_POLICY;
 
     @Option(names = "--skip-limit", description = "LIMIT skip policy: max number of failed items before considering the transfer has failed (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")
-    private int skipLimit = StepOptions.DEFAULT_SKIP_LIMIT;
+    int skipLimit = StepOptions.DEFAULT_SKIP_LIMIT;
 
     @Option(names = "--retry-limit", description = "Maximum number of times to try a failed item. 0 and 1 both translate to no retry. (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")
     private int retryLimit = StepOptions.DEFAULT_RETRY_LIMIT;
 
+    @Option(names = "--progress", description = "Progress style: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<style>")
+    ProgressStyle style = ProgressStyle.ASCII;
+
+    @Option(names = "--progress-interval", description = "Progress update interval in millis (default: ${DEFAULT-VALUE}).", paramLabel = "<ms>", hidden = true)
+    int updateInterval = 300;
+
+    public ProgressBarStyle style() {
+        switch (style) {
+            case BAR:
+                return ProgressBarStyle.COLORFUL_UNICODE_BAR;
+            case BLOCK:
+                return ProgressBarStyle.COLORFUL_UNICODE_BLOCK;
+            default:
+                return ProgressBarStyle.ASCII;
+        }
+    }
+
+    public boolean isLog() {
+        return style == ProgressStyle.LOG;
+    }
+
     public StepOptions stepOptions() {
         StepOptions options = new StepOptions();
         options.setChunkSize(chunkSize);
diff --git a/plugins/riot/src/test/java/com/redis/riot/cli/AbstractIntegrationTests.java b/plugins/riot/src/test/java/com/redis/riot/cli/AbstractIntegrationTests.java
index 88b1269db..29f84f2a8 100644
--- a/plugins/riot/src/test/java/com/redis/riot/cli/AbstractIntegrationTests.java
+++ b/plugins/riot/src/test/java/com/redis/riot/cli/AbstractIntegrationTests.java
@@ -274,7 +274,8 @@ void fileImportGlob() throws Exception {
     }
 
     private int executeImportGlob(ParseResult parseResult) {
-        FileImportCommand command = parseResult.subcommands().get(0).commandSpec().commandLine().getCommand();
+
+        FileImportCommand command = command(parseResult);
         try {
             Path dir = Files.createTempDirectory("import-glob");
             FileCopyUtils.copy(getClass().getClassLoader().getResourceAsStream("files/beers1.csv"),
@@ -369,13 +370,13 @@ void fileDumpImport() throws Exception {
     }
 
     private int executeFileDumpImport(ParseResult parseResult) {
-        FileDumpImportCommand command = parseResult.subcommands().get(0).commandSpec().commandLine().getCommand();
+        FileDumpImportCommand command = command(parseResult);
         command.files = command.files.stream().map(this::replace).collect(Collectors.toList());
         return ExitCode.OK;
     }
 
     private int executeFileDumpExport(ParseResult parseResult) {
-        FileDumpExportCommand command = parseResult.subcommands().get(0).commandSpec().commandLine().getCommand();
+        FileDumpExportCommand command = command(parseResult);
         command.file = replace(command.file);
         return ExitCode.OK;
     }
@@ -722,7 +723,7 @@ protected void runLiveReplication(String filename) throws Exception {
             generator.setCurrentItemCount(3000);
             generator.setMaxItemCount(3500);
             try {
-                run(filename + "-generate-live", 1, generator, writer(client));
+                run(filename + "-generate-live", 1, generator, structWriter(client));
             } catch (Exception e) {
                 log.error("Could not generate data", e);
             }
diff --git a/plugins/riot/src/test/java/com/redis/riot/cli/AbstractRiotTests.java b/plugins/riot/src/test/java/com/redis/riot/cli/AbstractRiotTests.java
index bcac708d4..e5691db89 100644
--- a/plugins/riot/src/test/java/com/redis/riot/cli/AbstractRiotTests.java
+++ b/plugins/riot/src/test/java/com/redis/riot/cli/AbstractRiotTests.java
@@ -34,7 +34,7 @@
 import com.redis.lettucemod.api.sync.RedisModulesCommands;
 import com.redis.lettucemod.util.ClientBuilder;
 import com.redis.lettucemod.util.RedisModulesUtils;
-import com.redis.riot.cli.ProgressArgs.Style;
+import com.redis.riot.cli.StepArgs.ProgressStyle;
 import com.redis.riot.cli.operation.OperationCommand;
 import com.redis.spring.batch.RedisItemWriter;
 import com.redis.spring.batch.ValueType;
@@ -111,6 +111,10 @@ void setupTest() throws Exception {
         awaitEquals(() -> 0l, sync::dbsize);
     }
 
+    protected <T> T command(ParseResult parseResult) {
+        return parseResult.subcommands().get(0).commandSpec().commandLine().getCommand();
+    }
+
     protected void awaitUntilFalse(Callable<Boolean> conditionEvaluator) {
         awaitUntil(() -> !conditionEvaluator.call());
     }
@@ -155,7 +159,7 @@ private int execute(ParseResult parseResult) {
 
     protected void configureCommand(Object command) {
         if (command instanceof AbstractJobCommand) {
-            ((AbstractJobCommand) command).progressArgs.style = Style.NONE;
+            ((AbstractJobCommand) command).stepArgs.style = ProgressStyle.NONE;
         }
     }
 
@@ -189,10 +193,10 @@ protected void generate(String name, GeneratorItemReader reader) throws JobExecu
     }
 
     protected void generate(String name, int chunkSize, GeneratorItemReader reader) throws JobExecutionException {
-        run(name + "-generate", chunkSize, reader, writer(client));
+        run(name + "-generate", chunkSize, reader, structWriter(client));
     }
 
-    protected RedisItemWriter<String, String> writer(AbstractRedisClient client) {
+    protected RedisItemWriter<String, String> structWriter(AbstractRedisClient client) {
         RedisItemWriter<String, String> writer = new RedisItemWriter<>(client, StringCodec.UTF8);
         writer.setValueType(ValueType.STRUCT);
         return writer;
diff --git a/connectors/riot-db/src/test/java/com/redis/riot/db/DatabaseTests.java b/plugins/riot/src/test/java/com/redis/riot/cli/DatabaseTests.java
similarity index 74%
rename from connectors/riot-db/src/test/java/com/redis/riot/db/DatabaseTests.java
rename to plugins/riot/src/test/java/com/redis/riot/cli/DatabaseTests.java
index 76af86589..385ef9f67 100644
--- a/connectors/riot-db/src/test/java/com/redis/riot/db/DatabaseTests.java
+++ b/plugins/riot/src/test/java/com/redis/riot/cli/DatabaseTests.java
@@ -1,4 +1,4 @@
-package com.redis.riot.db;
+package com.redis.riot.cli;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -14,11 +14,13 @@
 import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
 import org.testcontainers.containers.JdbcDatabaseContainer;
 
-import com.redis.spring.batch.test.AbstractTestBase;
 import com.redis.testcontainers.RedisServer;
 import com.redis.testcontainers.RedisStackContainer;
 
-abstract class DatabaseTests extends AbstractTestBase {
+import picocli.CommandLine.ExitCode;
+import picocli.CommandLine.ParseResult;
+
+abstract class DatabaseTests extends AbstractRiotTests {
 
     private static final RedisStackContainer REDIS = new RedisStackContainer(
             RedisStackContainer.DEFAULT_IMAGE_NAME.withTag(RedisStackContainer.DEFAULT_TAG));
@@ -63,11 +65,23 @@ protected void executeScript(String file) throws IOException, SQLException {
         scriptRunner.runScript(new InputStreamReader(inputStream));
     }
 
-    private void configure(DataSourceOptions dataSourceOptions) {
+    protected int executeDatabaseImport(ParseResult parseResult) {
+        DatabaseImportCommand command = command(parseResult);
+        configureDatabase(command.args);
+        return ExitCode.OK;
+    }
+    
+    protected int executeDatabaseExport(ParseResult parseResult) {
+        DatabaseExportCommand command = command(parseResult);
+        configureDatabase(command.args);
+        return ExitCode.OK;
+    }
+
+    private void configureDatabase(DatabaseArgs args) {
         JdbcDatabaseContainer<?> container = getJdbcDatabaseContainer();
-        dataSourceOptions.setUrl(container.getJdbcUrl());
-        dataSourceOptions.setUsername(container.getUsername());
-        dataSourceOptions.setPassword(container.getPassword());
+        args.url = container.getJdbcUrl();
+        args.username = container.getUsername();
+        args.password = container.getPassword();
     }
 
 }
diff --git a/connectors/riot-db/src/test/java/com/redis/riot/db/PostgresTests.java b/plugins/riot/src/test/java/com/redis/riot/cli/PostgresTests.java
similarity index 92%
rename from connectors/riot-db/src/test/java/com/redis/riot/db/PostgresTests.java
rename to plugins/riot/src/test/java/com/redis/riot/cli/PostgresTests.java
index 94a2980e7..0b8a4f9ba 100644
--- a/connectors/riot-db/src/test/java/com/redis/riot/db/PostgresTests.java
+++ b/plugins/riot/src/test/java/com/redis/riot/cli/PostgresTests.java
@@ -1,4 +1,4 @@
-package com.redis.riot.db;
+package com.redis.riot.cli;
 
 import java.io.IOException;
 import java.sql.ResultSet;
@@ -56,8 +56,8 @@ void export() throws Exception {
             statement.execute("ALTER TABLE ONLY mytable ADD CONSTRAINT pk_mytable PRIMARY KEY (id)");
             GeneratorItemReader generator = generator();
             generator.setTypes(DataType.HASH);
-            // TODO generate(filename, DEFAULT_BATCH_SIZE, generator);
-            execute(filename);
+            generate(filename, DEFAULT_BATCH_SIZE, generator);
+            execute(filename, this::executeDatabaseExport);
             statement.execute("SELECT COUNT(*) AS count FROM mytable");
             ResultSet countResultSet = statement.getResultSet();
             countResultSet.next();
@@ -74,10 +74,6 @@ void export() throws Exception {
         }
     }
 
-    private void execute(String filename) {
-        // TODO Auto-generated method stub
-    }
-
     @Test
     void nullValueExport() throws Exception {
         try (Statement statement = databaseConnection.createStatement()) {
@@ -91,7 +87,7 @@ void nullValueExport() throws Exception {
             Map<String, String> hash2 = new HashMap<>();
             hash2.put("field2", "value2");
             sync.hmset("gen:2", hash2);
-            execute("db-export-postgresql");
+            execute("db-export-postgresql", this::executeDatabaseExport);
             statement.execute("SELECT COUNT(*) AS count FROM mytable");
             ResultSet countResultSet = statement.getResultSet();
             countResultSet.next();
@@ -109,7 +105,7 @@ void nullValueExport() throws Exception {
 
     @Test
     void hashImport() throws Exception {
-        execute("db-import-postgresql");
+        execute("db-import-postgresql", this::executeDatabaseImport);
         try (Statement statement = databaseConnection.createStatement()) {
             statement.execute("SELECT COUNT(*) AS count FROM orders");
             RedisModulesCommands<String, String> sync = connection.sync();
@@ -123,15 +119,9 @@ void hashImport() throws Exception {
         }
     }
 
-    @Test
-    void noopImport() throws Exception {
-        execute("db-import-postgresql-noop");
-        Assertions.assertEquals(0, connection.sync().dbsize());
-    }
-
     @Test
     void multiThreadedImport() throws Exception {
-        execute("db-import-postgresql-multithreaded");
+        execute("db-import-postgresql-multithreaded", this::executeDatabaseImport);
         RedisModulesCommands<String, String> sync = connection.sync();
         List<String> keys = sync.keys("order:*");
         try (Statement statement = databaseConnection.createStatement()) {
@@ -147,7 +137,7 @@ void multiThreadedImport() throws Exception {
 
     @Test
     void setImport() throws Exception {
-        execute("db-import-postgresql-set");
+        execute("db-import-postgresql-set", this::executeDatabaseImport);
         try (Statement statement = databaseConnection.createStatement()) {
             statement.execute("SELECT * FROM orders");
             ResultSet resultSet = statement.getResultSet();
diff --git a/connectors/riot-db/src/test/java/com/redis/riot/db/SqlScriptRunner.java b/plugins/riot/src/test/java/com/redis/riot/cli/SqlScriptRunner.java
similarity index 99%
rename from connectors/riot-db/src/test/java/com/redis/riot/db/SqlScriptRunner.java
rename to plugins/riot/src/test/java/com/redis/riot/cli/SqlScriptRunner.java
index dd1b43aa4..809840748 100644
--- a/connectors/riot-db/src/test/java/com/redis/riot/db/SqlScriptRunner.java
+++ b/plugins/riot/src/test/java/com/redis/riot/cli/SqlScriptRunner.java
@@ -1,4 +1,4 @@
-package com.redis.riot.db;
+package com.redis.riot.cli;
 
 import java.io.IOException;
 import java.io.LineNumberReader;