Kafka sink #3127

rajeshLovesToCode · 2023-08-09T16:14:18Z

Description

Kafka-sink auto topic creation and schema registration on the fly

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 <[email protected]>

kkondaka · 2023-08-11T05:27:15Z

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

+
+    private static final Integer NUM_OF_PARTITIONS = 3;
+    private static final Short REPLICATION_FACTOR = 1;
+


Nit: rename NUM_OF_PARTITIONS to DEFAULT_NUM_OF_PARTITIONS
rename REPLICATION_FACTOR to DEFAULT_REPLICATION_FACTOR

changes have been incorporated.

kkondaka · 2023-08-11T05:29:06Z

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

            } else if (MessageFormat.AVRO.toString().equalsIgnoreCase(serdeFormat)) {
-                publishAvroMessage(record, topic, key, dataForDlq);
+                publishAvroMessage(record, key);


Why this check Why not serdeFormat == MessageFormat.AVRO?

changes have been incorporated.

kkondaka · 2023-08-11T05:35:36Z

...ka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/service/SchemaService.java

+import java.util.Map;
+import java.util.Properties;
+
+public class SchemaService {


Does this only support Confluent Schema Registry? How about AWS GLUE Registry? Even if you do not implement it, it would be great if this can support that without having to change much.

Right now it only supports confluent schema registry and we are not sure about AWS GLUE Registry.

kkondaka · 2023-08-11T05:39:09Z

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

@@ -97,10 +105,12 @@ private void doInitializeInternal() {

    @Override
    public void doOutput(Collection<Record<Event>> records) {
+        reentrantLock.lock();


What is this lock protecting? Is it possible to use synchronized(<object>) { instead

we have imlemented this as other sinks like s3 sink.

Signed-off-by: rajeshLovesToCode <[email protected]>

hshardeesi · 2023-08-18T20:43:36Z

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

+
+  public static class S3FileConfig {
+    @Valid
+    @Size(max = 0, message = "Schema from file is not supported.")


all 3 messages needs to be updated as per the field

changes have been incorporated.

hshardeesi · 2023-08-18T20:44:45Z

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

+    return s3FileConfig;
+  }
+
+  @AssertTrue(message = "Only one of Inline schema or Schema file location or S3 file config config must be specified")


minro: config repeated 2 times after S3 file

changes have been incorporated.

hshardeesi · 2023-08-18T20:52:31Z

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

+  @JsonProperty("s3_file_config")
+  private S3FileConfig s3FileConfig;
+
+  @JsonProperty("is_create")


not sure what does is_create mean? create topic? should be renamed as per functionality

changes have been incorporated.

hshardeesi · 2023-08-18T20:54:32Z

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

@@ -38,6 +38,11 @@ public class TopicConfig {
    static final Integer DEFAULT_NUM_OF_WORKERS = 2;
    static final Duration DEFAULT_HEART_BEAT_INTERVAL_DURATION = Duration.ofSeconds(5);

+
+    private static final Integer DEFAULT_NUM_OF_PARTITIONS = 3;


pleasae set values as per kafka defaults. if its 3 and then ignore this comment.

changes have been incorporated.

hshardeesi · 2023-08-18T21:00:30Z

...fka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/service/TopicService.java

+        try {
+            final NewTopic newTopic = new NewTopic(topicName, numberOfPartitions, replicationFactor);
+            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
+            LOG.info(topicName + " created successfully");


There's also retention period.

changes have been incorporated.

hshardeesi

Few minro comments, can be addressed in next review as well.

Signed-off-by: rajeshLovesToCode <[email protected]>

* -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <[email protected]>

rajeshLovesToCode and others added 13 commits July 26, 2023 21:06

-Support for kafka-sink

e252f19

Signed-off-by: rajeshLovesToCode <[email protected]>

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

288dded

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

4509740

-Support for kafka-sink

5e33025

Signed-off-by: rajeshLovesToCode <[email protected]>

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

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

042fcca

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

ce747f6

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

b102fe7

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

3327799

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

18506b0

-Support for kafka-sink

2708c67

Signed-off-by: rajeshLovesToCode <[email protected]>

rajeshLovesToCode requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners August 9, 2023 16:14

rajeshLovesToCode and others added 2 commits August 10, 2023 05:54

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

6daf21d

-Support for kafka-sink

a92caa1

Signed-off-by: rajeshLovesToCode <[email protected]>

kkondaka reviewed Aug 11, 2023

View reviewed changes

rajeshLovesToCode and others added 4 commits August 11, 2023 06:36

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

ffec16e

-Support for kafka-sink

574c5f9

Signed-off-by: rajeshLovesToCode <[email protected]>

-Support for kafka-sink

3a73019

Signed-off-by: rajeshLovesToCode <[email protected]>

-Support for kafka-sink

6050fa6

Signed-off-by: rajeshLovesToCode <[email protected]>

rajeshLovesToCode requested a review from kkondaka August 16, 2023 07:32

hshardeesi reviewed Aug 18, 2023

View reviewed changes

hshardeesi approved these changes Aug 18, 2023

View reviewed changes

kkondaka previously approved these changes Aug 18, 2023

View reviewed changes

rajeshLovesToCode added 2 commits August 21, 2023 16:03

-Support for kafka-sink

19da476

Signed-off-by: rajeshLovesToCode <[email protected]>

-Support for kafka-sink

d427f80

Signed-off-by: rajeshLovesToCode <[email protected]>

rajeshLovesToCode dismissed kkondaka’s stale review via d427f80 August 21, 2023 10:37

rajeshLovesToCode and others added 7 commits August 21, 2023 16:59

-Support for kafka-sink

c09e92a

Signed-off-by: rajeshLovesToCode <[email protected]>

-Support for kafka-sink

eb97d1c

Signed-off-by: rajeshLovesToCode <[email protected]>

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

1490dcb

-Support for kafka-sink

a8f6547

Signed-off-by: rajeshLovesToCode <[email protected]>

Merge remote-tracking branch 'origin/kafka-sink' into kafka-sink

fc32f7c

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

aa0fe36

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

f979bbc

rajeshLovesToCode requested a review from kkondaka August 25, 2023 06:08

kkondaka approved these changes Aug 26, 2023

View reviewed changes

kkondaka merged commit b5e38e3 into opensearch-project:main Aug 26, 2023
24 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka sink #3127

Kafka sink #3127

rajeshLovesToCode commented Aug 9, 2023

kkondaka Aug 11, 2023

rajeshLovesToCode Aug 16, 2023

kkondaka Aug 11, 2023

rajeshLovesToCode Aug 16, 2023

kkondaka Aug 11, 2023

rajeshLovesToCode Aug 11, 2023

kkondaka Aug 11, 2023

rajeshLovesToCode Aug 11, 2023

hshardeesi Aug 18, 2023

rajeshLovesToCode Aug 21, 2023

hshardeesi Aug 18, 2023

rajeshLovesToCode Aug 21, 2023

hshardeesi Aug 18, 2023

rajeshLovesToCode Aug 21, 2023

hshardeesi Aug 18, 2023

rajeshLovesToCode Aug 21, 2023

hshardeesi Aug 18, 2023

rajeshLovesToCode Aug 21, 2023

hshardeesi left a comment


		private static final Integer NUM_OF_PARTITIONS = 3;
		private static final Short REPLICATION_FACTOR = 1;

Kafka sink #3127

Kafka sink #3127

Conversation

rajeshLovesToCode commented Aug 9, 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

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