Redis sink

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

src/main/java/io/odpf/depot/message/json/JsonOdpfMessageParser.java

@@ -10,6 +10,7 @@
 import io.odpf.depot.message.ParsedOdpfMessage;
 import io.odpf.depot.metrics.Instrumentation;
 import io.odpf.depot.metrics.JsonParserMetrics;
+import io.odpf.depot.utils.JsonUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.json.JSONException;
 import org.json.JSONObject;
@@ -54,21 +55,9 @@ public ParsedOdpfMessage parse(OdpfMessage message, SinkConnectorSchemaMessageMo
                 throw new EmptyMessageException();
             }
             Instant instant = Instant.now();
-            JSONObject jsonObject = new JSONObject(new String(payload));
-            JSONObject jsonWithStringValues = new JSONObject();
-            jsonObject.keySet()
-                    .forEach(k -> {
-                        Object value = jsonObject.get(k);
-                        if (value instanceof JSONObject) {
-                            throw new UnsupportedOperationException("nested json structure not supported yet");
-                        }
-                        if (JSONObject.NULL.equals(value)) {
-                            return;
-                        }
-                        jsonWithStringValues.put(k, value.toString());
-                    });
+            JSONObject jsonObject = JsonUtils.getJsonObject(config, payload);
             instrumentation.captureDurationSince(jsonParserMetrics.getJsonParseTimeTakenMetric(), instant);
-            return new JsonOdpfParsedMessage(jsonWithStringValues);
+            return new JsonOdpfParsedMessage(jsonObject);
         } catch (JSONException ex) {
             throw new IOException("invalid json error", ex);
         }

src/main/java/io/odpf/depot/message/json/JsonOdpfParsedMessage.java

@@ -1,5 +1,8 @@
 package io.odpf.depot.message.json;
 
+import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.spi.json.JsonOrgJsonProvider;
 import io.odpf.depot.config.OdpfSinkConfig;
 import io.odpf.depot.message.OdpfMessageSchema;
 import io.odpf.depot.message.ParsedOdpfMessage;
@@ -32,8 +35,17 @@ public void validate(OdpfSinkConfig config) {
     @Override
     public Map<String, Object> getMapping(OdpfMessageSchema schema) {
         if (jsonObject == null || jsonObject.isEmpty()) {
-            return  Collections.emptyMap();
+            return Collections.emptyMap();
         }
         return jsonObject.toMap();
     }
+
+    public Object getFieldByName(String name, OdpfMessageSchema odpfMessageSchema) {
+        String jsonPathName = "$." + name;
+        Configuration configuration = Configuration.builder()
+                .jsonProvider(new JsonOrgJsonProvider())
+                .build();
+        JsonPath jsonPath = JsonPath.compile(jsonPathName);
+        return jsonPath.read(jsonObject, configuration);
+    }
 }

src/main/java/io/odpf/depot/message/proto/ProtoField.java

@@ -1,11 +1,13 @@
 package io.odpf.depot.message.proto;
 
 import com.google.protobuf.DescriptorProtos;
+import lombok.EqualsAndHashCode;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+@EqualsAndHashCode
 public class ProtoField {
     private String name;
     private String typeName;

src/main/java/io/odpf/depot/message/proto/ProtoOdpfMessageSchema.java

@@ -3,13 +3,15 @@
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import io.odpf.depot.message.OdpfMessageSchema;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.Map;
 import java.util.Properties;
 
+@EqualsAndHashCode
 public class ProtoOdpfMessageSchema implements OdpfMessageSchema {
 
     @Getter
@@ -21,13 +23,13 @@ public ProtoOdpfMessageSchema(ProtoField protoField) throws IOException {
         this(protoField, createProperties(protoField));
     }
 
-    public ProtoOdpfMessageSchema(ProtoField protoField, Properties properties) throws IOException {
+    public ProtoOdpfMessageSchema(ProtoField protoField, Properties properties) {
         this.protoField = protoField;
         this.properties = properties;
     }
 
     @Override
-    public Properties getSchema() throws IOException {
+    public Properties getSchema() {
         return this.properties;
     }