Add JSON Field Typecast Support

[ekawinataa]

New Classes:

• TypecastedJsonSerializer
• SerializerConfigConverter

Purpose:
TypecastedJsonSerializer is a decorator for MessageToTemplatizedJson and MessageToJson serializers.
It enables typecasting for the resulting JSON from the wrapped serializer.
Addresses the issue where default Protobuf to JSON conversion(using JSONFormat) treatsing int64 as a String.

Introduces the capability for users to specify JSON paths to be typecasted to INTEGER, LONG, or DOUBLE.
This issue specifically aimed to provide backward compability for users that want to migrate from EGLC to Firehose

Example Parameter:
SERIALIZER_JSON_TYPECAST=[{"jsonPath": "$.someField.nominal", "type": "INTEGER"}]

it should cast from :
{"someField": {"nominal": "12345"}}

to
{"someField": {"nominal":12345}}

Manually casts the result into a specific type.

Reason for the PR
Current Issue: The JsonFormat library converts int64 to String.
User Need: Some users need int64 to be serialized as a JSON integer/number due to strict schema contracts in their Sink.
Solution: A decorator to typecast the JSON output to meet the user's requirements for the HTTP Sink.

Impact
This change should only impact the HTTP Sink, specifically the parts which utilize MessageToTemplatizedJson and MessageToJson serializer

Reference :

• JsonFormat: why is Int64 being converted to a string ? protocolbuffers/protobuf#1823
• [java] Add config option to print 64-bit integers in JSON as unquoted ints if they can be losslessly converted into a 64-bit float protocolbuffers/protobuf#16270

[sumitaich1998]

change test method name to camelCase like shouldIgnoreWhenGivenMessageWithNullMessage

[sumitaich1998]

change test method name to camelcase like shouldCastToNumberWhenGivenMessageWithQuoteWrappedNumber

[sumitaich1998]

build is failing due to checkstyle errors

[sumitaich1998]

please add these additional unit tests

@Test
    public void convertShouldHandleEmptyJsonConfig() {
        String emptyConfigJson = "[]";

        Map<String, Function<String, Object>> result = serializerConfigConverter.convert(null, emptyConfigJson);

        Assertions.assertTrue(result.isEmpty());
    }

    @Test
    public void convertShouldHandleNullJsonConfig() {
        String nullConfigJson = null;

        Map<String, Function<String, Object>> result = serializerConfigConverter.convert(null, nullConfigJson);

        Assertions.assertTrue(result.isEmpty());
    }

    @Test
    public void convertShouldThrowExceptionForUnsupportedDataType() {
        String unsupportedTypeConfigJson = "[{\"jsonPath\": \"$.root.field\", \"type\": \"UNSUPPORTED_TYPE\"}]";

        Assertions.assertThrows(IllegalArgumentException.class,
                () -> serializerConfigConverter.convert(null, unsupportedTypeConfigJson));
    }

    @Test
    public void convertShouldHandleMultipleValidConfigs() {
        String multipleConfigJson = "[{\"jsonPath\": \"$.root.field1\", \"type\": \"LONG\"}, {\"jsonPath\": \"$.root.field2\", \"type\": \"STRING\"}]";

        Map<String, Function<String, Object>> result = serializerConfigConverter.convert(null, multipleConfigJson);

        Function<String, Object> mapper1 = result.get("$.root.field1");
        Function<String, Object> mapper2 = result.get("$.root.field2");

        Assertions.assertNotNull(mapper1);
        Assertions.assertNotNull(mapper2);
        Assertions.assertTrue(mapper1.apply("4") instanceof Long);
        Assertions.assertTrue(mapper2.apply("test") instanceof String);
    }

    @Test
    public void convertShouldThrowExceptionForMissingFieldsInConfig() {
        String missingFieldsConfigJson = "[{\"jsonPath\": \"$.root.field\"}]";

        Assertions.assertThrows(IllegalArgumentException.class,
                () -> serializerConfigConverter.convert(null, missingFieldsConfigJson));
    }

[sumitaich1998]

remove serialize from the starting of the test name , test name should with "should"

[sumitaich1998]

remove serialize from the starting of the test name , test name should with "should"

[sumitaich1998]

remove serialize from the starting of the test name , test name should with "should"

[sumitaich1998]

"There is no need for this separate interface. Can you move this configuration to HttpSinkConfig?"

[sumitaich1998]

after moving you can rename this config and prefix "SINK_HTTP_" to it

[sumitaich1998]

Remove this serializerConfig after you move the config to the HttpSinkConfig class

[sumitaich1998]

Remove this serializerConfig parameter after you move the config to the HttpSinkConfig

[sumitaich1998]

add add the new config to the firehose documentation here https://github.com/goto/firehose/blob/main/docs/docs/sinks/http-sink.md and give an example

[sumitaich1998]

Also bump up Firehose version, just a minor version bump

[sumitaich1998]

Rename method to getSinkHttpSerializerJsonTypecast()

[sumitaich1998]

refactor this code to the following

        serializerConfig.getJsonTypecastMapping().forEach((jsonPath, typecastFunction) -> {
            documentContext.map(jsonPath, (currentValue, configuration) -> 
                Optional.ofNullable(currentValue)
                        .map(value -> typecastFunction.apply(value.toString()))
                        .orElse(null)

[sumitaich1998]

Add try-catch not to allow invalid input to make it future-proof

            try {
                return Integer.valueOf(input);
            } catch (NumberFormatException e) {
                throw new IllegalArgumentException("Invalid input for INTEGER: " + input, e);
            }

[sumitaich1998]

Add try-catch not to allow invalid input to make it future-proof

            try {
                return Long.valueOf(input);
            } catch (NumberFormatException e) {
                throw new IllegalArgumentException("Invalid input for LONG: " + input, e);
            }

[sumitaich1998]

Add try-catch not to allow invalid input to make it future-proof

            try {
                return Double.valueOf(input);
            } catch (NumberFormatException e) {
                throw new IllegalArgumentException("Invalid input for DOUBLE: " + input, e);
            }

[sumitaich1998]

        throw new IllegalArgumentException("Error parsing serializer json config: " + e.getMessage(), e);