test: Fixed unsupported data structures

Author: jruaux

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/source/DataStructureConverter.java

@@ -34,7 +34,6 @@ public class DataStructureConverter implements Function<DataStructure<String>, S
 	public static final Schema HASH_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).optional();
 	public static final Schema LIST_SCHEMA = SchemaBuilder.array(Schema.STRING_SCHEMA).optional();
 	public static final Schema SET_SCHEMA = SchemaBuilder.array(Schema.STRING_SCHEMA).optional();
-	public static final Schema STREAM_SCHEMA = SchemaBuilder.array(StreamMessageConverter.VALUE_SCHEMA).optional();
 	public static final Schema STRING_SCHEMA = Schema.OPTIONAL_STRING_SCHEMA;
 	public static final Schema ZSET_SCHEMA = SchemaBuilder.map(Schema.FLOAT64_SCHEMA, Schema.STRING_SCHEMA).optional();
 	public static final Schema VALUE_SCHEMA = SchemaBuilder.struct().field(FIELD_KEY, Schema.STRING_SCHEMA)
@@ -48,20 +47,39 @@ public Struct apply(DataStructure<String> input) {
 		struct.put(FIELD_KEY, input.getKey());
 		struct.put(FIELD_TTL, input.getTtl());
 		struct.put(FIELD_TYPE, input.getType());
-		struct.put(fieldName(input), fieldValue(input));
-		return struct;
-	}
-
-	@SuppressWarnings("unchecked")
-	private Object fieldValue(DataStructure<String> input) {
 		switch (input.getType()) {
-		case DataStructure.ZSET:
-			return zsetMap((Collection<ScoredValue<String>>) input.getValue());
+		case DataStructure.HASH:
+			struct.put(FIELD_HASH, input.getValue());
+			break;
+		case DataStructure.JSON:
+			struct.put(FIELD_JSON, input.getValue());
+			break;
+		case DataStructure.LIST:
+			struct.put(FIELD_LIST, input.getValue());
+			break;
 		case DataStructure.SET:
-			return new ArrayList<>(((Collection<String>) input.getValue()));
+			struct.put(FIELD_SET, list(input));
+			break;
+		case DataStructure.STRING:
+			struct.put(FIELD_STRING, input.getValue());
+			break;
+		case DataStructure.ZSET:
+			struct.put(FIELD_ZSET, zsetMap(input));
+			break;
 		default:
-			return input.getValue();
+			break;
 		}
+		return struct;
+	}
+
+	@SuppressWarnings("unchecked")
+	private Object list(DataStructure<String> input) {
+		return new ArrayList<>((Collection<String>) input.getValue());
+	}
+
+	public static Map<Double, String> zsetMap(DataStructure<String> input) {
+		Collection<ScoredValue<String>> value = input.getValue();
+		return zsetMap(value);
 	}
 
 	public static Map<Double, String> zsetMap(Collection<ScoredValue<String>> value) {
@@ -72,23 +90,4 @@ public static Map<Long, Double> timeseriesMap(Collection<Sample> samples) {
 		return samples.stream().collect(Collectors.toMap(Sample::getTimestamp, Sample::getValue));
 	}
 
-	public static String fieldName(DataStructure<String> input) {
-		switch (input.getType()) {
-		case DataStructure.HASH:
-			return FIELD_HASH;
-		case DataStructure.JSON:
-			return FIELD_JSON;
-		case DataStructure.LIST:
-			return FIELD_LIST;
-		case DataStructure.SET:
-			return FIELD_SET;
-		case DataStructure.STRING:
-			return FIELD_STRING;
-		case DataStructure.ZSET:
-			return FIELD_ZSET;
-		default:
-			return FIELD_STRING;
-		}
-	}
-
 }

core/redis-kafka-connect/src/test/java/com/redis/kafka/connect/AbstractSourceIntegrationTests.java

@@ -444,7 +444,9 @@ void pollKeys() throws Exception {
 		for (SourceRecord record : sourceRecords) {
 			Assertions.assertEquals(topic, record.topic());
 			Compare compare = values((Struct) record.value());
-			Assertions.assertEquals(compare.expected, compare.actual);
+			if (compare != null) {
+				Assertions.assertEquals(compare.expected, compare.actual);
+			}
 		}
 
 //		DataStructure<String> stringDS = new DataStructure<>();
@@ -491,7 +493,7 @@ private Compare values(Struct struct) {
 			return compare(DataStructureConverter.zsetMap(commands.zrangeWithScores(key, 0, -1)),
 					struct.getMap(DataStructureConverter.FIELD_ZSET));
 		default:
-			throw new RuntimeException("Unknown type: " + type);
+			return null;
 		}
 	}