feat: Add dynamic source and sink kafka properties #31

ekawinataa · 2024-09-02T08:22:21Z

No description provided.

dagger-core/src/main/java/com/gotocompany/dagger/core/sink/kafka/util/SinkKafkaConfigUtil.java

ekawinataa · 2024-09-10T06:53:20Z

Producer configs :

some changes needed

sumitaich1998

please add unit tests for DaggerKafkaConsumerAdditionalConfigurationsAdaptor and KafkaConfigUtil

sumitaich1998 · 2024-09-16T08:29:35Z

...agger/core/source/config/adapter/DaggerKafkaConsumerAdditionalConfigurationsAdaptorTest.java

+        try {
+            daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader);
+        } catch (IllegalArgumentException e) {
+            assertEquals("Invalid additional kafka consumer configuration properties found: [INVALID_KEY]", e.getMessage());
+        }
+    }


instead of try catch use expected -
@Test(expected = IllegalArgumentException.class) public void shouldThrowExceptionForInvalidProperties() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_1\":\"value1\",\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_2\":\"value2\",\"INVALID_KEY\":\"value3\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); DaggerKafkaConsumerAdditionalConfigurationsAdaptor daggerKafkaConsumerAdditionalConfigurationsAdaptor = new DaggerKafkaConsumerAdditionalConfigurationsAdaptor(); daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader); }

Intention is to assert the exception message as well, but let me update it

@ekawinataa check if (expected = xyz, exception-message = abc) "exception-message" or some other property can be used to assert this, but if the exception message is dynamic we can't use this.

But yes, its good to assert the exception message. So verify it, if not possible keep as is.

use rule for exception message assertions -

@Rule public ExpectedException thrown = ExpectedException.none(); @Test public void shouldThrowExceptionForInvalidProperties() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_1\":\"value1\",\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_2\":\"value2\",\"INVALID_KEY\":\"value3\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); DaggerKafkaConsumerAdditionalConfigurationsAdaptor daggerKafkaConsumerAdditionalConfigurationsAdaptor = new DaggerKafkaConsumerAdditionalConfigurationsAdaptor(); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Invalid additional kafka consumer configuration properties found: [INVALID_KEY]"); daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader); }```

@rajuGT @sumitaich1998 okay, I've checked the current Test annotation and there is no option to assert the message. Let's keep the message assertion, I've added explicit fail in case the expected exception is not thrown. Thanks

no that is not the right way, please use rule to assert message like I've shown above

updated as per discussion @sumitaich1998

…any exception

sumitaich1998 · 2024-09-16T08:41:58Z

...agger/core/source/config/adapter/DaggerKafkaConsumerAdditionalConfigurationsAdaptorTest.java

+    @Test(expected = IllegalArgumentException.class)
    public void shouldThrowExceptionForInvalidProperties() throws IOException {
        String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_1\":\"value1\",\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_2\":\"value2\",\"INVALID_KEY\":\"value3\"}";
        JsonReader jsonReader = new JsonReader(new StringReader(input));
        DaggerKafkaConsumerAdditionalConfigurationsAdaptor daggerKafkaConsumerAdditionalConfigurationsAdaptor = new DaggerKafkaConsumerAdditionalConfigurationsAdaptor();

-        try {
-            daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader);
-        } catch (IllegalArgumentException e) {
-            assertEquals("Invalid additional kafka consumer configuration properties found: [INVALID_KEY]", e.getMessage());
-        }
+        daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader);


use rule for exception message assertions -

@Rule public ExpectedException thrown = ExpectedException.none(); @Test public void shouldThrowExceptionForInvalidProperties() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_1\":\"value1\",\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY_2\":\"value2\",\"INVALID_KEY\":\"value3\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); DaggerKafkaConsumerAdditionalConfigurationsAdaptor daggerKafkaConsumerAdditionalConfigurationsAdaptor = new DaggerKafkaConsumerAdditionalConfigurationsAdaptor(); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Invalid additional kafka consumer configuration properties found: [INVALID_KEY]"); daggerKafkaConsumerAdditionalConfigurationsAdaptor.read(jsonReader); }```

sumitaich1998 · 2024-09-16T09:38:13Z

...agger/core/source/config/adapter/DaggerKafkaConsumerAdditionalConfigurationsAdaptorTest.java

+public class DaggerKafkaConsumerAdditionalConfigurationsAdaptorTest {
+


please add more unit tests covering all possible edge cases , few I've written below -

@Test public void shouldHandleEmptyMap() throws IOException { String input = "{}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertTrue(result.isEmpty()); } @Test public void shouldHandleSingleValidProperty() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"value\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertEquals(1, result.size()); assertEquals("value", result.get("SOURCE_KAFKA_CONSUMER_CONFIG_KEY")); } @Test(expected = IllegalArgumentException.class) public void shouldThrowExceptionForAllInvalidProperties() throws IOException { String input = "{\"INVALID_KEY_1\":\"value1\",\"INVALID_KEY_2\":\"value2\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); adaptor.read(jsonReader); } @Test public void shouldIgnoreNullValues() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":null}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertTrue(result.isEmpty()); } @Test public void shouldHandleMixOfValidAndInvalidProperties() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"value\",\"INVALID_KEY\":\"invalid\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); try { adaptor.read(jsonReader); fail("Expected IllegalArgumentException"); } catch (IllegalArgumentException e) { assertTrue(e.getMessage().contains("INVALID_KEY")); } } @Test public void shouldWriteMapToJsonString() throws IOException { Map<String, String> map = new HashMap<>(); map.put("SOURCE_KAFKA_CONSUMER_CONFIG_KEY", "value"); StringWriter stringWriter = new StringWriter(); JsonWriter jsonWriter = new JsonWriter(stringWriter); adaptor.write(jsonWriter, map); String result = stringWriter.toString(); assertTrue(result.contains("\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"value\"")); } @Test public void shouldHandleSpecialCharactersInValues() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"value with spaces and $pecial ch@racters\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertEquals("value with spaces and $pecial ch@racters", result.get("SOURCE_KAFKA_CONSUMER_CONFIG_KEY")); } @Test public void shouldHandleNumericValues() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"123\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertEquals("123", result.get("SOURCE_KAFKA_CONSUMER_CONFIG_KEY")); } @Test public void shouldHandleBooleanValues() throws IOException { String input = "{\"SOURCE_KAFKA_CONSUMER_CONFIG_KEY\":\"true\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); Map<String, String> result = adaptor.read(jsonReader); assertEquals("true", result.get("SOURCE_KAFKA_CONSUMER_CONFIG_KEY")); } @Test public void shouldHandleCaseInsensitiveKeys() throws IOException { String input = "{\"source_kafka_consumer_config_key\":\"value\"}"; JsonReader jsonReader = new JsonReader(new StringReader(input)); try { adaptor.read(jsonReader); fail("Expected IllegalArgumentException"); } catch (IllegalArgumentException e) { assertTrue(e.getMessage().contains("source_kafka_consumer_config_key")); } }

sumitaich1998 · 2024-09-16T09:46:04Z

dagger-core/src/test/java/com/gotocompany/dagger/core/utils/KafkaConfigUtilTest.java

+public class KafkaConfigUtilTest {
+


please add more unit tests covering all possible edge cases in the pr

@Test public void shouldReturnEmptyPropertiesWhenInputIsEmpty() { Properties properties = new Properties(); Properties kafkaProperties = KafkaConfigUtil.parseKafkaConfiguration(KafkaConnectorTypesMetadata.SOURCE, properties); assertEquals(0, kafkaProperties.size()); } @Test public void shouldReturnEmptyPropertiesWhenAllKeysAreInvalid() { Properties properties = new Properties(); properties.put("INVALID_KEY_1", "value1"); properties.put("INVALID_KEY_2", "value2"); properties.put("ANOTHER_INVALID_KEY", "value3"); Properties kafkaProperties = KafkaConfigUtil.parseKafkaConfiguration(KafkaConnectorTypesMetadata.SOURCE, properties); assertEquals(0, kafkaProperties.size()); } @Test public void shouldParseOnlyValidKeysWhenMixedWithInvalidOnes() { Properties properties = new Properties(); properties.put("SOURCE_KAFKA_CONSUMER_CONFIG_KEY_1", "value1"); properties.put("INVALID_KEY", "value2"); properties.put("SOURCE_KAFKA_CONSUMER_CONFIG_KEY_2", "value3"); properties.put("ANOTHER_INVALID_KEY", "value4"); Properties kafkaProperties = KafkaConfigUtil.parseKafkaConfiguration(KafkaConnectorTypesMetadata.SOURCE, properties); assertEquals(2, kafkaProperties.size()); assertEquals("value1", kafkaProperties.getProperty("key.1")); assertEquals("value3", kafkaProperties.getProperty("key.2")); } @Test public void shouldParseCaseInsensitiveKeys() { Properties properties = new Properties(); properties.put("source_kafka_consumer_config_KEY_1", "value1"); properties.put("SOURCE_KAFKA_CONSUMER_CONFIG_key_2", "value2"); Properties kafkaProperties = KafkaConfigUtil.parseKafkaConfiguration(KafkaConnectorTypesMetadata.SOURCE, properties); assertEquals(2, kafkaProperties.size()); assertEquals("value1", kafkaProperties.getProperty("key.1")); assertEquals("value2", kafkaProperties.getProperty("key.2")); } @Test public void shouldParseKeysWithMultipleUnderscores() { Properties properties = new Properties(); properties.put("SOURCE_KAFKA_CONSUMER_CONFIG_MULTI_WORD_KEY", "value1"); properties.put("SOURCE_KAFKA_CONSUMER_CONFIG_ANOTHER_MULTI_WORD_KEY", "value2"); Properties kafkaProperties = KafkaConfigUtil.parseKafkaConfiguration(KafkaConnectorTypesMetadata.SOURCE, properties); assertEquals(2, kafkaProperties.size()); assertEquals("value1", kafkaProperties.getProperty("multi.word.key")); assertEquals("value2", kafkaProperties.getProperty("another.multi.word.key")); }

ekawinataa added 6 commits September 2, 2024 15:21

add kafka security module

328c44a

aDD SASL class callback config for producer and consumer

1f6d09c

Add config map

19574f9

remove build.gradle

28e31de

Add dynamic props

6c33ff4

Update regex

8ef873b

ekawinataa commented Sep 3, 2024

View reviewed changes

dagger-core/src/main/java/com/gotocompany/dagger/core/sink/kafka/util/SinkKafkaConfigUtil.java Outdated Show resolved Hide resolved

ekawinataa added 2 commits September 3, 2024 14:49

rename var

c3d4556

Remove redundant imports

f19b90b

ekawinataa changed the title ~~add kafka security module~~ Add Consumer and Sink SASL Related Config Sep 3, 2024

ekawinataa changed the title ~~Add Consumer and Sink SASL Related Config~~ Add Consumer and Sink Kafka SASL Related Config Sep 3, 2024

ekawinataa added 5 commits September 3, 2024 15:58

Rename prefix

b731a38

Remove unused import

ef3d7e9

Update test

e643dc6

Add implementation for sink dynamic props

d9cfc8a

Add null checking for the additional props

fc26377

ekawinataa changed the title ~~Add Consumer and Sink Kafka SASL Related Config~~ Add dynamic source and sink kafka properties Sep 5, 2024

ekawinataa added 7 commits September 10, 2024 14:39

Added validations on source config

d2c981d

Add docs and refactor pattern to enum

11c8da5

chECKSTYLE

fb8695b

Add readme

cc2485f

Make the pattern more specific and embedded to the enum

d110153

Add more test

33b0cf9

bump version

5f3a80c

sumitaich1998 previously approved these changes Sep 12, 2024

View reviewed changes

sumitaich1998 reviewed Sep 12, 2024

View reviewed changes

Add unit tests

62b4eca

sumitaich1998 reviewed Sep 16, 2024

View reviewed changes

ekawinataa added 2 commits September 16, 2024 15:34

Use expected annotation

6c4e8ae

Assert exception message. Add fail mechanism in case of not throwing …

1063b25

…any exception

sumitaich1998 reviewed Sep 16, 2024

View reviewed changes

Use rule for asserting exception

07f5e53

sumitaich1998 reviewed Sep 16, 2024

View reviewed changes

ekawinataa added 4 commits September 16, 2024 18:05

Add more test case

edf2e4d

add more unit test

0c2d621

feat: Enable multiple underscore parsing

a262868

test: Add test on multiple underscore parsing

d31f834

ekawinataa changed the title ~~Add dynamic source and sink kafka properties~~ feat: Add dynamic source and sink kafka properties Sep 16, 2024

sumitaich1998 approved these changes Sep 16, 2024

View reviewed changes

sumitaich1998 merged commit d7b2709 into main Sep 16, 2024

		public class DaggerKafkaConsumerAdditionalConfigurationsAdaptorTest {

feat: Add dynamic source and sink kafka properties #31

feat: Add dynamic source and sink kafka properties #31

Uh oh!

Conversation

ekawinataa commented Sep 2, 2024

Uh oh!

Uh oh!

ekawinataa commented Sep 10, 2024

Uh oh!

sumitaich1998 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants