-Support for kafka-sink (#2999)

* -Support for kafka-sink Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com> * -Support for kafka-sink Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>
opensearch-project · Aug 7, 2023 · 0582788 · 0582788
1 parent 259fea1
commit 0582788
Show file tree

Hide file tree

Showing 20 changed files with 2,271 additions and 3 deletions.
diff --git a/data-prepper-plugins/kafka-plugins/build.gradle b/data-prepper-plugins/kafka-plugins/build.gradle
@@ -43,6 +43,13 @@ dependencies {
     testImplementation 'org.apache.kafka:kafka-clients:3.4.0:test'
     testImplementation 'org.apache.kafka:connect-json:3.4.0'
     testImplementation('com.kjetland:mbknor-jackson-jsonschema_2.13:1.0.39')
+    implementation project(':data-prepper-plugins:failures-common')
+    testImplementation group: 'org.powermock', name: 'powermock-api-mockito2', version: '2.0.9'
+    implementation 'org.apache.kafka:connect-json:3.4.0'
+    implementation 'com.github.fge:json-schema-validator:2.2.14'
+    implementation 'commons-collections:commons-collections:3.2.2'
+
+
 }
 
 test {

diff --git a/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java b/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java
@@ -6,8 +6,8 @@
 package org.opensearch.dataprepper.plugins.kafka.configuration;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.constraints.AssertTrue;
 import jakarta.validation.Valid;
+import jakarta.validation.constraints.AssertTrue;
 
 import java.util.stream.Stream;
 
@@ -27,6 +27,9 @@ public static class SaslAuthConfig {
         @JsonProperty("aws_msk_iam")
         private AwsIamAuthConfig awsIamAuthConfig;
 
+        @JsonProperty("ssl_endpoint_identification_algorithm")
+        private String sslEndpointAlgorithm;
+
         public AwsIamAuthConfig getAwsIamAuthConfig() {
             return awsIamAuthConfig;
         }
@@ -39,14 +42,19 @@ public OAuthConfig getOAuthConfig() {
             return oAuthConfig;
         }
 
+        public String getSslEndpointAlgorithm() {
+            return sslEndpointAlgorithm;
+        }
+
         @AssertTrue(message = "Only one of AwsIam or oAuth or PlainText auth config must be specified")
         public boolean hasOnlyOneConfig() {
-            return Stream.of(awsIamAuthConfig, plainTextAuthConfig, oAuthConfig).filter(n -> n!=null).count() == 1;
+            return Stream.of(awsIamAuthConfig, plainTextAuthConfig, oAuthConfig).filter(n -> n != null).count() == 1;
         }
 
     }
 
-    public static  class SslAuthConfig {
+
+    public static class SslAuthConfig {
         // TODO Add Support for SSL authentication types like
         // one-way or two-way authentication
 

diff --git a/.../java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaProducerProperties.java b/.../java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaProducerProperties.java
@@ -0,0 +1,250 @@
+package org.opensearch.dataprepper.plugins.kafka.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.Duration;
+import java.util.List;
+
+public class KafkaProducerProperties {
+    private static final String DEFAULT_BYTE_CAPACITY = "50mb";
+
+    @JsonProperty("buffer_memory")
+    private String bufferMemory = DEFAULT_BYTE_CAPACITY;
+
+    @JsonProperty("compression_type")
+    private String compressionType;
+
+    @JsonProperty("retries")
+    private int retries;
+
+    @JsonProperty("batch_size")
+    private int batchSize;
+
+    @JsonProperty("client_dns_lookup")
+    private String clientDnsLookup;
+
+    @JsonProperty("client_id")
+    private String clientId;
+
+    @JsonProperty("connections_max_idle")
+    private Duration connectionsMaxIdleMs;
+
+    @JsonProperty("delivery_timeout")
+    private Duration deliveryTimeoutMs;
+
+    @JsonProperty("linger_ms")
+    private Long lingerMs;
+
+    @JsonProperty("max_block")
+    private Duration maxBlockMs;
+
+    @JsonProperty("max_request_size")
+    private int maxRequestSize;
+
+    @JsonProperty("partitioner_class")
+    private Class partitionerClass;
+
+    @JsonProperty("partitioner_ignore_keys")
+    private Boolean partitionerIgnoreKeys;
+
+    @JsonProperty("receive_buffer")
+    private String receiveBufferBytes=DEFAULT_BYTE_CAPACITY;
+
+    @JsonProperty("request_timeout")
+    private Duration requestTimeoutMs;
+
+    @JsonProperty("send_buffer")
+    private String sendBufferBytes=DEFAULT_BYTE_CAPACITY;
+
+    @JsonProperty("socket_connection_setup_timeout_max")
+    private Duration socketConnectionSetupMaxTimeout;
+
+    @JsonProperty("socket_connection_setup_timeout")
+    private Duration socketConnectionSetupTimeout;
+
+    @JsonProperty("acks")
+    private String acks;
+
+    @JsonProperty("enable_idempotence")
+    private Boolean enableIdempotence;
+
+    @JsonProperty("interceptor_classes")
+    private List interceptorClasses;
+
+    @JsonProperty("max_in_flight_requests_per_connection")
+    private int maxInFlightRequestsPerConnection;
+
+    @JsonProperty("metadata_max_age")
+    private Duration metadataMaxAgeMs;
+
+    @JsonProperty("metadata_max_idle")
+    private Duration metadataMaxIdleMs;
+
+    @JsonProperty("metric_reporters")
+    private List metricReporters;
+
+    @JsonProperty("metrics_num_samples")
+    private int metricsNumSamples;
+
+    @JsonProperty("metrics_recording_level")
+    private String metricsRecordingLevel;
+
+    @JsonProperty("metrics_sample_window")
+    private Duration metricsSampleWindowMs;
+
+    @JsonProperty("partitioner_adaptive_partitioning_enable")
+    private boolean partitionerAdaptivePartitioningEnable;
+
+    @JsonProperty("partitioner_availability_timeout")
+    private Duration partitionerAvailabilityTimeoutMs;
+
+    @JsonProperty("reconnect_backoff_max")
+    private Duration reconnectBackoffMaxMs;
+
+    @JsonProperty("reconnect_backoff")
+    private Duration reconnectBackoffMs;
+
+    @JsonProperty("retry_backoff")
+    private Duration retryBackoffMs;
+
+
+    public String getCompressionType() {
+        return compressionType;
+    }
+
+    public int getRetries() {
+        if (retries == 0) {
+            retries = 5;
+        }
+        return retries;
+    }
+
+    public int getBatchSize() {
+        return batchSize;
+    }
+
+    public String getClientDnsLookup() {
+        return clientDnsLookup;
+    }
+
+    public String getClientId() {
+        return clientId;
+    }
+
+
+    public Long getLingerMs() {
+        return lingerMs;
+    }
+
+
+    public int getMaxRequestSize() {
+        return maxRequestSize;
+    }
+
+    public Class getPartitionerClass() {
+        return partitionerClass;
+    }
+
+    public Boolean getPartitionerIgnoreKeys() {
+        return partitionerIgnoreKeys;
+    }
+
+
+    public String getAcks() {
+        return acks;
+    }
+
+    public Boolean getEnableIdempotence() {
+        return enableIdempotence;
+    }
+
+    public List getInterceptorClasses() {
+        return interceptorClasses;
+    }
+
+    public int getMaxInFlightRequestsPerConnection() {
+        return maxInFlightRequestsPerConnection;
+    }
+
+
+    public List getMetricReporters() {
+        return metricReporters;
+    }
+
+    public int getMetricsNumSamples() {
+        return metricsNumSamples;
+    }
+
+    public String getMetricsRecordingLevel() {
+        return metricsRecordingLevel;
+    }
+
+
+    public boolean isPartitionerAdaptivePartitioningEnable() {
+        return partitionerAdaptivePartitioningEnable;
+    }
+
+    public String getBufferMemory() {
+        return bufferMemory;
+    }
+
+    public Long getConnectionsMaxIdleMs() {
+        return connectionsMaxIdleMs.toMillis();
+    }
+
+    public Long getDeliveryTimeoutMs() {
+        return deliveryTimeoutMs.toMillis();
+    }
+
+    public Long getMaxBlockMs() {
+        return maxBlockMs.toMillis();
+    }
+
+    public String getReceiveBufferBytes() {
+        return receiveBufferBytes;
+    }
+
+    public Long getRequestTimeoutMs() {
+        return requestTimeoutMs.toMillis();
+    }
+
+    public String getSendBufferBytes() {
+        return sendBufferBytes;
+    }
+
+    public Long getSocketConnectionSetupMaxTimeout() {
+        return socketConnectionSetupMaxTimeout.toMillis();
+    }
+
+    public Long getSocketConnectionSetupTimeout() {
+        return socketConnectionSetupTimeout.toMillis();
+    }
+
+    public Long getMetadataMaxAgeMs() {
+        return metadataMaxAgeMs.toMillis();
+    }
+
+    public Long getMetadataMaxIdleMs() {
+        return metadataMaxIdleMs.toMillis();
+    }
+
+    public Long getMetricsSampleWindowMs() {
+        return metricsSampleWindowMs.toMillis();
+    }
+
+    public Long getPartitionerAvailabilityTimeoutMs() {
+        return partitionerAvailabilityTimeoutMs.toMillis();
+    }
+
+    public Long getReconnectBackoffMaxMs() {
+        return reconnectBackoffMaxMs.toMillis();
+    }
+
+    public Long getReconnectBackoffMs() {
+        return reconnectBackoffMs.toMillis();
+    }
+
+    public Long getRetryBackoffMs() {
+        return retryBackoffMs.toMillis();
+    }
+}