feat: redis sink (#38)

* Initial commit: add class diagram

* move diagram to docs/sinks

* Add config files

* Add files for KeyValue mode support

* Refactor: follow checkstyle

* Add and use REDIS_VALUE_BY_NAME property

* Add logic for parsing key templates

* Use JsonPath for nested Proto and Json field by name support

* add config SINK_DEFAULT_DATATYPE_STRING_ENABLE, add getFieldByName() in OdpfParsedMessage

* Fix instrumentation, error handling

* Add logic to extract nested Proto fields

* Simplify getFieldByName for proto, follow consistent convention for json fields

* Optimization: cache schema mapping

* Ref: Move Json parsing code to statis util method

* Fix getMapping caching, move common code to RedisParser from RedisKeyValueParser

* Add List parser module

* Add Hashset parser module

* Consistent key template formatting

* Add tests for ttl, dataentry, parsers packages

* Add test for JsonUtils class

* Minor refactor, fix tests

* Add tests for redis/client package

* Add tests for redis/parsers package

* Fix RedisHashSetParserTest

* chore: review changes

* test: add test for parseKeyTemplate(), fix test after review changes

* fix client tests

* feat: added response parsing

* Reduce parseKey code, add test for multiple variables in template, fix tests

* fix: review comments

* chore: rearrange packages

* chore: refactor packages

* fix and add tests, checkstyle

* chore: fix tests, improve coverage

* Add test for getErrorsFromResponse() method

* chore: minor change

* Add RedisParserTest, optimize getSchema in RedisParser

* feat: Template abstraction

* chore: move entry package inside client

* Fix tests

* Add RedisSinkTest

* Add StandaloneClientClientTest

* Improve client tests

* Fix Entry tests

* add TTL response

* fix: template constructor should validate arguments

* chore: fix checkstyle

* tests: add test for entries

* tests: add tests for redis clients

* fix: remove schema from entry classes

* fix: minor fixes after qa

Co-authored-by: Vinay Verma <vinayverma610@gmail.com>

Author: eyeofvinay

build.gradle

@@ -39,9 +39,11 @@ dependencies {
     implementation 'com.google.cloud:google-cloud-bigquery:1.115.0'
     implementation "io.grpc:grpc-all:1.38.0"
     implementation group: 'org.slf4j', name: 'jul-to-slf4j', version: '1.7.35'
+    implementation group: 'redis.clients', name: 'jedis', version: '3.0.1'
+    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.5'
     implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.2.1'
     implementation 'org.json:json:20220320'
-
+    implementation group: 'com.jayway.jsonpath', name: 'json-path', version: '2.4.0'
     testImplementation group: 'junit', name: 'junit', version: '4.13'
     testImplementation 'org.hamcrest:hamcrest-all:1.3'
     testImplementation 'org.mockito:mockito-core:4.5.1'

src/main/java/io/odpf/depot/config/OdpfSinkConfig.java

@@ -64,6 +64,10 @@ public interface OdpfSinkConfig extends Config {
     @DefaultValue("")
     String getSinkConnectorSchemaProtoKeyClass();
 
+    @Key("SINK_CONNECTOR_SCHEMA_JSON_PARSER_STRING_MODE_ENABLED")
+    @DefaultValue("true")
+    boolean getSinkConnectorSchemaJsonParserStringModeEnabled();
+
     @Key("SINK_CONNECTOR_SCHEMA_DATA_TYPE")
     @ConverterClass(SinkConnectorSchemaDataTypeConverter.class)
     @DefaultValue("PROTOBUF")

src/main/java/io/odpf/depot/config/RedisSinkConfig.java

@@ -0,0 +1,55 @@
+package io.odpf.depot.config;
+
+import io.odpf.depot.config.converter.JsonToPropertiesConverter;
+import io.odpf.depot.config.converter.RedisSinkDataTypeConverter;
+import io.odpf.depot.config.converter.RedisSinkDeploymentTypeConverter;
+import io.odpf.depot.config.converter.RedisSinkTtlTypeConverter;
+import io.odpf.depot.redis.enums.RedisSinkDataType;
+import io.odpf.depot.redis.enums.RedisSinkDeploymentType;
+import io.odpf.depot.redis.enums.RedisSinkTtlType;
+import org.aeonbits.owner.Config;
+
+import java.util.Properties;
+
+
+@Config.DisableFeature(Config.DisableableFeature.PARAMETER_FORMATTING)
+public interface RedisSinkConfig extends OdpfSinkConfig {
+    @Key("SINK_REDIS_URLS")
+    String getSinkRedisUrls();
+
+    @Key("SINK_REDIS_KEY_TEMPLATE")
+    String getSinkRedisKeyTemplate();
+
+    @Key("SINK_REDIS_DATA_TYPE")
+    @DefaultValue("HASHSET")
+    @ConverterClass(RedisSinkDataTypeConverter.class)
+    RedisSinkDataType getSinkRedisDataType();
+
+    @Key("SINK_REDIS_TTL_TYPE")
+    @DefaultValue("DISABLE")
+    @ConverterClass(RedisSinkTtlTypeConverter.class)
+    RedisSinkTtlType getSinkRedisTtlType();
+
+    @Key("SINK_REDIS_TTL_VALUE")
+    @DefaultValue("0")
+    long getSinkRedisTtlValue();
+
+    @Key("SINK_REDIS_DEPLOYMENT_TYPE")
+    @DefaultValue("Standalone")
+    @ConverterClass(RedisSinkDeploymentTypeConverter.class)
+    RedisSinkDeploymentType getSinkRedisDeploymentType();
+
+    @Key("SINK_REDIS_LIST_DATA_PROTO_INDEX")
+    String getSinkRedisListDataProtoIndex();
+
+    @Key("SINK_REDIS_KEY_VALUE_DATA_FIELD_NAME")
+    String getSinkRedisKeyValueDataFieldName();
+
+    @Key("SINK_REDIS_LIST_DATA_FIELD_NAME")
+    String getSinkRedisListDataFieldName();
+
+    @Key("SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING")
+    @ConverterClass(JsonToPropertiesConverter.class)
+    @DefaultValue("")
+    Properties getSinkRedisHashsetFieldToColumnMapping();
+}

src/main/java/io/odpf/depot/config/converter/JsonToPropertiesConverter.java

@@ -0,0 +1,95 @@
+package io.odpf.depot.config.converter;
+
+import com.google.common.base.Strings;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+
+public class JsonToPropertiesConverter implements org.aeonbits.owner.Converter<Properties> {
+    private static final Gson GSON = new Gson();
+
+    @Override
+    public Properties convert(Method method, String input) {
+        if (Strings.isNullOrEmpty(input)) {
+            return null;
+        }
+        Type type = new TypeToken<Map<String, Object>>() {
+        }.getType();
+        Map<String, Object> m = GSON.fromJson(input, type);
+        Properties properties = getProperties(m);
+        validate(properties);
+        return properties;
+    }
+
+    private Properties getProperties(Map<String, Object> inputMap) {
+        Properties properties = new Properties();
+        for (String key : inputMap.keySet()) {
+            Object value = inputMap.get(key);
+            if (value instanceof String) {
+                properties.put(key, value);
+            } else if (value instanceof Map) {
+                properties.put(key, getProperties((Map) value));
+            }
+        }
+        return properties;
+    }
+
+    private void validate(Properties properties) {
+        DuplicateFinder duplicateFinder = flattenValues(properties)
+                .collect(DuplicateFinder::new, DuplicateFinder::accept, DuplicateFinder::combine);
+        if (duplicateFinder.duplicates.size() > 0) {
+            throw new IllegalArgumentException("duplicates found in SINK_REDIS_HASHSET_FIELD_TO_COLUMN_MAPPING for : " + duplicateFinder.duplicates);
+        }
+    }
+
+    private Stream<String> flattenValues(Properties properties) {
+        return properties
+                .values()
+                .stream()
+                .flatMap(v -> {
+                    if (v instanceof String) {
+                        return Stream.of((String) v);
+                    } else if (v instanceof Properties) {
+                        return flattenValues((Properties) v);
+                    } else {
+                        return Stream.empty();
+                    }
+                });
+    }
+
+    private static class DuplicateFinder implements Consumer<String> {
+        private final Set<String> processedValues = new HashSet<>();
+        private final List<String> duplicates = new ArrayList<>();
+
+        @Override
+        public void accept(String o) {
+            if (processedValues.contains(o)) {
+                duplicates.add(o);
+            } else {
+                processedValues.add(o);
+            }
+        }
+
+        void combine(DuplicateFinder other) {
+            other.processedValues
+                    .forEach(v -> {
+                        if (processedValues.contains(v)) {
+                            duplicates.add(v);
+                        } else {
+                            processedValues.add(v);
+                        }
+                    });
+        }
+    }
+}

src/main/java/io/odpf/depot/config/converter/RedisSinkDataTypeConverter.java

@@ -0,0 +1,13 @@
+package io.odpf.depot.config.converter;
+
+import io.odpf.depot.redis.enums.RedisSinkDataType;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+
+public class RedisSinkDataTypeConverter implements Converter<RedisSinkDataType> {
+    @Override
+    public RedisSinkDataType convert(Method method, String input) {
+        return RedisSinkDataType.valueOf(input.toUpperCase());
+    }
+}

src/main/java/io/odpf/depot/config/converter/RedisSinkDeploymentTypeConverter.java

@@ -0,0 +1,13 @@
+package io.odpf.depot.config.converter;
+
+import io.odpf.depot.redis.enums.RedisSinkDeploymentType;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+
+public class RedisSinkDeploymentTypeConverter implements Converter<RedisSinkDeploymentType> {
+    @Override
+    public RedisSinkDeploymentType convert(Method method, String input) {
+        return RedisSinkDeploymentType.valueOf(input.toUpperCase());
+    }
+}

src/main/java/io/odpf/depot/config/converter/RedisSinkTtlTypeConverter.java

@@ -0,0 +1,13 @@
+package io.odpf.depot.config.converter;
+
+import io.odpf.depot.redis.enums.RedisSinkTtlType;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+
+public class RedisSinkTtlTypeConverter implements Converter<RedisSinkTtlType> {
+    @Override
+    public RedisSinkTtlType convert(Method method, String input) {
+        return RedisSinkTtlType.valueOf(input.toUpperCase());
+    }
+}

src/main/java/io/odpf/depot/error/ErrorInfo.java

@@ -8,10 +8,11 @@
 @Data
 public class ErrorInfo {
 
-    @EqualsAndHashCode.Exclude private Exception exception;
+    @EqualsAndHashCode.Exclude
+    private Exception exception;
     private ErrorType errorType;
 
     public String toString() {
-        return errorType.name();
+        return String.format("Exception %s, ErrorType: %s", exception != null ? exception.getMessage() : "NULL", errorType.name());
     }
 }

src/main/java/io/odpf/depot/message/OdpfMessageSchema.java

@@ -1,8 +1,5 @@
 package io.odpf.depot.message;
-
-import java.io.IOException;
-
 public interface OdpfMessageSchema {
 
-    Object getSchema() throws IOException;
+    Object getSchema();
 }

src/main/java/io/odpf/depot/message/ParsedOdpfMessage.java

@@ -11,4 +11,6 @@ public interface ParsedOdpfMessage {
     void validate(OdpfSinkConfig config);
 
     Map<String, Object> getMapping(OdpfMessageSchema schema) throws IOException;
+
+    Object getFieldByName(String name, OdpfMessageSchema odpfMessageSchema);
 }