Updates Kafka configurations such that plugin has its own topic configurations #3551

dlvenable · 2023-10-25T21:48:59Z

Description

Prior to this PR, all three Kafka plugins shared a common configuration for topic configurations - TopicConfig. This is somewhat problematic as each plugin does have different configurations for the topic depending on what the plugin is doing (producing, consuming, or both).

Some of the configuration changes:

Removes serde_format from the Kafka buffer topic configuration. The Kafka buffer will always use bytes.
Removes schema from the Kafka buffer configuration as it won't use a schema registry.
Removes number_of_partitions, replication_factor, retention_period, is_topic_create from the Kafka source's topic configuration.
Removes commit_interval and key_mode from the Kafka sink's topic configuration.

As part of this change, I split the TopicConfig into three distinct interfaces and classes. This allows each plugin to either accept a configuration or provide a value of the plugin's own choosing.

This PR also adds copyright headers to all files modified as part of this commit.

Issues Resolved

None, but contributes toward #3322.

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- 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.

…are not needed - schema and the serde_format for the topic configuration. As part of this change, I also split the TopicConfig into three distinct interfaces and classes. This allows each plugin to either accept a configuration or provide a value of the plugin's own choosing. Also adds copyright headers to all files modified as part of this commit. Signed-off-by: David Venable <[email protected]>

kkondaka · 2023-10-25T22:39:22Z

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

+
+    @JsonProperty("fetch_max_bytes")
+    private String fetchMaxBytes = DEFAULT_FETCH_MAX_BYTES;
+
+    @JsonProperty("fetch_max_wait")
+    @Valid
+    @Size(min = 1)
+    private Integer fetchMaxWait = DEFAULT_FETCH_MAX_WAIT;
+
+    @JsonProperty("fetch_min_bytes")
+    private String fetchMinBytes = DEFAULT_FETCH_MIN_BYTES;


I think a lot of these are not applicable to Producer. Things like fetch_max_bytes, auto_commit etc are consumer specific.

Thanks! I wasn't quite sure if I had caught them all or not.

… and buffer configurations. Also cleaned up the CommonTopicConfig somewhat. Adds a zeroBytes() static method to ByteCount as a convenience. Signed-off-by: David Venable <[email protected]>

hshardeesi · 2023-10-27T19:54:24Z

...plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/BufferTopicConfig.java

+    @JsonProperty("retention_period")
+    private Long retentionPeriod = DEFAULT_RETENTION_PERIOD;
+
+    @JsonProperty("is_topic_create")


there is some confusion on meaning of this property. if this is for whether kafka buffer should create topic then may be its should called createTopic.

I agree with this comment. I didn't change the name to avoid breaking anything, but this is only used in the sink and buffer, neither of which have been released. So I'll change this to create_topic.

…ernally instead of Boolean since it will have a value. Signed-off-by: David Venable <[email protected]>

hshardeesi · 2023-10-27T20:15:49Z

...rc/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/ProducerTopicConfig.java

+ * An extension of the {@link TopicConfig} specifically for
+ * producers to Kafka topics.
+ */
+public interface ProducerTopicConfig extends TopicConfig {


nit: ProducerTopicConfig and ConsumerTopicConfig gives impression that there are 2 different topics one for producing and one for consuming data. may TopicProducerConfig or TopicConfigProducer may be better, but not a blocker.

@hshardeesi , That suggestion makes sense and I hadn't seen it from that perspective until you called it out. I went ahead and renamed the files.

…picConfig and ProducerTopicConfig. Signed-off-by: David Venable <[email protected]>

kkondaka · 2023-10-27T22:19:04Z

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

+    static final Duration DEFAULT_THREAD_WAITING_TIME = Duration.ofSeconds(5);
+    static final Duration DEFAULT_RETRY_BACKOFF = Duration.ofSeconds(10);
+    static final Duration DEFAULT_RECONNECT_BACKOFF = Duration.ofSeconds(10);
+    static final Duration DEFAULT_MAX_POLL_INTERVAL = Duration.ofSeconds(300);


MAX_POLL_INTERVAL also doesn't belong in the common. common should be intersection of Producer and Consumer. We can get this info from https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html and https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html.

This is the common list between the two. At the topic level it looks like only the following two are common?

reconnect.backoff.ms retry.backoff.ms

Things like bootstrap servers, auth (ssl, sasl, oauth configs) are common between source and sink. But the at topic level very few are common.

… along with the corresponding subclasses. Signed-off-by: David Venable <[email protected]>

kkondaka · 2023-10-30T21:49:40Z

...fka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/sink/KafkaSinkConfig.java

@@ -61,7 +67,7 @@ public void setDlqConfig(final PluginSetting pluginSetting) {


    @JsonProperty("topic")
-    TopicConfig topic;
+    SinkTopicConfig topic;

    @JsonProperty("authentication")
    private AuthConfig authConfig;


I should have pointed this out previously. It looks like "AuthConfig" and "AwsConfig" and "bootstrapServers", "SchemaConfig" are common between "KafkaSourceConfig" and "KafkaSinkConfig", does it make sense to move put these in a common interface and a common base class?

Same with getThreadWaitTime()/getThreadWaitingTime()

I think that is a good idea, but probably beyond the scope of this PR. The goal of this PR is to decouple the topic configurations as they do need to vary between the different plugins.

dlvenable requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed and oeyh as code owners October 25, 2023 21:49

kkondaka reviewed Oct 25, 2023

View reviewed changes

Moved more fields from the common topic configuration into the source…

94dbb51

… and buffer configurations. Also cleaned up the CommonTopicConfig somewhat. Adds a zeroBytes() static method to ByteCount as a convenience. Signed-off-by: David Venable <[email protected]>

hshardeesi reviewed Oct 27, 2023

View reviewed changes

Renamed is_topic_create to create_topic. Also made this a boolean int…

c447d98

…ernally instead of Boolean since it will have a value. Signed-off-by: David Venable <[email protected]>

hshardeesi reviewed Oct 27, 2023

View reviewed changes

hshardeesi approved these changes Oct 27, 2023

View reviewed changes

Use TopicConsumerConfig and TopicProducerConfig instead of ConsumerTo…

62fe0cc

…picConfig and ProducerTopicConfig. Signed-off-by: David Venable <[email protected]>

hshardeesi approved these changes Oct 27, 2023

View reviewed changes

kkondaka reviewed Oct 27, 2023

View reviewed changes

Move more fields from the common TopicConfig into TopicConsumerConfig…

6dcd581

… along with the corresponding subclasses. Signed-off-by: David Venable <[email protected]>

hshardeesi approved these changes Oct 30, 2023

View reviewed changes

kkondaka reviewed Oct 30, 2023

View reviewed changes

kkondaka approved these changes Oct 31, 2023

View reviewed changes

dlvenable merged commit 7323c9e into opensearch-project:main Oct 31, 2023
36 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Updates Kafka configurations such that plugin has its own topic configurations #3551

Updates Kafka configurations such that plugin has its own topic configurations #3551

dlvenable commented Oct 25, 2023 •

edited

Loading

kkondaka Oct 25, 2023

dlvenable Oct 26, 2023

hshardeesi Oct 27, 2023

dlvenable Oct 27, 2023

hshardeesi Oct 27, 2023

dlvenable Oct 27, 2023

kkondaka Oct 27, 2023

kkondaka Oct 27, 2023

kkondaka Oct 30, 2023

kkondaka Oct 30, 2023

dlvenable Oct 31, 2023

Updates Kafka configurations such that plugin has its own topic configurations #3551

Updates Kafka configurations such that plugin has its own topic configurations #3551

Conversation

dlvenable commented Oct 25, 2023 • edited Loading

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

dlvenable commented Oct 25, 2023 •

edited

Loading