This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 7d9a8e5bd99 NIFI-15836 Refactored Kafka3ConnectionService to use
SSLContextProvider (#11239)
7d9a8e5bd99 is described below
commit 7d9a8e5bd9909a064dc9c52f6bf85bc3409a62a9
Author: David Handermann <[email protected]>
AuthorDate: Wed May 13 04:48:35 2026 -0500
NIFI-15836 Refactored Kafka3ConnectionService to use SSLContextProvider
(#11239)
- Added StandardSslEngineFactory implementation of Apache Kafka
SslEngineFactory
- Replaced Keystore and Truststore property passing with SSLContext from
SSLContextProvider
---
.../nifi-kafka-service-shared/pom.xml | 6 +
.../kafka/service/Kafka3ConnectionService.java | 36 ++----
.../service/security/StandardSslEngineFactory.java | 136 +++++++++++++++++++++
.../service/Kafka3ConnectionServiceSSLIT.java | 72 ++++++++---
.../security/StandardSslEngineFactoryTest.java | 68 +++++++++++
.../shared/component/KafkaClientComponent.java | 4 +-
.../kafka/shared/property/KafkaClientProperty.java | 10 +-
.../provider/StandardKafkaPropertyProvider.java | 33 -----
8 files changed, 274 insertions(+), 91 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/pom.xml
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/pom.xml
index 272bc89b43d..3921a083fef 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/pom.xml
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/pom.xml
@@ -91,5 +91,11 @@
<version>2.10.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>2.10.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
index c7cc4782ff0..225a21ca068 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
@@ -23,6 +23,7 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.nifi.annotation.behavior.DynamicProperty;
@@ -51,6 +52,7 @@ import
org.apache.nifi.kafka.service.consumer.Kafka3ConsumerService;
import org.apache.nifi.kafka.service.consumer.Subscription;
import org.apache.nifi.kafka.service.producer.Kafka3ProducerService;
import org.apache.nifi.kafka.service.security.OAuthBearerLoginCallbackHandler;
+import org.apache.nifi.kafka.service.security.StandardSslEngineFactory;
import org.apache.nifi.kafka.shared.component.KafkaClientComponent;
import org.apache.nifi.kafka.shared.property.IsolationLevel;
import org.apache.nifi.kafka.shared.property.SaslMechanism;
@@ -61,7 +63,7 @@ import
org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.ssl.SSLContextProvider;
import java.time.Duration;
import java.util.ArrayList;
@@ -76,14 +78,8 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import static
org.apache.nifi.kafka.service.security.OAuthBearerLoginCallbackHandler.PROPERTY_KEY_NIFI_OAUTH_2_ACCESS_TOKEN_PROVIDER;
+import static
org.apache.nifi.kafka.service.security.StandardSslEngineFactory.SSL_CONTEXT_PROVIDER_PROPERTY;
import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SASL_LOGIN_CALLBACK_HANDLER_CLASS;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_LOCATION;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_TYPE;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEY_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_LOCATION;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_TYPE;
import static
org.apache.nifi.kafka.shared.util.SaslExtensionUtil.SASL_EXTENSION_PROPERTY_PREFIX;
import static
org.apache.nifi.kafka.shared.util.SaslExtensionUtil.isSaslExtensionProperty;
import static
org.apache.nifi.kafka.shared.util.SaslExtensionUtil.removeSaslExtensionPropertyPrefix;
@@ -396,25 +392,11 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
}
private void setSslProperties(final Properties properties, final
PropertyContext context) {
- final PropertyValue sslContextServiceProperty =
context.getProperty(SSL_CONTEXT_SERVICE);
- if (sslContextServiceProperty.isSet()) {
- final SSLContextService sslContextService =
sslContextServiceProperty.asControllerService(SSLContextService.class);
- if (sslContextService.isKeyStoreConfigured()) {
- properties.put(SSL_KEYSTORE_LOCATION.getProperty(),
sslContextService.getKeyStoreFile());
- properties.put(SSL_KEYSTORE_TYPE.getProperty(),
sslContextService.getKeyStoreType());
-
- final String keyStorePassword =
sslContextService.getKeyStorePassword();
- properties.put(SSL_KEYSTORE_PASSWORD.getProperty(),
keyStorePassword);
-
- final String keyPassword = sslContextService.getKeyPassword();
- final String configuredKeyPassword = keyPassword == null ?
keyStorePassword : keyPassword;
- properties.put(SSL_KEY_PASSWORD.getProperty(),
configuredKeyPassword);
- }
- if (sslContextService.isTrustStoreConfigured()) {
- properties.put(SSL_TRUSTSTORE_LOCATION.getProperty(),
sslContextService.getTrustStoreFile());
- properties.put(SSL_TRUSTSTORE_TYPE.getProperty(),
sslContextService.getTrustStoreType());
- properties.put(SSL_TRUSTSTORE_PASSWORD.getProperty(),
sslContextService.getTrustStorePassword());
- }
+ final PropertyValue sslContextProperty =
context.getProperty(SSL_CONTEXT_SERVICE);
+ if (sslContextProperty.isSet()) {
+ final SSLContextProvider sslContextProvider =
sslContextProperty.asControllerService(SSLContextProvider.class);
+ properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG,
StandardSslEngineFactory.class.getName());
+ properties.put(SSL_CONTEXT_PROVIDER_PROPERTY, sslContextProvider);
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactory.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactory.java
new file mode 100644
index 00000000000..0f012fe9b53
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactory.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.kafka.service.security;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.security.auth.SslEngineFactory;
+import org.apache.nifi.ssl.SSLContextProvider;
+
+import java.security.KeyStore;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+
+/**
+ * Standard implementation of Apache Kafka SslEngineFactory using Apache NiFi
SSLContextProvider Controller Services
+ */
+public class StandardSslEngineFactory implements SslEngineFactory {
+
+ public static final String SSL_CONTEXT_PROVIDER_PROPERTY =
SSLContextProvider.class.getName();
+
+ private volatile SSLContext sslContext;
+
+ /**
+ * Configure the Factory using provided properties
+ *
+ * @param configuration Map of properties with SSLContextProvider property
required
+ */
+ @Override
+ public void configure(final Map<String, ?> configuration) {
+ final Object provider =
configuration.get(SSL_CONTEXT_PROVIDER_PROPERTY);
+ if (provider == null) {
+ throw new KafkaException("Required property [%s] not
configured".formatted(SSL_CONTEXT_PROVIDER_PROPERTY));
+ }
+
+ if (provider instanceof SSLContextProvider sslContextProvider) {
+ sslContext = sslContextProvider.createContext();
+ } else {
+ throw new KafkaException("Required property [%s] not valid
[%s]".formatted(SSL_CONTEXT_PROVIDER_PROPERTY, provider.getClass().getName()));
+ }
+ }
+
+ /**
+ * Create Client SSLEngine for Peer Address using provided Endpoint
Identification Algorithm
+ *
+ * @param peerHost Peer host address
+ * @param peerPort Peer port number
+ * @param endpointIdentificationAlgorithm Endpoint identification
algorithm for client mode verification
+ * @return SSLEngine created according to provided properties
+ */
+ @Override
+ public SSLEngine createClientSslEngine(final String peerHost, final int
peerPort, final String endpointIdentificationAlgorithm) {
+ final SSLEngine sslEngine = sslContext.createSSLEngine(peerHost,
peerPort);
+ sslEngine.setUseClientMode(true);
+
+ final SSLParameters sslParameters = sslEngine.getSSLParameters();
+
sslParameters.setEndpointIdentificationAlgorithm(endpointIdentificationAlgorithm);
+ sslEngine.setSSLParameters(sslParameters);
+
+ return sslEngine;
+ }
+
+ /**
+ * Create Server SSLEngine is not supported
+ *
+ * @param peerHost Peer host address
+ * @param peerPort Peer port number
+ * @throws UnsupportedOperationException indicating Server SSLEngine
creation not supported
+ */
+ @Override
+ public SSLEngine createServerSslEngine(final String peerHost, final int
peerPort) {
+ throw new UnsupportedOperationException("Server SSLEngine creation not
supported");
+ }
+
+ /**
+ * Should be rebuilt based on proposed configuration
+ *
+ * @param configuration Proposed configuration properties
+ * @return Rebuild is never required or supported
+ */
+ @Override
+ public boolean shouldBeRebuilt(final Map<String, Object> configuration) {
+ return false;
+ }
+
+ /**
+ * Get reconfiguration properties returns an empty set indicating nothing
can be reconfigured
+ *
+ * @return Empty set of property names
+ */
+ @Override
+ public Set<String> reconfigurableConfigs() {
+ return Collections.emptySet();
+ }
+
+ /**
+ * Get Keystore is not supported with the SSLContextProvider
+ *
+ * @return Null value indicating unsupported status
+ */
+ @Override
+ public KeyStore keystore() {
+ return null;
+ }
+
+ /**
+ * Get Truststore is not supported with the SSLContextProvider
+ *
+ * @return Null value indicating unsupported status
+ */
+ @Override
+ public KeyStore truststore() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
index 017de9f8e14..fd1e64033c9 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceSSLIT.java
@@ -17,20 +17,32 @@
package org.apache.nifi.kafka.service;
import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.nifi.kafka.shared.property.KafkaClientProperty;
+import org.apache.kafka.common.config.SslConfigs;
import org.apache.nifi.kafka.shared.property.SecurityProtocol;
import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.ssl.SSLContextProvider;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.util.TestRunner;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.SecureRandom;
import java.util.LinkedHashMap;
import java.util.Map;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class Kafka3ConnectionServiceSSLIT extends
Kafka3ConnectionServiceBaseIT {
+ private static final String TLS_PROTOCOL = "TLS";
+
@Override
protected Map<String, String> getKafkaContainerConfigProperties() {
final Map<String, String> properties = new
LinkedHashMap<>(super.getKafkaContainerConfigProperties());
@@ -60,32 +72,52 @@ public class Kafka3ConnectionServiceSSLIT extends
Kafka3ConnectionServiceBaseIT
protected Map<String, String> getAdminClientConfigProperties() {
final Map<String, String> properties = new
LinkedHashMap<>(super.getAdminClientConfigProperties());
properties.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG,
SecurityProtocol.SSL.name());
- properties.put(KafkaClientProperty.SSL_KEY_PASSWORD.getProperty(),
KEY_PASSWORD);
-
properties.put(KafkaClientProperty.SSL_KEYSTORE_LOCATION.getProperty(),
keyStorePath.toString());
- properties.put(KafkaClientProperty.SSL_KEYSTORE_TYPE.getProperty(),
keyStoreType);
-
properties.put(KafkaClientProperty.SSL_KEYSTORE_PASSWORD.getProperty(),
KEY_STORE_PASSWORD);
-
properties.put(KafkaClientProperty.SSL_TRUSTSTORE_LOCATION.getProperty(),
trustStorePath.toString());
- properties.put(KafkaClientProperty.SSL_TRUSTSTORE_TYPE.getProperty(),
keyStoreType);
-
properties.put(KafkaClientProperty.SSL_TRUSTSTORE_PASSWORD.getProperty(),
KEY_STORE_PASSWORD);
+ properties.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, KEY_PASSWORD);
+ properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
keyStorePath.toString());
+ properties.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, keyStoreType);
+ properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,
KEY_STORE_PASSWORD);
+ properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
trustStorePath.toString());
+ properties.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, keyStoreType);
+ properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
KEY_STORE_PASSWORD);
return properties;
}
private String addSSLContextService(final TestRunner runner) throws
InitializationException {
final String identifier = SSLContextService.class.getSimpleName();
- final SSLContextService service = mock(SSLContextService.class);
- when(service.getIdentifier()).thenReturn(identifier);
- runner.addControllerService(identifier, service);
+ final SSLContextProvider provider = mock(SSLContextProvider.class);
+ when(provider.getIdentifier()).thenReturn(identifier);
+
+ runner.addControllerService(identifier, provider);
+ runner.enableControllerService(provider);
- when(service.isKeyStoreConfigured()).thenReturn(true);
- when(service.getKeyStoreFile()).thenReturn(keyStorePath.toString());
- when(service.getKeyStoreType()).thenReturn(keyStoreType);
- when(service.getKeyStorePassword()).thenReturn(KEY_STORE_PASSWORD);
- when(service.isTrustStoreConfigured()).thenReturn(true);
-
when(service.getTrustStoreFile()).thenReturn(trustStorePath.toString());
- when(service.getTrustStoreType()).thenReturn(keyStoreType);
- when(service.getTrustStorePassword()).thenReturn(KEY_STORE_PASSWORD);
+ try {
+ final SSLContext sslContext = buildSslContext();
+ when(provider.createContext()).thenReturn(sslContext);
+ } catch (final GeneralSecurityException | IOException e) {
+ throw new InitializationException(e);
+ }
- runner.enableControllerService(service);
return identifier;
}
+
+ private SSLContext buildSslContext() throws GeneralSecurityException,
IOException {
+ final KeyStore keyStore = KeyStore.getInstance(keyStoreType);
+ try (InputStream inputStream = Files.newInputStream(keyStorePath)) {
+ keyStore.load(inputStream, KEY_STORE_PASSWORD.toCharArray());
+ }
+ final KeyManagerFactory keyManagerFactory =
KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyManagerFactory.init(keyStore, KEY_PASSWORD.toCharArray());
+
+ final KeyStore trustStore = KeyStore.getInstance(keyStoreType);
+ try (InputStream inputStream = Files.newInputStream(trustStorePath)) {
+ trustStore.load(inputStream, KEY_STORE_PASSWORD.toCharArray());
+ }
+ final TrustManagerFactory trustManagerFactory =
TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ trustManagerFactory.init(trustStore);
+
+ final SSLContext sslContext = SSLContext.getInstance(TLS_PROTOCOL);
+ final SecureRandom secureRandom = new SecureRandom();
+ sslContext.init(keyManagerFactory.getKeyManagers(),
trustManagerFactory.getTrustManagers(), secureRandom);
+ return sslContext;
+ }
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactoryTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactoryTest.java
new file mode 100644
index 00000000000..b9219a76e10
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/security/StandardSslEngineFactoryTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.kafka.service.security;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.nifi.ssl.SSLContextProvider;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Map;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class StandardSslEngineFactoryTest {
+
+ private static final String LOCALHOST_ADDRESS = "127.0.0.1";
+
+ private static final int PEER_PORT = 443;
+
+ @Mock
+ private SSLContextProvider sslContextProvider;
+
+ private final StandardSslEngineFactory factory = new
StandardSslEngineFactory();
+
+ @Test
+ void testConfigureProviderNotFound() {
+ final Map<String, Object> configuration = Map.of();
+ assertThrows(KafkaException.class, () ->
factory.configure(configuration));
+ }
+
+ @Test
+ void testCreateClientSslEngine() throws Exception {
+ final SSLContext sslContext = SSLContext.getDefault();
+ when(sslContextProvider.createContext()).thenReturn(sslContext);
+
+ final Map<String, Object> configuration = Map.of(
+ StandardSslEngineFactory.SSL_CONTEXT_PROVIDER_PROPERTY,
sslContextProvider
+ );
+ factory.configure(configuration);
+
+ final SSLEngine sslEngine =
factory.createClientSslEngine(LOCALHOST_ADDRESS, PEER_PORT, null);
+
+ assertNotNull(sslEngine);
+ assertTrue(sslEngine.getUseClientMode());
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/component/KafkaClientComponent.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/component/KafkaClientComponent.java
index 65b3c6765f7..ffda9cebd56 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/component/KafkaClientComponent.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/component/KafkaClientComponent.java
@@ -24,7 +24,7 @@ import org.apache.nifi.kafka.shared.property.SecurityProtocol;
import org.apache.nifi.kerberos.SelfContainedKerberosUserService;
import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.ssl.SSLContextProvider;
/**
* Kafka Client Component interface with common Property Descriptors
@@ -175,7 +175,7 @@ public interface KafkaClientComponent {
.name("SSL Context Service")
.description("Service supporting SSL communication with Kafka
brokers")
.required(false)
- .identifiesControllerService(SSLContextService.class)
+ .identifiesControllerService(SSLContextProvider.class)
.dependsOn(
SECURITY_PROTOCOL,
SecurityProtocol.SSL.name(),
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KafkaClientProperty.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KafkaClientProperty.java
index 4b41d3cc49c..f38a125cc10 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KafkaClientProperty.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KafkaClientProperty.java
@@ -23,15 +23,7 @@ public enum KafkaClientProperty {
SASL_JAAS_CONFIG("sasl.jaas.config"),
SASL_LOGIN_CLASS("sasl.login.class"),
SASL_CLIENT_CALLBACK_HANDLER_CLASS("sasl.client.callback.handler.class"),
- SASL_LOGIN_CALLBACK_HANDLER_CLASS("sasl.login.callback.handler.class"),
-
- SSL_KEYSTORE_LOCATION("ssl.keystore.location"),
- SSL_KEYSTORE_PASSWORD("ssl.keystore.password"),
- SSL_KEYSTORE_TYPE("ssl.keystore.type"),
- SSL_KEY_PASSWORD("ssl.key.password"),
- SSL_TRUSTSTORE_LOCATION("ssl.truststore.location"),
- SSL_TRUSTSTORE_PASSWORD("ssl.truststore.password"),
- SSL_TRUSTSTORE_TYPE("ssl.truststore.type");
+ SASL_LOGIN_CALLBACK_HANDLER_CLASS("sasl.login.callback.handler.class");
private final String property;
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/provider/StandardKafkaPropertyProvider.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/provider/StandardKafkaPropertyProvider.java
index f146e26f2e6..d0259952ae6 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/provider/StandardKafkaPropertyProvider.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/provider/StandardKafkaPropertyProvider.java
@@ -26,7 +26,6 @@ import org.apache.nifi.kafka.shared.login.LoginConfigProvider;
import org.apache.nifi.kafka.shared.property.SaslMechanism;
import org.apache.nifi.kafka.shared.property.SecurityProtocol;
import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.util.FormatUtils;
import java.util.LinkedHashMap;
@@ -39,17 +38,9 @@ import java.util.stream.Collectors;
import static
org.apache.nifi.kafka.shared.component.KafkaClientComponent.AWS_WEB_IDENTITY_TOKEN_PROVIDER;
import static
org.apache.nifi.kafka.shared.component.KafkaClientComponent.SASL_MECHANISM;
import static
org.apache.nifi.kafka.shared.component.KafkaClientComponent.SECURITY_PROTOCOL;
-import static
org.apache.nifi.kafka.shared.component.KafkaClientComponent.SSL_CONTEXT_SERVICE;
import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SASL_CLIENT_CALLBACK_HANDLER_CLASS;
import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SASL_JAAS_CONFIG;
import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SASL_LOGIN_CLASS;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_LOCATION;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEYSTORE_TYPE;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_KEY_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_LOCATION;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_PASSWORD;
-import static
org.apache.nifi.kafka.shared.property.KafkaClientProperty.SSL_TRUSTSTORE_TYPE;
/**
* Standard implementation of Kafka Property Provider based on shared Kafka
Property Descriptors
@@ -75,7 +66,6 @@ public class StandardKafkaPropertyProvider implements
KafkaPropertyProvider {
final Map<String, Object> properties = new LinkedHashMap<>();
setClientProperties(properties, context);
setSecurityProperties(properties, context);
- setSslProperties(properties, context);
return properties;
}
@@ -102,29 +92,6 @@ public class StandardKafkaPropertyProvider implements
KafkaPropertyProvider {
}
}
- private void setSslProperties(final Map<String, Object> properties, final
PropertyContext context) {
- final PropertyValue sslContextServiceProperty =
context.getProperty(SSL_CONTEXT_SERVICE);
- if (sslContextServiceProperty.isSet()) {
- final SSLContextService sslContextService =
sslContextServiceProperty.asControllerService(SSLContextService.class);
- if (sslContextService.isKeyStoreConfigured()) {
- properties.put(SSL_KEYSTORE_LOCATION.getProperty(),
sslContextService.getKeyStoreFile());
- properties.put(SSL_KEYSTORE_TYPE.getProperty(),
sslContextService.getKeyStoreType());
-
- final String keyStorePassword =
sslContextService.getKeyStorePassword();
- properties.put(SSL_KEYSTORE_PASSWORD.getProperty(),
keyStorePassword);
-
- final String keyPassword = sslContextService.getKeyPassword();
- final String configuredKeyPassword = keyPassword == null ?
keyStorePassword : keyPassword;
- properties.put(SSL_KEY_PASSWORD.getProperty(),
configuredKeyPassword);
- }
- if (sslContextService.isTrustStoreConfigured()) {
- properties.put(SSL_TRUSTSTORE_LOCATION.getProperty(),
sslContextService.getTrustStoreFile());
- properties.put(SSL_TRUSTSTORE_TYPE.getProperty(),
sslContextService.getTrustStoreType());
- properties.put(SSL_TRUSTSTORE_PASSWORD.getProperty(),
sslContextService.getTrustStorePassword());
- }
- }
- }
-
private void setClientProperties(final Map<String, Object> properties,
final PropertyContext context) {
final Set<PropertyDescriptor> propertyDescriptors =
getPropertyDescriptors(context).stream()
.filter(propertyDescriptor ->
clientPropertyNames.contains(propertyDescriptor.getName()))
