fix: read-threads option caused class cast exception

core/riot-core/src/main/java/com/redis/riot/core/AbstractJobCommand.java

@@ -138,8 +138,10 @@ protected boolean shouldShowProgress() {
 	protected abstract Job job() throws Exception;
 
 	private <I, O> TaskletStep step(Step<I, O> step) {
+		log.info("Creating {}", step);
 		SimpleStepBuilder<I, O> builder = simpleStep(step);
 		if (stepArgs.getRetryPolicy() == RetryPolicy.NEVER && stepArgs.getSkipPolicy() == SkipPolicy.NEVER) {
+			log.info("");
 			return builder.build();
 		}
 		FaultTolerantStepBuilder<I, O> ftStep = JobUtils.faultTolerant(builder);

core/riot-core/src/main/java/com/redis/riot/core/MainCommand.java

@@ -5,8 +5,6 @@
 
 import org.springframework.util.unit.DataSize;
 
-import com.redis.spring.batch.Range;
-
 import picocli.CommandLine;
 import picocli.CommandLine.IExecutionStrategy;
 import picocli.CommandLine.RunLast;
@@ -50,7 +48,6 @@ protected CommandLine commandLine() {
 		commandLine.setUnmatchedOptionsAllowedAsOptionParameters(false);
 		commandLine.setExecutionExceptionHandler(new PrintExceptionMessageHandler());
 		commandLine.registerConverter(DataSize.class, DataSize::parse);
-		commandLine.registerConverter(Range.class, Range::parse);
 		commandLine.registerConverter(Expression.class, Expression::parse);
 		commandLine.registerConverter(TemplateExpression.class, Expression::parseTemplate);
 		commandLine.setExecutionStrategy(LoggingMixin.executionStrategy(executionStrategy()));

core/riot-core/src/main/java/com/redis/riot/core/Step.java

@@ -195,4 +195,16 @@ public Collection<Class<? extends Throwable>> getRetry() {
 	public Collection<Class<? extends Throwable>> getSkip() {
 		return skip;
 	}
+
+	@Override
+	public String toString() {
+		return "Step [name=" + name + ", reader=" + reader + ", writer=" + writer + ", taskName=" + taskName
+				+ ", statusMessageSupplier=" + statusMessageSupplier + ", maxItemCountSupplier=" + maxItemCountSupplier
+				+ ", processor=" + processor + ", executionListeners=" + executionListeners + ", readListeners="
+				+ readListeners + ", writeListeners=" + writeListeners + ", live=" + live + ", flushInterval="
+				+ flushInterval + ", idleTimeout=" + idleTimeout + ", skip=" + skip + ", noSkip=" + noSkip + ", retry="
+				+ retry + ", noRetry=" + noRetry + "]";
+	}
+
+
 }

plugins/riot/src/main/java/com/redis/riot/AbstractExportCommand.java

@@ -69,10 +69,13 @@ protected <K, V, T, O> Step<KeyValue<K, T>, O> step(String name, RedisItemReader
 			ItemWriter<O> writer) {
 		Step<KeyValue<K, T>, O> step = new Step<>(name, reader, writer);
 		if (reader.getMode() != ReaderMode.LIVEONLY) {
+			log.info("Configuring step with scan size estimator");
 			step.maxItemCountSupplier(reader.scanSizeEstimator());
 		}
 		if (reader.getMode() != ReaderMode.SCAN) {
 			checkNotifyConfig(reader.getClient());
+			log.info("Configuring step {} with live true, flushInterval {}, idleTimeout {}", name,
+					reader.getFlushInterval(), reader.getIdleTimeout());
 			step.live(true);
 			step.flushInterval(reader.getFlushInterval());
 			step.idleTimeout(reader.getIdleTimeout());
@@ -86,10 +89,12 @@ private void checkNotifyConfig(AbstractRedisClient client) {
 			try {
 				valueMap = conn.sync().configGet(NOTIFY_CONFIG);
 			} catch (RedisException e) {
+				log.info("Could not check keyspace notification config", e);
 				return;
 			}
 		}
 		String actual = valueMap.getOrDefault(NOTIFY_CONFIG, "");
+		log.info("Retrieved config {}: {}", NOTIFY_CONFIG, actual);
 		Set<Character> expected = characterSet(NOTIFY_CONFIG_VALUE);
 		Assert.isTrue(characterSet(actual).containsAll(expected),
 				String.format("Keyspace notifications not property configured: expected '%s' but was '%s'.",

plugins/riot/src/main/java/com/redis/riot/GenerateArgs.java

@@ -3,8 +3,8 @@
 import java.time.Instant;
 import java.util.List;
 
-import com.redis.spring.batch.Range;
 import com.redis.spring.batch.item.redis.common.DataType;
+import com.redis.spring.batch.item.redis.common.Range;
 import com.redis.spring.batch.item.redis.gen.CollectionOptions;
 import com.redis.spring.batch.item.redis.gen.GeneratorItemReader;
 import com.redis.spring.batch.item.redis.gen.MapOptions;
@@ -28,10 +28,10 @@ public class GenerateArgs {
 	@Option(names = "--keyspace", description = "Keyspace prefix for generated data structures (default: ${DEFAULT-VALUE}).", paramLabel = "<str>")
 	private String keyspace = GeneratorItemReader.DEFAULT_KEYSPACE;
 
-	@Option(names = "--keys", description = "Range of keys to generate in the form '<start>:<end>' (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")
+	@Option(names = "--key-range", description = "Range of keys to generate in the form '<start>-<end>' (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")
 	private Range keyRange = GeneratorItemReader.DEFAULT_KEY_RANGE;
 
-	@Option(arity = "1..*", names = "--types", description = "Types of data structures to generate: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<type>")
+	@Option(arity = "1..*", names = "--type", description = "Types of data structures to generate: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<type>")
 	private List<DataType> types = GeneratorItemReader.defaultTypes();
 
 	@Option(names = "--index", description = "Name of index to create that matches JSON or hash type.", paramLabel = "<name>")

plugins/riot/src/main/java/com/redis/riot/KeyFilterArgs.java

@@ -8,9 +8,9 @@
 
 import org.springframework.util.CollectionUtils;
 
-import com.redis.spring.batch.Range;
 import com.redis.spring.batch.item.redis.common.BatchUtils;
 import com.redis.spring.batch.item.redis.common.GlobPredicate;
+import com.redis.spring.batch.item.redis.common.Range;
 
 import io.lettuce.core.cluster.SlotHash;
 import io.lettuce.core.codec.RedisCodec;
@@ -24,7 +24,7 @@ public class KeyFilterArgs {
 	@Option(names = "--key-exclude", arity = "1..*", description = "Glob pattern to match keys for exclusion. E.g. 'mykey:*' will exclude keys starting with 'mykey:'.", paramLabel = "<exp>")
 	private List<String> excludes;
 
-	@Option(names = "--key-slots", arity = "1..*", description = "Ranges of key slots to consider for processing. For example '0:8000' will only consider keys that fall within the range 0 to 8000.", paramLabel = "<range>")
+	@Option(names = "--key-slot", arity = "1..*", description = "Ranges of key slots to consider for processing. For example '0:8000' will only consider keys that fall within the range 0 to 8000.", paramLabel = "<range>")
 	private List<Range> slots;
 
 	public <K> Optional<Predicate<K>> predicate(RedisCodec<K, ?> codec) {

plugins/riot/src/main/java/com/redis/riot/KeyValueProcessorArgs.java

@@ -28,10 +28,10 @@ public class KeyValueProcessorArgs {
 	@Option(names = "--ttl-proc", description = "SpEL expression to transform key expiration times.", paramLabel = "<exp>")
 	private Expression ttlExpression;
 
-	@Option(names = "--ttls", description = "Propagate key expiration times. True by default.", negatable = true, defaultValue = "true", fallbackValue = "true")
+	@Option(names = "--ttl", description = "Propagate key expiration times. True by default.", negatable = true, defaultValue = "true", fallbackValue = "true")
 	private boolean propagateTtl = true;
 
-	@Option(names = "--stream-ids", description = "Propagate stream message IDs. True by default.", negatable = true, defaultValue = "true", fallbackValue = "true")
+	@Option(names = "--stream-id", description = "Propagate stream message IDs. True by default.", negatable = true, defaultValue = "true", fallbackValue = "true")
 	private boolean propagateIds = true;
 
 	@Option(names = "--stream-prune", description = "Drop empty streams.")

plugins/riot/src/main/java/com/redis/riot/RangeConverter.java

@@ -0,0 +1,32 @@
+package com.redis.riot;
+
+import org.springframework.util.StringUtils;
+
+import com.redis.spring.batch.item.redis.common.Range;
+
+import picocli.CommandLine.ITypeConverter;
+
+public class RangeConverter implements ITypeConverter<Range> {
+
+	public static final String SEPARATOR = "-";
+
+	@Override
+	public Range convert(String value) {
+		int pos = value.indexOf(SEPARATOR);
+		if (pos == -1) {
+			int intValue = Integer.parseInt(value);
+			return new Range(intValue, intValue);
+		}
+		int min = Integer.parseInt(value.substring(0, pos).trim());
+		int max = max(value.substring(pos + 1).trim());
+		return new Range(min, max);
+	}
+
+	private static int max(String value) {
+		if (StringUtils.hasLength(value)) {
+			return Integer.parseInt(value);
+		}
+		return Integer.MAX_VALUE;
+	}
+
+}

plugins/riot/src/main/java/com/redis/riot/RedisURIConverter.java

@@ -0,0 +1,17 @@
+package com.redis.riot;
+
+import io.lettuce.core.RedisURI;
+import picocli.CommandLine.ITypeConverter;
+
+public class RedisURIConverter implements ITypeConverter<RedisURI> {
+
+	@Override
+	public RedisURI convert(String value) {
+		try {
+			return RedisURI.create(value);
+		} catch (IllegalArgumentException e) {
+			return RedisURI.create(RedisURI.URI_SCHEME_REDIS + "://" + value);
+		}
+	}
+
+}

plugins/riot/src/main/java/com/redis/riot/Replicate.java

@@ -87,7 +87,7 @@ private ItemProcessor<KeyValue<byte[], Object>, KeyValue<byte[], Object>> proces
 
 	private ItemProcessor<KeyValue<byte[], Object>, KeyValue<byte[], Object>> keyValueProcessor() {
 		if (isIgnoreStreamMessageId()) {
-			Assert.isTrue(isStruct(), "--no-stream-ids can only be used with --struct");
+			Assert.isTrue(isStruct(), "--no-stream-id can only be used with --struct");
 		}
 		StandardEvaluationContext evaluationContext = evaluationContext();
 		log.info("Creating processor with {}", processorArgs);

plugins/riot/src/main/java/com/redis/riot/Riot.java

@@ -2,6 +2,7 @@
 
 import com.redis.riot.core.MainCommand;
 import com.redis.riot.operation.OperationCommand;
+import com.redis.spring.batch.item.redis.common.Range;
 
 import io.lettuce.core.RedisURI;
 import picocli.AutoComplete.GenerateCompletion;
@@ -25,8 +26,9 @@ public static void main(String[] args) {
 	@Override
 	protected CommandLine commandLine() {
 		CommandLine commandLine = super.commandLine();
-		commandLine.registerConverter(RedisURI.class, Riot::parseRedisURI);
+		commandLine.registerConverter(RedisURI.class, new RedisURIConverter());
 		commandLine.registerConverter(Region.class, Region::of);
+		commandLine.registerConverter(Range.class, new RangeConverter());
 		return commandLine;
 	}
 
@@ -35,14 +37,6 @@ protected IExecutionStrategy executionStrategy() {
 		return Riot::executionStrategy;
 	}
 
-	public static RedisURI parseRedisURI(String string) {
-		try {
-			return RedisURI.create(string);
-		} catch (IllegalArgumentException e) {
-			return RedisURI.create(RedisURI.URI_SCHEME_REDIS + "://" + string);
-		}
-	}
-
 	public static int executionStrategy(ParseResult parseResult) {
 		for (ParseResult subcommand : parseResult.subcommands()) {
 			Object command = subcommand.commandSpec().userObject();

plugins/riot/src/main/java/com/redis/riot/operation/MemberOperationArgs.java

@@ -9,7 +9,7 @@ public class MemberOperationArgs {
 	@Option(names = "--member-space", description = "Keyspace prefix for member IDs.", paramLabel = "<str>")
 	private String memberSpace;
 
-	@Option(arity = "1..*", names = "--members", description = "Member field names for collections.", paramLabel = "<fields>")
+	@Option(arity = "1..*", names = "--member", description = "Member field names for collections.", paramLabel = "<fields>")
 	private List<String> memberFields;
 
 	public String getMemberSpace() {

plugins/riot/src/main/java/com/redis/riot/operation/TsAddCommand.java

@@ -38,7 +38,7 @@ public class TsAddCommand extends AbstractOperationCommand {
 	@Option(names = "--on-duplicate", description = "Duplicate policy: ${COMPLETION-CANDIDATES} (default: ${DEFAULT-VALUE}).", paramLabel = "<name>")
 	private DuplicatePolicy duplicatePolicy = DEFAULT_DUPLICATE_POLICY;
 
-	@Option(arity = "1..*", names = "--labels", description = "Labels in the form label1=field1 label2=field2...", paramLabel = "SPEL")
+	@Option(arity = "1..*", names = "--label", description = "Labels in the form label1=field1 label2=field2...", paramLabel = "SPEL")
 	private Map<String, String> labels = new LinkedHashMap<>();
 
 	@Override