-Support for kafka-sink #2999

rajeshLovesToCode · 2023-07-10T15:54:13Z

Description

Kafka 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.

hshardeesi · 2023-07-14T22:39:54Z

...ain/java/org/opensearch/dataprepper/plugins/kafka/util/AuthenticationPropertyConfigurer.java

+        properties.put(SASL_SECURITY_PROTOCOL, SASL_SSL_PROTOCOL);
+        properties.put("ssl.endpoint.identification.algorithm",
+                kafkaSinkConfig.getAuthConfig().getSslConfig().getSslEndpointAlgorithm());
+        if("USER_INFO".equalsIgnoreCase(kafkaSinkConfig.getAuthConfig().getSslConfig().getBasicAuthSource())){


I am not quite getting the purpose of USER_INFO

This is dedicated when you want to connect ot confluent schema registry.

hshardeesi · 2023-07-14T22:40:49Z

data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines-sink-ssl.yaml

+            region: "ap-south-1"
+
+        serde_format: plaintext
+        topics:


Whats the use case of having more than 1 topics in sink? are we going to write same message to multiple topics?

yes, its as per the HLD.

hshardeesi · 2023-07-14T22:41:18Z

data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines-sink-ssl.yaml

+          sasl_ssl:
+            ssl_endpoint_identification_algorithm: https
+            basic_auth_credentials_source: USER_INFO
+            api_key: XXXXXXXXXXX


Whats the difference between api_key/secret and username/passowrd?

api key and secret is for schema registry. client_key and client_secret can be compared to username and password.

hshardeesi · 2023-07-14T22:42:02Z

data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines-sink-ssl.yaml

+        schema:
+          #registry_url: https://psrc-znpo0.ap-southeast-2.aws.confluent.cloud
+          version: 1
+        authentication:


There is authentication block in source also. I think it should gel well with what we accept in source config. please check with krishna on authentication block

ok we will make the changes.

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

rajeshLovesToCode · 2023-07-26T15:38:49Z

updated changes.

kkondaka · 2023-07-28T04:47:29Z

...plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java

+        @JsonProperty("ssl_endpoint_identification_algorithm")
+        private String sslEndpointAlgorithm;
+
+        @JsonProperty("plain_config")


Why this? The plaintext config above has the exactly same config.

Yes , we have incorporated the changes.

kkondaka · 2023-07-28T04:49:31Z

...lugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/PlainConfig.java

@@ -0,0 +1,26 @@
+package org.opensearch.dataprepper.plugins.kafka.configuration;
+


This is pretty much same as PlainTextConfig

Yes , we have incorporated the changes.

kkondaka · 2023-07-28T05:15:27Z

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

+    }
+
+    private void publishPlaintextMessage(Record<Event> record, TopicConfig topic, String key, Object dataForDlq) {
+        producer.send(new ProducerRecord(topic.getName(), key, record.getData().toJsonString()), callBack(dataForDlq));


I just realized that you are not handling "tagsTargetKey". I think Umair/Omkar is familiar with this. You can also see OpenSearchSink code for example.

Yes , we have incorporated the changes.

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

hshardeesi · 2023-08-04T23:01:35Z

...ain/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaProducerProperties.java

+    }
+
+    public int getRetries() {
+        if (retries == 0) {


might be better to assign default value?

kafka already has default values.

hshardeesi · 2023-08-04T23:06:04Z

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

+        if (record.getData().getEventHandle() != null) {
+            bufferedEventHandles.add(record.getData().getEventHandle());
+        }
+        TopicConfig topic = kafkaSinkConfig.getTopic();


This can be a class lvel property

we will incorporate this change in next PR.

hshardeesi · 2023-08-04T23:06:43Z

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

+        Event event = getEvent(record);
+        final String key = event.formatString(kafkaSinkConfig.getPartitionKey(), expressionEvaluator);
+        Object dataForDlq = event.toJsonString();
+        LOG.info("Producing record " + dataForDlq);


this has to be removed ..

we will incorporate this change in next PR.

hshardeesi · 2023-08-04T23:07:15Z

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

+        Object dataForDlq = event.toJsonString();
+        LOG.info("Producing record " + dataForDlq);
+        try {
+            final String serdeFormat = kafkaSinkConfig.getSerdeFormat();


this also seem to be a good candidate for class level property

we will incorporate this change in next PR.

hshardeesi · 2023-08-04T23:09:17Z

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

+    private void publishJsonMessage(Record<Event> record, TopicConfig topic, String key, Object dataForDlq) throws IOException, RestClientException, ProcessingException {
+        final JsonNode dataNode = new ObjectMapper().convertValue(record.getData().toJsonString(), JsonNode.class);
+        if (validateJson(topic.getName(), dataForDlq)) {
+            producer.send(new ProducerRecord(topic.getName(), key, dataNode), callBack(dataForDlq));


producer.send is repeated in all functions. functions can be refarctoed to getJsonMessage etc and producer.send at one place

we will incorporate this change in next PR.

hshardeesi · 2023-08-04T23:13:42Z

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

+        }
+    }
+
+    private boolean validateSchema(final String jsonData, final String schemaJson) throws IOException, ProcessingException {


does this need to happen for every produce record?

yes , this is implemented as per the requirement.

hshardeesi · 2023-08-04T23:19:42Z

...ins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/sink/KafkaSink.java

+        }
+        try {
+            final KafkaSinkProducer producer = createProducer();
+            records.forEach(record -> {


For each record do we need to create Producer worker.? Please check how Consumer thread is created,

yes, this is as per the requirement only.

hshardeesi · 2023-08-04T23:21:13Z

...ins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/sink/KafkaSink.java

+            Map propertiesMap = schemaProps;
+            return new CachedSchemaRegistryClient(
+                    kafkaSinkConfig.getSchemaConfig().getRegistryURL(),
+                    100, propertiesMap);


What is 100? can you please define a constant

we will incorporate this change in next PR.

hshardeesi

Left few comments. I am approving it, but some of the items in produce record code path require refactoring/optimizations.

rajeshLovesToCode requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners July 10, 2023 15:54

rajeshLovesToCode mentioned this pull request Jul 11, 2023

-Support for kafka-sink #2950

Closed

4 tasks

hshardeesi reviewed Jul 14, 2023

View reviewed changes

rajeshLovesToCode closed this Jul 25, 2023

rajeshLovesToCode force-pushed the kafka-sink branch 2 times, most recently from ba3bcd1 to 07f00c2 Compare July 25, 2023 07:38

-Support for kafka-sink

e252f19

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

rajeshLovesToCode reopened this Jul 26, 2023

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

288dded

kkondaka previously approved these changes Jul 28, 2023

View reviewed changes

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

4509740

kkondaka reviewed Jul 28, 2023

View reviewed changes

-Support for kafka-sink

5e33025

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

rajeshLovesToCode dismissed kkondaka’s stale review via 5e33025 July 28, 2023 11:20

kkondaka approved these changes Jul 28, 2023

View reviewed changes

rajeshLovesToCode added 3 commits July 30, 2023 19:01

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

3370ba2

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

3b59f68

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

996d99f

rajeshLovesToCode added 2 commits August 3, 2023 10:38

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

042fcca

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

ce747f6

hshardeesi reviewed Aug 4, 2023

View reviewed changes

hshardeesi approved these changes Aug 4, 2023

View reviewed changes

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

b102fe7

kkondaka merged commit 0582788 into opensearch-project:main Aug 7, 2023
24 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

-Support for kafka-sink #2999

-Support for kafka-sink #2999

rajeshLovesToCode commented Jul 10, 2023

hshardeesi Jul 14, 2023

rajeshLovesToCode Jul 18, 2023

hshardeesi Jul 14, 2023

rajeshLovesToCode Jul 18, 2023

hshardeesi Jul 14, 2023

rajeshLovesToCode Jul 18, 2023

hshardeesi Jul 14, 2023

rajeshLovesToCode Jul 18, 2023

rajeshLovesToCode commented Jul 26, 2023

kkondaka Jul 28, 2023

rajeshLovesToCode Jul 28, 2023

kkondaka Jul 28, 2023

rajeshLovesToCode Jul 28, 2023

kkondaka Jul 28, 2023

rajeshLovesToCode Jul 28, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi Aug 4, 2023

rajeshLovesToCode Aug 7, 2023

hshardeesi left a comment

		@@ -0,0 +1,26 @@
		package org.opensearch.dataprepper.plugins.kafka.configuration;

-Support for kafka-sink #2999

-Support for kafka-sink #2999

Conversation

rajeshLovesToCode commented Jul 10, 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

rajeshLovesToCode commented Jul 26, 2023

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

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

hshardeesi left a comment

Choose a reason for hiding this comment