-Support for kafka-sink #2950

rajeshLovesToCode · 2023-06-28T16:38:51Z

Signed-off-by: rajeshLovesToCode rajesh.dharamdasani3021@gmail.com

Description

Kafk Sink with SASL_PLAINTEXT, OAuth, schema registry, multithreaded producer and DLQ.

Issues Resolved

Github issue #1986

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…ani3021@gmail.com> Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…haramdasani3021@gmail.com>" This reverts commit fff5398.

* feat: add include_key options to KeyValueProcessor Signed-off-by: Haidong <whaidong@amazon.com> --------- Signed-off-by: Haidong <whaidong@amazon.com> Co-authored-by: Haidong <whaidong@amazon.com> Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…-project#2976) * Added Kafka config to support acknowledgments and MSK arn Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Modified to use data-prepper-core in testImplementation Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Addressed failing test Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> Co-authored-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

into kafka-sink-draft

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

kkondaka · 2023-07-06T15:43:48Z

...ugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsDLQConfig.java

+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Size;
+
+public class AwsDLQConfig {


Let's merge this with AwsConfig.java

This file is no longer available. Please review the updated PR #2999.

kkondaka · 2023-07-06T15:49:36Z

...ugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaSinkProducer.java

+                String serdeFormat=kafkaSinkConfig.getSerdeFormat();
+                if (MessageFormat.JSON.toString().equalsIgnoreCase(serdeFormat)) {
+                    JsonNode dataNode = new ObjectMapper().convertValue(record.getData().toJsonString(), JsonNode.class);
+                    dataForDlq = dataNode;


Can this be set in the exception path? We don't need this in the normal path, right?

Please elaborate.

kkondaka · 2023-07-06T15:54:50Z

...ugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaSinkProducer.java

+                    Schema schema =new Schema.Parser().parse(valueToParse);
+                    GenericRecord genericRecord = getGenericRecord(record.getData(),schema);
+                    dataForDlq = genericRecord;
+                    producer.send(new ProducerRecord(topic.getName(), genericRecord));


Also you should release event handle once the event is successfully sent to Kafka or to the DLQ

Please review the updated PR #2999.

kkondaka · 2023-07-06T15:59:50Z

...s/src/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsDlqConfigTest.java

+	void test_aws_props(){
+		assertThat(awsDLQConfig.getBucket(), notNullValue());
+		assertThat(awsDLQConfig.getRegion(), notNullValue());
+		assertThat(awsDLQConfig.getRoleArn(), notNullValue());


Assertions should be more than not null. For example, You can have something like String testArn which can be used in the setup and then can be compared in the assertion here. Take a look at the unit test cases added recently in the KafkaSource, like KafkaSourceTest.java.

This file is no longer available. Please review the updated PR #2999.

kkondaka · 2023-07-06T16:01:49Z

...rc/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaSinkConfigTest.java

+	@BeforeEach
+	void setUp() throws IOException {
+		//Added to load Yaml file - Start
+		Yaml yaml = new Yaml();


Prefer this to be not read from a YAML file. You can set fields being tested and mock everything else.

This has been addressed. Please review the updated PR #2999.

kkondaka · 2023-07-06T16:07:07Z

...s/src/test/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaSinkProducerTest.java

+        producer = new KafkaSinkProducer(mockProducer, kafkaSinkConfig,dlqSink);
+        sinkProducer = spy(producer);
+        sinkProducer.produceRecords(record);
+        verify(sinkProducer).produceRecords(record);


This is not enough. We need to assert that the records are put in Kafka properly by mocking the KafkaProducer. See KafkaSourceCustomerConsumerTest.java on how this can be done.

This has been addressed. Please review the updated PR #2999.

kkondaka · 2023-07-06T16:07:51Z

...gins/src/test/java/org/opensearch/dataprepper/plugins/kafka/producer/ProducerWorkerTest.java

+        multithreadedProducer = createObjectUnderTest();
+        Thread spySink = spy(new Thread(multithreadedProducer));
+        spySink.start();
+        verify(spySink).start();


We need to be able to assert that it has functioned as expected in addition to verify.

This file does not have any scope for assertions but produce method has some proper assertions in KafkaSinkProducerTest. Please review the updated PR #2999 .

…pper into kafka-sink-draft Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…k-draft # Conflicts: # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsDLQConfig.java # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaSinkConfig.java # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/sink/DLQSink.java # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/sink/KafkaSink.java # data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/SinkPropertyConfigurer.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsDlqConfigTest.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaSinkConfigTest.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaSinkProducerTest.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/producer/ProducerWorkerTest.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/sink/DLQSinkTest.java # data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/sink/KafkasinkTest.java

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…de/data-prepper into HEAD Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

rajeshLovesToCode · 2023-07-11T09:07:46Z

we are closing this PR as we have raised the new PR #2999.

rajeshLovesToCode and others added 9 commits June 28, 2023 21:50

-Support for kafka-sink

3f4d9b4

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

-Support for kafka-sink

4173487

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Merge branch 'opensearch-project:main' into kafka-sink-draft

4118511

Merge branch 'opensearch-project:main' into kafka-sink-draft

155ef8c

Fixing import issues

0350412

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

fixing build issue Signed-off-by: rajeshLovesToCode <rajesh.dharamdas…

fff5398

…ani3021@gmail.com> Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Revert "fixing build issue Signed-off-by: rajeshLovesToCode <rajesh.d…

3d7f695

…haramdasani3021@gmail.com>" This reverts commit fff5398.

rajeshLovesToCode force-pushed the kafka-sink-draft branch from 4a4f1d6 to 657e89c Compare July 5, 2023 08:06

rajeshLovesToCode and others added 4 commits July 5, 2023 13:38

Merge branch 'opensearch-project:main' into kafka-sink-draft

63a6de4

-Support for kafka-sink

66696bf

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Merge branch 'main' of https://github.com/rajeshLovesToCode/data-prepper

c1ee576

into kafka-sink-draft

Merge branch 'opensearch-project:main' into kafka-sink-draft

aeb6aba

ajeeshakd mentioned this pull request Jul 6, 2023

Kafka Sink - Implementation of configuration, multithreading and plaintext, oauth authentication mechanisms #2889

Closed

4 tasks

rajeshLovesToCode marked this pull request as ready for review July 6, 2023 13:28

rajeshLovesToCode requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners July 6, 2023 13:28

-Support for kafka-sink

7857192

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

kkondaka reviewed Jul 6, 2023

View reviewed changes

rajeshLovesToCode and others added 3 commits July 7, 2023 13:12

Merge branch 'opensearch-project:main' into kafka-sink-draft

a25d8c0

Merge branch 'main' of https://github.com/opensearch-project/data-pre…

070b3be

…pper into kafka-sink-draft Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

-Support for kafka-sink

379bc07

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

rajeshLovesToCode and others added 6 commits July 7, 2023 22:11

Fixing import issues

53dca9b

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Merge branch 'kafka-sink-draft' of https://github.com/rajeshLovesToCo…

5d1f427

…de/data-prepper into HEAD Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

-Support for kafka-sink

280e521

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

Merge branch 'opensearch-project:main' into kafka-sink-draft

1139aec

-Support for kafka-sink.

aeb3496

Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

rajeshLovesToCode closed this Jul 11, 2023

rajeshLovesToCode deleted the kafka-sink-draft branch September 21, 2023 07:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

-Support for kafka-sink #2950

-Support for kafka-sink #2950

rajeshLovesToCode commented Jun 28, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 7, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

kkondaka Jul 6, 2023

rajeshLovesToCode Jul 11, 2023

rajeshLovesToCode commented Jul 11, 2023

-Support for kafka-sink #2950

-Support for kafka-sink #2950

Conversation

rajeshLovesToCode commented Jun 28, 2023

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajeshLovesToCode commented Jul 11, 2023