Add JSON Field Typecast Support

src/main/java/com/gotocompany/firehose/config/SerializerConfig.java

@@ -0,0 +1,14 @@
+package com.gotocompany.firehose.config;
+
+import com.gotocompany.firehose.config.converter.SerializerConfigConverter;
+import org.aeonbits.owner.Config;
+
+import java.util.Map;
+import java.util.function.Function;
+
+public interface SerializerConfig extends Config {
+    @Config.Key("SERIALIZER_JSON_TYPECAST")
+    @Config.ConverterClass(SerializerConfigConverter.class)
+    @DefaultValue("[]")
+    Map<String, Function<String, Object>> getJsonTypecastMapping();
+}

src/main/java/com/gotocompany/firehose/config/converter/SerializerConfigConverter.java

@@ -0,0 +1,48 @@
+package com.gotocompany.firehose.config.converter;
+
+import com.gotocompany.firehose.serializer.constant.TypecastTarget;
+import lombok.*;
+import lombok.extern.slf4j.Slf4j;
+import org.aeonbits.owner.Converter;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SerializerConfigConverter implements Converter<Map<String, Function<String, Object>>> {
+
+    private final ObjectMapper objectMapper;
+
+    public SerializerConfigConverter() {
+        this.objectMapper = new ObjectMapper();
+    }
+
+    @Override
+    public Map<String, Function<String, Object>> convert(Method method, String input) {
+        try {
+            List<JsonTypecast> jsonTypecasts =
+                    objectMapper.readValue(input, new TypeReference<List<JsonTypecast>>(){});
+            return jsonTypecasts.stream()
+                    .collect(Collectors.toMap(JsonTypecast::getJsonPath, jsonTypecast -> jsonTypecast.getType()::cast));
+        } catch (IOException e) {
+            log.error("Error when parsing serializer json config", e);
+            throw new IllegalArgumentException(e.getMessage(), e.getCause());
+        }
+    }
+
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Data
+    @Builder
+    private static class JsonTypecast {
+        private String jsonPath;
+        private TypecastTarget type;
+    }
+
+}

src/main/java/com/gotocompany/firehose/serializer/TypecastedJsonSerializer.java

@@ -0,0 +1,66 @@
+package com.gotocompany.firehose.serializer;
+
+import com.gotocompany.firehose.config.SerializerConfig;
+import com.gotocompany.firehose.exception.DeserializerException;
+import com.gotocompany.firehose.message.Message;
+import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.Option;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+/***
+ * MessageSerializer wrapping other JSON MessageSerializer which add capability to typecast
+ * some of the fields from the inner serializer
+ */
+@Slf4j
+public class TypecastedJsonSerializer implements MessageSerializer {
+
+    private final MessageSerializer messageSerializer;
+    private final SerializerConfig serializerConfig;
+    private final Configuration configuration;
+
+    /**
+     * Constructor for TypecastedJsonSerializer.
+     *
+     * @param messageSerializer the inner serializer to be wrapped
+     * @param serializerConfig  the configuration for typecasting, where each map contains
+     *                          a JSON path and the desired type
+     */
+    public TypecastedJsonSerializer(MessageSerializer messageSerializer,
+                                    SerializerConfig serializerConfig) {
+        this.messageSerializer = messageSerializer;
+        this.serializerConfig = serializerConfig;
+        this.configuration = Configuration.builder()
+                .options(Option.SUPPRESS_EXCEPTIONS)
+                .build();
+    }
+
+    /**
+     * Serializes the given message, then applies typecasting to specified fields in the resulting JSON.
+     *
+     * @param message the message to be serialized
+     * @return the serialized and typecasted JSON string
+     * @throws DeserializerException if an error occurs during serialization or typecasting
+     */
+    @Override
+    public String serialize(Message message) throws DeserializerException {
+        String jsonString = messageSerializer.serialize(message);
+        DocumentContext documentContext = JsonPath
+                .using(configuration)
+                .parse(jsonString);
+
+        for (Map.Entry<String, Function<String, Object>> entry : serializerConfig.getJsonTypecastMapping()
+                .entrySet()) {
+            documentContext.map(entry.getKey(), (currentValue, configuration) -> Optional.ofNullable(currentValue)
+                    .map(v -> entry.getValue().apply(v.toString()))
+                    .orElse(null)
+            );
+        }
+        return documentContext.jsonString();
+    }
+}

src/main/java/com/gotocompany/firehose/serializer/constant/TypecastTarget.java

@@ -0,0 +1,22 @@
+package com.gotocompany.firehose.serializer.constant;
+
+public enum TypecastTarget {
+    INTEGER {
+        @Override
+        public Object cast(String input) {
+            return Integer.valueOf(input);
+        }
+    }, LONG {
+        @Override
+        public Object cast(String input) {
+            return Long.valueOf(input);
+        }
+    }, DOUBLE {
+        @Override
+        public Object cast(String input) {
+            return Double.valueOf(input);
+        }
+    };
+
+    public abstract Object cast(String input);
+}

src/main/java/com/gotocompany/firehose/sink/http/HttpSinkFactory.java

@@ -2,6 +2,7 @@
 
 
 import com.gotocompany.firehose.config.HttpSinkConfig;
+import com.gotocompany.firehose.config.SerializerConfig;
 import com.gotocompany.firehose.metrics.FirehoseInstrumentation;
 import com.gotocompany.firehose.sink.http.auth.OAuth2Credential;
 import com.gotocompany.firehose.sink.http.request.RequestFactory;
@@ -37,15 +38,15 @@ public class HttpSinkFactory {
      */
     public static AbstractSink create(Map<String, String> configuration, StatsDReporter statsDReporter, StencilClient stencilClient) {
         HttpSinkConfig httpSinkConfig = ConfigFactory.create(HttpSinkConfig.class, configuration);
-
+        SerializerConfig serializerConfig = ConfigFactory.create(SerializerConfig.class, configuration);
         FirehoseInstrumentation firehoseInstrumentation = new FirehoseInstrumentation(statsDReporter, HttpSinkFactory.class);
 
         CloseableHttpClient closeableHttpClient = newHttpClient(httpSinkConfig, statsDReporter);
         firehoseInstrumentation.logInfo("HTTP connection established");
 
         UriParser uriParser = new UriParser(stencilClient.getParser(httpSinkConfig.getInputSchemaProtoClass()), httpSinkConfig.getKafkaRecordParserMode());
 
-        Request request = new RequestFactory(statsDReporter, httpSinkConfig, stencilClient, uriParser).createRequest();
+        Request request = new RequestFactory(statsDReporter, httpSinkConfig, stencilClient, uriParser, serializerConfig).createRequest();
 
         return new HttpSink(new FirehoseInstrumentation(statsDReporter, HttpSink.class), request, closeableHttpClient, stencilClient, httpSinkConfig.getSinkHttpRetryStatusCodeRanges(), httpSinkConfig.getSinkHttpRequestLogStatusCodeRanges());
     }

src/main/java/com/gotocompany/firehose/sink/http/factory/SerializerFactory.java

@@ -1,13 +1,15 @@
 package com.gotocompany.firehose.sink.http.factory;
 
 import com.gotocompany.firehose.config.HttpSinkConfig;
+import com.gotocompany.firehose.config.SerializerConfig;
 import com.gotocompany.firehose.config.enums.HttpSinkDataFormatType;
 import com.gotocompany.firehose.metrics.FirehoseInstrumentation;
 import com.gotocompany.firehose.serializer.JsonWrappedProtoByte;
 import com.gotocompany.firehose.serializer.MessageSerializer;
 import com.gotocompany.firehose.serializer.MessageToJson;
 import com.gotocompany.firehose.serializer.MessageToTemplatizedJson;
 import com.gotocompany.depot.metrics.StatsDReporter;
+import com.gotocompany.firehose.serializer.TypecastedJsonSerializer;
 import com.gotocompany.stencil.client.StencilClient;
 import com.gotocompany.stencil.Parser;
 import lombok.AllArgsConstructor;
@@ -19,6 +21,7 @@
 public class SerializerFactory {
 
     private HttpSinkConfig httpSinkConfig;
+    private SerializerConfig serializerConfig;
     private StencilClient stencilClient;
     private StatsDReporter statsDReporter;
 
@@ -34,10 +37,11 @@ public MessageSerializer build() {
             Parser protoParser = stencilClient.getParser(httpSinkConfig.getInputSchemaProtoClass());
             if (httpSinkConfig.getSinkHttpJsonBodyTemplate().isEmpty()) {
                 firehoseInstrumentation.logDebug("Serializer type: EsbMessageToJson", HttpSinkDataFormatType.JSON);
-                return new MessageToJson(protoParser, false, httpSinkConfig.getSinkHttpSimpleDateFormatEnable());
+                return getTypecastedJsonSerializer(new MessageToJson(protoParser, false, httpSinkConfig.getSinkHttpSimpleDateFormatEnable()));
             } else {
                 firehoseInstrumentation.logDebug("Serializer type: EsbMessageToTemplatizedJson");
-                return MessageToTemplatizedJson.create(new FirehoseInstrumentation(statsDReporter, MessageToTemplatizedJson.class), httpSinkConfig.getSinkHttpJsonBodyTemplate(), protoParser);
+                return getTypecastedJsonSerializer(
+                        MessageToTemplatizedJson.create(new FirehoseInstrumentation(statsDReporter, MessageToTemplatizedJson.class), httpSinkConfig.getSinkHttpJsonBodyTemplate(), protoParser));
             }
         }
 
@@ -48,6 +52,10 @@ public MessageSerializer build() {
         return new JsonWrappedProtoByte();
     }
 
+    private MessageSerializer getTypecastedJsonSerializer(MessageSerializer messageSerializer) {
+        return new TypecastedJsonSerializer(messageSerializer, serializerConfig);
+    }
+
     private boolean isProtoSchemaEmpty() {
         return httpSinkConfig.getInputSchemaProtoClass() == null || httpSinkConfig.getInputSchemaProtoClass().equals("");
     }

src/main/java/com/gotocompany/firehose/sink/http/request/RequestFactory.java

@@ -2,6 +2,7 @@
 
 
 import com.gotocompany.firehose.config.HttpSinkConfig;
+import com.gotocompany.firehose.config.SerializerConfig;
 import com.gotocompany.firehose.config.enums.HttpSinkRequestMethodType;
 import com.gotocompany.firehose.metrics.FirehoseInstrumentation;
 import com.gotocompany.firehose.proto.ProtoToFieldMapper;
@@ -30,6 +31,7 @@
 public class RequestFactory {
 
     private HttpSinkConfig httpSinkConfig;
+    private SerializerConfig serializerConfig;
     private UriParser uriParser;
     private StencilClient stencilClient;
     private StatsDReporter statsDReporter;
@@ -43,11 +45,12 @@ public class RequestFactory {
      * @param stencilClient  the stencil client
      * @param uriParser      the uri parser
      */
-    public RequestFactory(StatsDReporter statsDReporter, HttpSinkConfig httpSinkConfig, StencilClient stencilClient, UriParser uriParser) {
+    public RequestFactory(StatsDReporter statsDReporter, HttpSinkConfig httpSinkConfig, StencilClient stencilClient, UriParser uriParser, SerializerConfig serializerConfig) {
         this.statsDReporter = statsDReporter;
         this.stencilClient = stencilClient;
         this.httpSinkConfig = httpSinkConfig;
         this.uriParser = uriParser;
+        this.serializerConfig = serializerConfig;
         firehoseInstrumentation = new FirehoseInstrumentation(this.statsDReporter, RequestFactory.class);
     }
 
@@ -81,6 +84,7 @@ private ProtoToFieldMapper getProtoToFieldMapper() {
     private JsonBody createBody() {
         MessageSerializer messageSerializer = new SerializerFactory(
                 httpSinkConfig,
+                serializerConfig,
                 stencilClient,
                 statsDReporter)
                 .build();

src/test/java/com/gotocompany/firehose/converter/SerializerConfigConverterTest.java

@@ -0,0 +1,36 @@
+package com.gotocompany.firehose.converter;
+
+import com.gotocompany.firehose.config.converter.SerializerConfigConverter;
+import com.gotocompany.firehose.exception.JsonParseException;
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+
+import java.util.Map;
+import java.util.function.Function;
+
+public class SerializerConfigConverterTest {
+
+    private final SerializerConfigConverter serializerConfigConverter = new SerializerConfigConverter();
+
+    @Test
+    public void convert_GivenValidJsonConfig_ShouldConvertToPropertyMap() {
+        String configJson = "[{\"jsonPath\": \"$.root.field\", \"type\": \"LONG\"}]";
+        String expectedPropertyMapKey = "$.root.field";
+
+        Map<String, Function<String, Object>> result = serializerConfigConverter.convert(null, configJson);
+        Function<String, Object> mapper = result.get(expectedPropertyMapKey);
+        Object mapperResult = mapper.apply("4");
+
+        Assertions.assertNotNull(mapper);
+        Assertions.assertTrue(mapperResult instanceof Long);
+        Assertions.assertEquals(4L, mapperResult);
+    }
+
+    @Test
+    public void convert_GivenInvalidJsonFormat_ShouldThrowJsonParseException() {
+        String malformedConfigJson = "[{\"jsonPath\": \"$.root.field\" \"type\": \"LONG\"";
+
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> serializerConfigConverter.convert(null, malformedConfigJson));
+    }
+}

src/test/java/com/gotocompany/firehose/serializer/TypecastedJsonSerializerTest.java

@@ -0,0 +1,86 @@
+package com.gotocompany.firehose.serializer;
+
+import com.gotocompany.firehose.config.SerializerConfig;
+import com.gotocompany.firehose.config.converter.SerializerConfigConverter;
+import com.gotocompany.firehose.message.Message;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.JsonPath;
+import net.minidev.json.JSONArray;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.function.Function;
+
+public class TypecastedJsonSerializerTest {
+
+    private static final String DEFAULT_JSON_MESSAGE = "{\"key\": \"value\", \"long\":\"1234568129012312\",\"nested\": {\"int\": \"1234\"}, \"double\": \"12.1\"}";
+    private static final String DEFAULT_PARAMETERS = "[{\"jsonPath\": \"$..int\", \"type\": \"INTEGER\"}, {\"jsonPath\": \"$..long\", \"type\": \"LONG\"}, {\"jsonPath\": \"$..double\", \"type\": \"DOUBLE\"}, {\"jsonPath\": \"$..unrecognizedPath\", \"type\": \"INTEGER\"}]";
+
+    private TypecastedJsonSerializer typecastedJsonSerializer;
+
+    @Mock
+    private MessageSerializer messageSerializer;
+
+    @Mock
+    private SerializerConfig serializerConfig;
+
+    private SerializerConfigConverter serializerConfigConverter = new SerializerConfigConverter();
+
+    @Before
+    public void setup() {
+        messageSerializer = Mockito.mock(MessageSerializer.class);
+        serializerConfig = Mockito.mock(SerializerConfig.class);
+        Map<String, Function<String, Object>> property = serializerConfigConverter.convert(null, DEFAULT_PARAMETERS);
+        Mockito.when(messageSerializer.serialize(Mockito.any())).thenReturn(DEFAULT_JSON_MESSAGE);
+        Mockito.when(serializerConfig.getJsonTypecastMapping()).thenReturn(property);
+        typecastedJsonSerializer = new TypecastedJsonSerializer(
+                messageSerializer, serializerConfig
+        );
+    }
+
+    @Test
+    public void serialize_GivenMessageWithQuoteWrappedNumber_ShouldCastToNumber() {
+        String processedJsonString = typecastedJsonSerializer.serialize(buildMessage("key", DEFAULT_JSON_MESSAGE));
+        DocumentContext jsonPath = JsonPath.parse(processedJsonString);
+        JSONArray integerJsonArray = jsonPath.read("$..int");
+        JSONArray longJsonArray = jsonPath.read("$..long");
+        JSONArray doubleJsonArray = jsonPath.read("$..double");
+
+        Assertions.assertTrue(integerJsonArray.get(0) instanceof Integer);
+        Assertions.assertTrue(longJsonArray.get(0) instanceof Long);
+        Assertions.assertTrue(doubleJsonArray.get(0) instanceof Double);
+        Assertions.assertEquals(integerJsonArray.get(0), 1234);
+        Assertions.assertEquals(longJsonArray.get(0), 1234568129012312L);
+        Assertions.assertEquals(doubleJsonArray.get(0), 12.1);
+    }
+
+    @Test
+    public void serialize_GivenMessageWithNullMessage_ShouldIgnore() {
+        String jsonWithNullMappedValue = "{\"key\": \"value\", \"long\":null}";
+        Mockito.when(messageSerializer.serialize(Mockito.any())).thenReturn(jsonWithNullMappedValue);
+        String processedJsonString = typecastedJsonSerializer.serialize(buildMessage("key", jsonWithNullMappedValue));
+        DocumentContext jsonPath = JsonPath.parse(processedJsonString);
+        JSONArray fieldWithValue = jsonPath.read("$..key");
+        JSONArray integerJsonArray = jsonPath.read("$..long");
+
+        Assertions.assertEquals("value", fieldWithValue.get(0));
+        Assertions.assertNull(integerJsonArray.get(0));
+    }
+
+
+    private Message buildMessage(String key, String payload) {
+        return new Message(
+                key.getBytes(StandardCharsets.UTF_8),
+                payload.getBytes(StandardCharsets.UTF_8),
+                "topic",
+                1,
+                1
+        );
+    }
+
+}