Add support for optional kafka SASL #24798

ShahimSharafudeen · 2025-03-26T07:56:42Z

Description

Currently the presto-kafka connector can be configured to use SASL authentication (https://docs.confluent.io/platform/current/security/authentication/sasl/plain/overview.html#kafka-sasl-auth-plain) those config properties were made known to the KafkaConfig and are used in the producer as well as consumer factory if being set (all optional).

As of now, we have configured the following parameters for the above requirement:

security.protocol
sasl.mechanism
sasl.jaas.config
ssl.truststore.location
ssl.truststore.password
ssl.truststore.type

As part of the code-level configuration, represent the parameter naming as below and map them to the corresponding Kafka config parameters:

kafka.security-protocol=SASL_SSL
kafka.sasl.mechanism=PLAIN
kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username= password=<API KEY/PASSWORD>;
kafka.truststore.path=XXX
kafka.truststore.password=XXX
kafka.truststore.type=XXX

Taking reference from Trinodb Kafka SSL implementation: trinodb/trino#6929

Motivation and Context

Impact

Test Plan

Contributor checklist

Please make sure your submission complies with our contributing guide, in particular code style and commit standards.
PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced.
Documented new properties (with its default value), SQL syntax, functions, or other functionality.
If release notes are required, they follow the release notes guidelines.
Adequate tests were added if applicable.
CI passed.

Release Notes

Please follow release notes guidelines and fill in the release notes below.

== RELEASE NOTES ==

Kafka Connector Changes 
* Add support for optional Apache Kafka SASL.

aaneja

Can you attribute trinodb/trino#6929 in your commit message ? See 'backport commits' section from the CONTRIBUTING doc
Can you update the kafka.rst file with the new SASL properties

ShahimSharafudeen · 2025-04-21T12:00:59Z

Can you attribute Add support for optional Kafka SSL trinodb/trino#6929 in your commit message ? See 'backport commits' section from the CONTRIBUTING doc

Can you update the kafka.rst file with the new SASL properties

@aaneja - Addressed the review comments.

steveburnett

Thanks for the doc! Looks good. Some minor formatting suggestions, and one question for you to consider.

presto-docs/src/main/sphinx/connector/kafka.rst

ShahimSharafudeen · 2025-04-22T09:45:18Z

Thanks for the doc! Looks good. Some minor formatting suggestions, and one question for you to consider.

Thanks @steveburnett for reviewing the documentation-related changes. I’ve addressed all your review comments.

aaneja

Can you add a product test that uses these new configs ? We currently have a singlenode-kafka test, lets make a new one called singlenode-kafka-sasl

aaneja · 2025-04-22T14:47:44Z

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaConsumerManager.java

+public class SaslKafkaConsumerManager
+        implements KafkaConsumerManager
+{
+    private final ImmutableMap<String, Object> map;


Change type to Map<String, Object>

aaneja · 2025-04-22T14:58:01Z

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaProducerFactory.java

+public class SaslKafkaProducerFactory
+        implements KafkaProducerFactory
+{
+    private final ImmutableMap<String, Object> map;


Let's rename this to kafkaClientConfigProperties, and make it a Map<String,Object>

aaneja · 2025-04-22T15:13:04Z

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaProducerFactory.java

+        Properties properties = new Properties();
+        properties.putAll(delegate.configure(bootstrapServers));
+        properties.putAll(map);
+        log.info("Loading SASL Configs for Kafka...");


Remove the log - it does not add any information

aaneja · 2025-04-22T15:14:35Z

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaProducerFactory.java

+    public SaslKafkaProducerFactory(@ForKafkaSasl KafkaProducerFactory delegate, KafkaSaslConfig saslConfig)
+    {
+        this.delegate = requireNonNull(delegate, "delegate is null");
+        map = ImmutableMap.copyOf(saslConfig.getKafkaClientProperties());


Why don't we just store a reference to the saslConfig and use it in configure when needed ?

aaneja · 2025-04-22T15:15:58Z

presto-kafka/src/main/java/com/facebook/presto/kafka/PlainTextKafkaProducerFactory.java

-                .build();
-        return new KafkaProducer<>(propertiesWithBootstrapServers, new ByteArraySerializer(), new ByteArraySerializer());
+                        .collect(joining(",")));
+        log.info("Loading Default Configs for Kafka...");


Remove the log, it does not add any value

aaneja · 2025-04-22T15:17:18Z

presto-kafka/src/main/java/com/facebook/presto/kafka/PlainTextKafkaProducerFactory.java

                .put(ACKS_CONFIG, "all")
                .put(LINGER_MS_CONFIG, 5)
                .build();
+        configurationProperties = readProperties(kafkaConfig.getResourceConfigFiles());


Instead of creating two maps, make a single Map<String, Object> called configurationProperties and store the ACKS_CONFIG and LINGER_MS_CONFIG in them

aaneja · 2025-04-22T15:18:05Z

presto-kafka/src/main/java/com/facebook/presto/kafka/PlainTextKafkaProducerFactory.java

    {
-        this.properties = ImmutableMap.<String, Object>builder()
+        this.map = ImmutableMap.<String, Object>builder()
                .put(ACKS_CONFIG, "all")


Why are these non-default values used, and why are they not configurable ?

aaneja · 2025-04-22T15:20:18Z

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaConsumerManager.java

+    public SaslKafkaConsumerManager(@ForKafkaSasl KafkaConsumerManager delegate, KafkaSaslConfig saslConfig)
+    {
+        this.delegate = requireNonNull(delegate, "delegate is null");
+        map = ImmutableMap.copyOf(saslConfig.getKafkaClientProperties());


map is redundant, just use properties.putAll(saslConfig.getKafkaClientProperties()) on L50

aaneja · 2025-04-22T15:21:21Z

presto-kafka/src/main/java/com/facebook/presto/kafka/PlainTextKafkaConsumerManager.java

+        properties.put(GROUP_ID_CONFIG, threadName);
+        properties.put(MAX_POLL_RECORDS_CONFIG, Integer.toString(maxPollRecords));
+        properties.put(MAX_PARTITION_FETCH_BYTES_CONFIG, maxPartitionFetchBytes);
+        properties.put(CLIENT_ID_CONFIG, String.format("%s-%s", threadName, hostAddress.toString()));


toString() is redundant

aaneja · 2025-04-22T15:29:41Z

presto-kafka/src/main/java/com/facebook/presto/kafka/security/KafkaSaslConfig.java

+        return this;
+    }
+
+    public Map<String, Object> getKafkaClientProperties()


Should we rename this to getKafkaSaslProperties()

steveburnett

Thanks for the quick update! I found a few more nits and suggestions, again nothing major.

presto-docs/src/main/sphinx/connector/kafka.rst

ShahimSharafudeen · 2025-04-25T09:17:13Z

@aaneja - I have addressed all your comments regarding the Java-related changes.

ShahimSharafudeen · 2025-04-25T09:19:16Z

Thanks for the quick update! I found a few more nits and suggestions, again nothing major.

Thanks again, @steveburnett, for your suggestions. I have addressed all your comments regarding the doc-related changes.

steveburnett

LGTM! (docs)

Pull updated branch, new local doc build, looks good. Thanks!

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaConsumerManager.java

presto-kafka/src/main/java/com/facebook/presto/kafka/SaslKafkaProducerFactory.java

aaneja · 2025-04-27T10:11:36Z

presto-kafka/src/main/java/com/facebook/presto/kafka/PlainTextKafkaConsumerManager.java

+        final Properties properties = new Properties();
+        properties.put(BOOTSTRAP_SERVERS_CONFIG, hostAddress.toString());
+        properties.put(GROUP_ID_CONFIG, threadName);
+        properties.put(MAX_POLL_RECORDS_CONFIG, Integer.toString(maxPollRecords));


Is the Integer.toString needed ? From a quick glance at ConsumerConfig, it looks like we could use the integer directly

@aaneja - We can pass the value either as an int or a String, but to avoid any issues during runtime, it's better to stick with the existing code pattern used in the initial implementation of the KafkaConsumerManager class, which passes it as a String. Let me know if you’d prefer using an Integer instead — happy to update it.

aaneja · 2025-04-28T06:19:27Z

Please create an issue to implement the singlenode-kafka-sasl product test. This can be punted to later

ShahimSharafudeen · 2025-04-28T07:42:53Z

Please create an issue to implement the singlenode-kafka-sasl product test. This can be punted to later

Git issue : #24993

steveburnett

LGTM! (docs)

Pull updated branch, new local doc build, looks good.

ZacBlanco

Minor things, overall this is good

presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaConnectorModule.java

presto-kafka/src/main/java/com/facebook/presto/kafka/security/KafkaKeystoreTruststoreType.java

presto-kafka/src/test/java/com/facebook/presto/kafka/TestKafkaConnectorConfig.java

ZacBlanco

Can you also rebase this PR on the latest master branch? I noticed the git log doesn't include anything from April

presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaConnectorModule.java

presto-kafka/src/main/java/com/facebook/presto/kafka/security/KafkaKeystoreTruststoreType.java

ShahimSharafudeen · 2025-04-30T18:42:50Z

Can you also rebase this PR on the latest master branch? I noticed the git log doesn't include anything from April

Rebased the PR with the latest commits from the master branch and addressed all your comments.

Cherry-pick of trinodb/trino#6929 Co-authored-by: Kenzyme Le <kl@kenzymele.com> Co-authored-by: Matthias Strobl <matthias.strobl@bmw.de>

steveburnett · 2025-04-30T19:28:33Z

Thanks for the release note! Minor nit suggestions:

== RELEASE NOTES ==

Kafka Connector Changes
* Add support for optional Apache Kafka SASL.

ShahimSharafudeen · 2025-04-30T19:40:21Z

Thanks for the release note! Minor nit suggestions:
== RELEASE NOTES ==

Kafka Connector Changes
* Add support for optional Apache Kafka SASL. 

Done

prestodb-ci added the from:IBM PR from IBM label Mar 26, 2025

ShahimSharafudeen force-pushed the kafka_sasl_config branch 2 times, most recently from 9bc1fd1 to 9cc5a3a Compare March 28, 2025 12:39

ShahimSharafudeen marked this pull request as ready for review April 7, 2025 09:32

ShahimSharafudeen requested a review from a team as a code owner April 7, 2025 09:32

ShahimSharafudeen requested a review from presto-oss April 7, 2025 09:32

prestodb-ci requested review from a team, aaneja and imjalpreet and removed request for a team April 7, 2025 09:32

aaneja requested changes Apr 17, 2025

View reviewed changes

ShahimSharafudeen force-pushed the kafka_sasl_config branch from 9cc5a3a to a004651 Compare April 21, 2025 11:56

ShahimSharafudeen requested review from steveburnett and elharo as code owners April 21, 2025 11:56

steveburnett requested changes Apr 21, 2025

View reviewed changes

ShahimSharafudeen force-pushed the kafka_sasl_config branch from a004651 to a26ab6a Compare April 22, 2025 09:37

aaneja requested changes Apr 22, 2025

View reviewed changes

steveburnett requested changes Apr 22, 2025

View reviewed changes

ShahimSharafudeen force-pushed the kafka_sasl_config branch from 08fbc5e to 6903a7e Compare April 25, 2025 09:11

steveburnett previously approved these changes Apr 25, 2025

View reviewed changes

aaneja previously approved these changes Apr 27, 2025

View reviewed changes

ShahimSharafudeen dismissed stale reviews from aaneja and steveburnett via 6fa2430 April 28, 2025 05:54

ShahimSharafudeen force-pushed the kafka_sasl_config branch from 6903a7e to 6fa2430 Compare April 28, 2025 05:54

aaneja previously approved these changes Apr 28, 2025

View reviewed changes

ShahimSharafudeen mentioned this pull request Apr 28, 2025

Implement product test for singlenode-kafka-sasl #24993

Open

steveburnett previously approved these changes Apr 28, 2025

View reviewed changes

ZacBlanco reviewed Apr 29, 2025

View reviewed changes

ShahimSharafudeen dismissed stale reviews from steveburnett and aaneja via dde1f1f April 30, 2025 07:11

ShahimSharafudeen force-pushed the kafka_sasl_config branch from 6fa2430 to dde1f1f Compare April 30, 2025 07:11

ZacBlanco requested changes Apr 30, 2025

View reviewed changes

ShahimSharafudeen force-pushed the kafka_sasl_config branch from dde1f1f to 51cf71e Compare April 30, 2025 18:39

Add support for optional kafka SASL

32c46eb

Cherry-pick of trinodb/trino#6929 Co-authored-by: Kenzyme Le <kl@kenzymele.com> Co-authored-by: Matthias Strobl <matthias.strobl@bmw.de>

ShahimSharafudeen force-pushed the kafka_sasl_config branch from 51cf71e to 32c46eb Compare April 30, 2025 18:54

ZacBlanco approved these changes Apr 30, 2025

View reviewed changes

ethanyzhang merged commit 60bbd2e into prestodb:master May 1, 2025
99 checks passed

ZacBlanco mentioned this pull request May 29, 2025

Add release notes for 0.293 #25222

Merged

21 tasks

This was referenced May 30, 2025

Add release notes for 0.293 unix280/presto#31

Closed

Add release notes for 0.293 unix280/presto#32

Closed

Add support for optional kafka SASL #24798

Add support for optional kafka SASL #24798

Uh oh!

Conversation

ShahimSharafudeen commented Mar 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Motivation and Context

Impact

Test Plan

Contributor checklist

Release Notes

Uh oh!

aaneja left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ShahimSharafudeen commented Apr 21, 2025

Uh oh!

steveburnett left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ShahimSharafudeen commented Apr 22, 2025

Uh oh!

aaneja left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

steveburnett left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ShahimSharafudeen commented Apr 25, 2025

Uh oh!

ShahimSharafudeen commented Apr 25, 2025

Uh oh!

steveburnett left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aaneja commented Apr 28, 2025

Uh oh!

ShahimSharafudeen commented Mar 26, 2025 •

edited

Loading

aaneja left a comment •

edited

Loading

ShahimSharafudeen commented Apr 30, 2025 •

edited

Loading