Added dynamic SSL initialization support for the Kafka client

[rajagopr]

Description

Added support to dynamically initialize the Kafka client SSL configs based on the configuration passed as part of the streamConfigs. Support is added for one-way SSL (client validates the servers certificate) and two-way SSL( client validates the servers certificate and the server validates the clients certificate).

Configuration

One-way SSL is enabled when the table config contains the following configuration.

ssl.truststore.location=<path to truststore file>
ssl.truststore.password=<truststore password>
stream.kafka.ssl.server.certificate=<base64 encoded server certificate file>

The server certificate should be passed via the streamConfigs by setting the following key in base64 encoded format.

stream.kafka.ssl.server.certificate=<base64 encoded certificate>

Two-way SSL is enabled when the table config contains the following configuration in addition to the config required to enable one-way SSL.

ssl.keystore.location=<path to keystore file>
ssl.keystore.password=<keystore password>
ssl.key.password=<key password>
stream.kafka.ssl.client.certificate=<base64 encoded client certificate file>

The client certificate and key should be passed via the streamConfigs by setting the following key in base64 encoded format.

stream.kafka.ssl.client.certificate=<base64 encoded certificate>
stream.kafka.ssl.client.key=<base64 encoded key>

Certificate Renewal

The certificates can be renewed by updating the certificate entries in the table configs. The Pinot API forceCommit can be called to flush out the current consuming segments and start new Kafka consumers. This process would update the certificates if the certificates have changed.

Testing

Tested e2e by connecting to a test cluster from Aiven which supports two-way SSL based Kafka clusters.

[codecov-commenter]

Codecov Report

Attention: 28 lines in your changes are missing coverage. Please review.

Comparison is base (298e8d9) 61.51% compared to head (bf11077) 61.59%.
Report is 14 commits behind head on master.

Files	Patch %	Lines
...che/pinot/plugin/stream/kafka20/KafkaSSLUtils.java	81.93%	22 Missing and 6 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #12249      +/-   ##
============================================
+ Coverage     61.51%   61.59%   +0.07%     
+ Complexity     1153     1152       -1     
============================================
  Files          2416     2417       +1     
  Lines        131179   131367     +188     
  Branches      20246    20262      +16     
============================================
+ Hits          80691    80911     +220     
+ Misses        44595    44550      -45     
- Partials       5893     5906      +13     

Flag	Coverage Δ
custom-integration1	<0.01% <0.00%> (-0.01%)	⬇️
integration	<0.01% <0.00%> (-0.01%)	⬇️
integration1	<0.01% <0.00%> (-0.01%)	⬇️
integration2	0.00% <0.00%> (ø)
java-11	61.55% <82.05%> (+0.06%)	⬆️
java-21	61.45% <82.05%> (+0.06%)	⬆️
skip-bytebuffers-false	61.58% <82.05%> (+0.08%)	⬆️
skip-bytebuffers-true	61.43% <82.05%> (+0.06%)	⬆️
temurin	61.59% <82.05%> (+0.07%)	⬆️
unittests	61.58% <82.05%> (+0.07%)	⬆️
unittests1	46.59% <ø> (-0.01%)	⬇️
unittests2	27.82% <82.05%> (+0.12%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[snleee]

LGTM