feat: Improved replication status reporting

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 {

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;

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 {

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 {

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>

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

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;

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

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}.
  *

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:
  * 

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;

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

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}.
  *

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");

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

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

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;

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

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

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

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

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