This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new fda09b7a70f Support Kafka config providers in realtime table configs
(#19202)
fda09b7a70f is described below
commit fda09b7a70f43be91319a37f00f846960d03c8a8
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Aug 18 11:48:15 2026 -0700
Support Kafka config providers in realtime table configs (#19202)
---
.../KafkaPartitionLevelConnectionHandler.java | 33 +++----
.../KafkaPartitionLevelConnectionHandlerTest.java | 7 ++
.../KafkaPartitionLevelConnectionHandler.java | 33 +++----
.../KafkaPartitionLevelConnectionHandlerTest.java | 7 ++
.../plugin/stream/kafka/KafkaConfigUtils.java | 104 +++++++++++++++++++++
.../pinot/plugin/stream/kafka/KafkaSSLUtils.java | 22 +++++
.../stream/kafka/KafkaConfigProviderTestUtils.java | 92 ++++++++++++++++++
.../plugin/stream/kafka/KafkaConfigUtilsTest.java | 97 +++++++++++++++++++
.../plugin/stream/kafka/KafkaSSLUtilsTest.java | 72 +++++++++++++-
.../org/apache/pinot/spi/config/ConfigUtils.java | 6 +-
.../apache/pinot/spi/config/ConfigUtilsTest.java | 29 ++++++
11 files changed, 460 insertions(+), 42 deletions(-)
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java
index eeb41b80146..18228339543 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java
@@ -38,6 +38,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.BytesDeserializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.pinot.plugin.stream.kafka.KafkaAdminClientManager;
+import org.apache.pinot.plugin.stream.kafka.KafkaConfigUtils;
import org.apache.pinot.plugin.stream.kafka.KafkaPartitionLevelStreamConfig;
import org.apache.pinot.plugin.stream.kafka.KafkaSSLUtils;
import org.apache.pinot.spi.stream.StreamConfig;
@@ -98,28 +99,14 @@ public abstract class KafkaPartitionLevelConnectionHandler {
return consumerProp;
}
- /// Filter properties to only include the specified Kafka configurations.
- /// This prevents "was supplied but isn't a known config" warnings from
Kafka clients.
- ///
- /// @param props The properties to filter
- /// @param validConfigNames The set of valid configuration names for the
target Kafka client
- /// @return A new Properties object containing only the valid configurations
- private Properties filterKafkaProperties(Properties props, Set<String>
validConfigNames) {
- Properties filteredProps = new Properties();
- for (String key : props.stringPropertyNames()) {
- if (validConfigNames.contains(key)) {
- filteredProps.put(key, props.get(key));
- }
- }
- return filteredProps;
- }
-
private Consumer<Bytes, Bytes> createConsumer(Properties consumerProp,
RetryPolicy retryPolicy) {
+ Properties filteredConsumerProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES);
AtomicReference<Consumer<Bytes, Bytes>> consumer = new AtomicReference<>();
try {
retryPolicy.attempt(() -> {
try {
- consumer.set(new KafkaConsumer<>(filterKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES)));
+ consumer.set(new KafkaConsumer<>(filteredConsumerProp));
return true;
} catch (Exception e) {
LOGGER.warn("Caught exception while creating Kafka consumer,
retrying.", e);
@@ -135,11 +122,15 @@ public abstract class
KafkaPartitionLevelConnectionHandler {
@VisibleForTesting
protected Consumer<Bytes, Bytes> createConsumer(Properties consumerProp) {
- return retry(() -> new KafkaConsumer<>(filterKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES)), 5);
+ Properties filteredConsumerProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES);
+ return retry(() -> new KafkaConsumer<>(filteredConsumerProp), 5);
}
protected AdminClient createAdminClient() {
- return retry(() -> AdminClient.create(filterKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES)), 5);
+ Properties filteredAdminClientProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES);
+ return retry(() -> AdminClient.create(filteredAdminClientProp), 5);
}
/// Gets or creates a reusable admin client instance. The admin client is
lazily initialized
@@ -166,7 +157,9 @@ public abstract class KafkaPartitionLevelConnectionHandler {
synchronized (this) {
ref = _sharedAdminClientRef;
if (ref == null) {
- ref =
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(_consumerProp);
+ Properties filteredAdminClientProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES);
+ ref =
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(filteredAdminClientProp);
_sharedAdminClientRef = ref;
}
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandlerTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandlerTest.java
index 1abe7681c76..407b534588f 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandlerTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandlerTest.java
@@ -20,6 +20,7 @@ package org.apache.pinot.plugin.stream.kafka30;
import java.util.HashMap;
import java.util.Map;
+import org.apache.pinot.plugin.stream.kafka.KafkaConfigProviderTestUtils;
import org.apache.pinot.spi.stream.StreamConfig;
import org.testng.annotations.Test;
@@ -44,6 +45,12 @@ public class KafkaPartitionLevelConnectionHandlerTest {
return new StreamConfig("testTable_REALTIME", streamConfigMap);
}
+ @Test
+ public void testConfigProviderReferencesReachKafkaClients()
+ throws Exception {
+
KafkaConfigProviderTestUtils.assertConfigProviderReferencesReachKafkaClients();
+ }
+
@Test
public void testSharedAdminClientReference() {
StreamConfig streamConfig = createTestStreamConfig();
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandler.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandler.java
index e9a015181b3..213c3b67f12 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandler.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandler.java
@@ -38,6 +38,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.BytesDeserializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.pinot.plugin.stream.kafka.KafkaAdminClientManager;
+import org.apache.pinot.plugin.stream.kafka.KafkaConfigUtils;
import org.apache.pinot.plugin.stream.kafka.KafkaPartitionLevelStreamConfig;
import org.apache.pinot.plugin.stream.kafka.KafkaSSLUtils;
import org.apache.pinot.spi.stream.StreamConfig;
@@ -98,28 +99,14 @@ public abstract class KafkaPartitionLevelConnectionHandler {
return consumerProp;
}
- /// Filter properties to only include the specified Kafka configurations.
- /// This prevents "was supplied but isn't a known config" warnings from
Kafka clients.
- ///
- /// @param props The properties to filter
- /// @param validConfigNames The set of valid configuration names for the
target Kafka client
- /// @return A new Properties object containing only the valid configurations
- private Properties filterKafkaProperties(Properties props, Set<String>
validConfigNames) {
- Properties filteredProps = new Properties();
- for (String key : props.stringPropertyNames()) {
- if (validConfigNames.contains(key)) {
- filteredProps.put(key, props.get(key));
- }
- }
- return filteredProps;
- }
-
private Consumer<Bytes, Bytes> createConsumer(Properties consumerProp,
RetryPolicy retryPolicy) {
+ Properties filteredConsumerProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES);
AtomicReference<Consumer<Bytes, Bytes>> consumer = new AtomicReference<>();
try {
retryPolicy.attempt(() -> {
try {
- consumer.set(new KafkaConsumer<>(filterKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES)));
+ consumer.set(new KafkaConsumer<>(filteredConsumerProp));
return true;
} catch (Exception e) {
LOGGER.warn("Caught exception while creating Kafka consumer,
retrying.", e);
@@ -135,11 +122,15 @@ public abstract class
KafkaPartitionLevelConnectionHandler {
@VisibleForTesting
protected Consumer<Bytes, Bytes> createConsumer(Properties consumerProp) {
- return retry(() -> new KafkaConsumer<>(filterKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES)), 5);
+ Properties filteredConsumerProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES);
+ return retry(() -> new KafkaConsumer<>(filteredConsumerProp), 5);
}
protected AdminClient createAdminClient() {
- return retry(() -> AdminClient.create(filterKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES)), 5);
+ Properties filteredAdminClientProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES);
+ return retry(() -> AdminClient.create(filteredAdminClientProp), 5);
}
/// Gets or creates a reusable admin client instance. The admin client is
lazily initialized
@@ -166,7 +157,9 @@ public abstract class KafkaPartitionLevelConnectionHandler {
synchronized (this) {
ref = _sharedAdminClientRef;
if (ref == null) {
- ref =
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(_consumerProp);
+ Properties filteredAdminClientProp =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES);
+ ref =
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(filteredAdminClientProp);
_sharedAdminClientRef = ref;
}
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandlerTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandlerTest.java
index a7a2ac9ee7a..01fc2734075 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandlerTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandlerTest.java
@@ -20,6 +20,7 @@ package org.apache.pinot.plugin.stream.kafka40;
import java.util.HashMap;
import java.util.Map;
+import org.apache.pinot.plugin.stream.kafka.KafkaConfigProviderTestUtils;
import org.apache.pinot.spi.stream.StreamConfig;
import org.testng.annotations.Test;
@@ -44,6 +45,12 @@ public class KafkaPartitionLevelConnectionHandlerTest {
return new StreamConfig("testTable_REALTIME", streamConfigMap);
}
+ @Test
+ public void testConfigProviderReferencesReachKafkaClients()
+ throws Exception {
+
KafkaConfigProviderTestUtils.assertConfigProviderReferencesReachKafkaClients();
+ }
+
@Test
public void testSharedAdminClientReference() {
StreamConfig streamConfig = createTestStreamConfig();
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtils.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtils.java
new file mode 100644
index 00000000000..e3456506ce9
--- /dev/null
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtils.java
@@ -0,0 +1,104 @@
+/**
+ * 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.pinot.plugin.stream.kafka;
+
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+import java.util.regex.Matcher;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.ConfigTransformer;
+import org.apache.kafka.common.config.provider.FileConfigProvider;
+
+
+/// Shared helpers for preparing Kafka client configuration. The helpers are
stateless and thread-safe.
+public class KafkaConfigUtils {
+ private static final String CONFIG_PROVIDERS_PREFIX =
AbstractConfig.CONFIG_PROVIDERS_CONFIG + ".";
+ private static final String CONFIG_PROVIDER_CLASS_SUFFIX = ".class";
+ private static final String CONFIG_PROVIDER_PARAM_PREFIX = ".param.";
+
+ private KafkaConfigUtils() {
+ }
+
+ /// Filters properties to the target Kafka client's known configurations and
Kafka's dynamic ConfigProvider
+ /// namespace. Kafka may still report the provider keys themselves as
unknown after using them for resolution.
+ ///
+ /// @param properties properties to filter
+ /// @param validConfigNames configuration names recognized by the target
Kafka client
+ /// @return a new Properties object containing the client and ConfigProvider
settings
+ /// @throws ConfigException if a referenced provider is undeclared, has no
class, or is a FileConfigProvider
+ /// without an allowed-path restriction
+ public static Properties filterAndValidateKafkaProperties(Properties
properties, Set<String> validConfigNames) {
+ Properties filteredProperties = new Properties();
+ for (String key : properties.stringPropertyNames()) {
+ if (validConfigNames.contains(key) ||
key.equals(AbstractConfig.CONFIG_PROVIDERS_CONFIG)
+ || key.startsWith(CONFIG_PROVIDERS_PREFIX)) {
+ filteredProperties.put(key, properties.get(key));
+ }
+ }
+ validateConfigProviderReferences(filteredProperties);
+ return filteredProperties;
+ }
+
+ private static void validateConfigProviderReferences(Properties properties) {
+ Set<String> configuredProviders = getConfiguredProviders(properties);
+ for (String key : properties.stringPropertyNames()) {
+ Matcher matcher =
ConfigTransformer.DEFAULT_PATTERN.matcher(properties.getProperty(key));
+ while (matcher.find()) {
+ String provider = matcher.group(1);
+ if (!configuredProviders.contains(provider)) {
+ throw new ConfigException("Kafka ConfigProvider alias '" + provider
+ "' referenced by '" + key
+ + "' is not listed in '" +
AbstractConfig.CONFIG_PROVIDERS_CONFIG + "'");
+ }
+ validateProviderConfiguration(properties, provider, key);
+ }
+ }
+ }
+
+ private static Set<String> getConfiguredProviders(Properties properties) {
+ Set<String> configuredProviders = new HashSet<>();
+ String providers =
properties.getProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG, "");
+ for (String provider : providers.split(",")) {
+ String trimmedProvider = provider.trim();
+ if (!trimmedProvider.isEmpty()) {
+ configuredProviders.add(trimmedProvider);
+ }
+ }
+ return configuredProviders;
+ }
+
+ private static void validateProviderConfiguration(Properties properties,
String provider, String referencingKey) {
+ String providerPrefix = CONFIG_PROVIDERS_PREFIX + provider;
+ String providerClassKey = providerPrefix + CONFIG_PROVIDER_CLASS_SUFFIX;
+ String providerClass = properties.getProperty(providerClassKey);
+ if (providerClass == null || providerClass.trim().isEmpty()) {
+ throw new ConfigException("Kafka ConfigProvider alias '" + provider + "'
referenced by '" + referencingKey
+ + "' does not define '" + providerClassKey + "'");
+ }
+ if (providerClass.equals(FileConfigProvider.class.getName())) {
+ String allowedPathsKey = providerPrefix + CONFIG_PROVIDER_PARAM_PREFIX +
FileConfigProvider.ALLOWED_PATHS_CONFIG;
+ String allowedPaths = properties.getProperty(allowedPathsKey);
+ if (allowedPaths == null || allowedPaths.trim().isEmpty()) {
+ throw new ConfigException("Kafka FileConfigProvider alias '" +
provider + "' referenced by '" + referencingKey
+ + "' must define '" + allowedPathsKey + "'");
+ }
+ }
+ }
+}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java
index 92e4a5d322f..19895e0fe55 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java
@@ -39,6 +39,7 @@ import java.util.Arrays;
import java.util.Base64;
import java.util.Properties;
import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.common.config.ConfigTransformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,6 +82,17 @@ public class KafkaSSLUtils {
String trustStoreLocation =
consumerProps.getProperty(SSL_TRUSTSTORE_LOCATION);
String trustStorePassword =
consumerProps.getProperty(SSL_TRUSTSTORE_PASSWORD);
String serverCertificate =
consumerProps.getProperty(STREAM_KAFKA_SSL_SERVER_CERTIFICATE);
+ if (StringUtils.isNotEmpty(serverCertificate)) {
+ validateAutoSslProperties(consumerProps, SSL_TRUSTSTORE_LOCATION,
SSL_TRUSTSTORE_PASSWORD,
+ STREAM_KAFKA_SSL_SERVER_CERTIFICATE,
STREAM_KAFKA_SSL_CERTIFICATE_TYPE, SSL_TRUSTSTORE_TYPE);
+
+ String clientCertificate =
consumerProps.getProperty(STREAM_KAFKA_SSL_CLIENT_CERTIFICATE);
+ if (StringUtils.isNotEmpty(clientCertificate)) {
+ validateAutoSslProperties(consumerProps, SSL_KEYSTORE_LOCATION,
SSL_KEYSTORE_PASSWORD, SSL_KEY_PASSWORD,
+ STREAM_KAFKA_SSL_CLIENT_CERTIFICATE, STREAM_KAFKA_SSL_CLIENT_KEY,
+ STREAM_KAFKA_SSL_CLIENT_KEY_ALGORITHM, SSL_KEYSTORE_TYPE);
+ }
+ }
if (StringUtils.isAnyEmpty(trustStoreLocation, trustStorePassword,
serverCertificate)) {
LOGGER.info("Skipping auto SSL server validation since it's not
configured.");
return;
@@ -108,6 +120,16 @@ public class KafkaSSLUtils {
}
}
+ private static void validateAutoSslProperties(Properties consumerProps,
String... propertyNames) {
+ for (String propertyName : propertyNames) {
+ String value = consumerProps.getProperty(propertyName);
+ if (value != null &&
ConfigTransformer.DEFAULT_PATTERN.matcher(value).find()) {
+ throw new IllegalArgumentException("Kafka ConfigProvider references
are not supported for '" + propertyName
+ + "' when Pinot auto-generates Kafka SSL stores; use a prebuilt
keystore or truststore file instead");
+ }
+ }
+ }
+
@VisibleForTesting
static void initTrustStore(Properties consumerProps) {
Path trustStorePath = getTrustStorePath(consumerProps);
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigProviderTestUtils.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigProviderTestUtils.java
new file mode 100644
index 00000000000..0a2ebcff219
--- /dev/null
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigProviderTestUtils.java
@@ -0,0 +1,92 @@
+/**
+ * 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.pinot.plugin.stream.kafka;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.provider.FileConfigProvider;
+import org.apache.kafka.common.serialization.BytesDeserializer;
+import org.apache.pinot.spi.config.ConfigUtils;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+
+
+/// Test helper that verifies ConfigProvider resolution against the Kafka
client version on the caller's classpath.
+public class KafkaConfigProviderTestUtils {
+ private KafkaConfigProviderTestUtils() {
+ }
+
+ public static void assertConfigProviderReferencesReachKafkaClients()
+ throws Exception {
+ Path providerFile = Files.createTempFile("kafka-config-provider",
".properties");
+ try {
+ Files.writeString(providerFile, "keystore.password=test-password\n");
+
+ String passwordReference = "${file:" + providerFile +
":keystore.password}";
+ Map<String, String> streamConfigs = new HashMap<>();
+ streamConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9092");
+ streamConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
BytesDeserializer.class.getName());
+ streamConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
BytesDeserializer.class.getName());
+ streamConfigs.put(AbstractConfig.CONFIG_PROVIDERS_CONFIG, "file");
+ streamConfigs.put("config.providers.file.class",
FileConfigProvider.class.getName());
+ streamConfigs.put("config.providers.file.param.allowed.paths",
providerFile.getParent().toString());
+ streamConfigs.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "$" +
passwordReference);
+ streamConfigs.put("streamType", "kafka");
+
+ IndexingConfig indexingConfig = new IndexingConfig();
+ indexingConfig.setStreamConfigs(streamConfigs);
+ IndexingConfig resolvedIndexingConfig =
+ ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(Map.of(),
indexingConfig);
+ Properties properties = new Properties();
+ properties.putAll(resolvedIndexingConfig.getStreamConfigs());
+
assertEquals(properties.getProperty(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG),
passwordReference);
+
+ Properties consumerProperties =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
ConsumerConfig.configNames());
+ assertProviderProperties(consumerProperties, providerFile);
+ assertEquals(new
ConsumerConfig(consumerProperties).getPassword(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).value(),
+ "test-password");
+
+ Properties adminProperties =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
AdminClientConfig.configNames());
+ assertProviderProperties(adminProperties, providerFile);
+ assertEquals(new
AdminClientConfig(adminProperties).getPassword(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).value(),
+ "test-password");
+ } finally {
+ Files.deleteIfExists(providerFile);
+ }
+ }
+
+ private static void assertProviderProperties(Properties properties, Path
providerFile) {
+
assertEquals(properties.getProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG),
"file");
+ assertEquals(properties.getProperty("config.providers.file.class"),
FileConfigProvider.class.getName());
+
assertEquals(properties.getProperty("config.providers.file.param.allowed.paths"),
+ providerFile.getParent().toString());
+ assertFalse(properties.containsKey("streamType"));
+ }
+}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtilsTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtilsTest.java
new file mode 100644
index 00000000000..8d884979a72
--- /dev/null
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtilsTest.java
@@ -0,0 +1,97 @@
+/**
+ * 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.pinot.plugin.stream.kafka;
+
+import java.util.Properties;
+import java.util.Set;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.provider.FileConfigProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.expectThrows;
+
+
+public class KafkaConfigUtilsTest {
+
+ @Test
+ public void testFilterKafkaPropertiesPreservesConfigProviders() {
+ Properties properties = createFileProviderProperties();
+ properties.setProperty("bootstrap.servers", "localhost:9092");
+ properties.setProperty("streamType", "kafka");
+
+ Properties filteredProperties =
+ KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
+ Set.of("bootstrap.servers", "ssl.keystore.password"));
+
+ assertEquals(filteredProperties.getProperty("bootstrap.servers"),
"localhost:9092");
+
assertEquals(filteredProperties.getProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG),
"file");
+
assertEquals(filteredProperties.getProperty("config.providers.file.class"),
FileConfigProvider.class.getName());
+
assertEquals(filteredProperties.getProperty("config.providers.file.param.allowed.paths"),
"/vault/secrets");
+ assertFalse(filteredProperties.containsKey("streamType"));
+ }
+
+ @Test
+ public void testFilterKafkaPropertiesRejectsUndeclaredProvider() {
+ Properties properties = new Properties();
+ properties.setProperty("ssl.keystore.password",
"${file:/vault/secrets/kafka.properties:password}");
+
+ ConfigException exception = expectThrows(ConfigException.class,
+ () -> KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
Set.of("ssl.keystore.password")));
+
+ assertEquals(exception.getMessage(), "Kafka ConfigProvider alias 'file'
referenced by 'ssl.keystore.password' "
+ + "is not listed in 'config.providers'");
+ }
+
+ @Test
+ public void testFilterKafkaPropertiesRejectsProviderWithoutClass() {
+ Properties properties = new Properties();
+ properties.setProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG, "file");
+ properties.setProperty("ssl.keystore.password",
"${file:/vault/secrets/kafka.properties:password}");
+
+ ConfigException exception = expectThrows(ConfigException.class,
+ () -> KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
Set.of("ssl.keystore.password")));
+
+ assertEquals(exception.getMessage(), "Kafka ConfigProvider alias 'file'
referenced by 'ssl.keystore.password' "
+ + "does not define 'config.providers.file.class'");
+ }
+
+ @Test
+ public void testFilterKafkaPropertiesRequiresAllowedPathsForFileProvider() {
+ Properties properties = createFileProviderProperties();
+ properties.remove("config.providers.file.param.allowed.paths");
+
+ ConfigException exception = expectThrows(ConfigException.class,
+ () -> KafkaConfigUtils.filterAndValidateKafkaProperties(properties,
Set.of("ssl.keystore.password")));
+
+ assertEquals(exception.getMessage(), "Kafka FileConfigProvider alias
'file' referenced by "
+ + "'ssl.keystore.password' must define
'config.providers.file.param.allowed.paths'");
+ }
+
+ private static Properties createFileProviderProperties() {
+ Properties properties = new Properties();
+ properties.setProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG, "file");
+ properties.setProperty("config.providers.file.class",
FileConfigProvider.class.getName());
+ properties.setProperty("config.providers.file.param.allowed.paths",
"/vault/secrets");
+ properties.setProperty("ssl.keystore.password",
"${file:/vault/secrets/kafka.properties:password}");
+ return properties;
+ }
+}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtilsTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtilsTest.java
index 64d2505ae0d..a9d7bf83979 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtilsTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtilsTest.java
@@ -34,6 +34,7 @@ import java.security.SecureRandom;
import java.security.Security;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
+import java.util.Arrays;
import java.util.Date;
import java.util.Enumeration;
import java.util.Properties;
@@ -141,6 +142,76 @@ public class KafkaSSLUtilsTest {
validateTrustStoreCertificateCount(1);
}
+ @Test
+ public void testInitSSLRejectsConfigProviderTrustStorePassword() {
+ Properties consumerProps = new Properties();
+ consumerProps.setProperty("ssl.truststore.location", _trustStorePath);
+ consumerProps.setProperty("ssl.truststore.password",
"${file:/vault/secrets/kafka.properties:password}");
+ consumerProps.setProperty("stream.kafka.ssl.server.certificate",
"certificate");
+
+ IllegalArgumentException exception =
+ Assert.expectThrows(IllegalArgumentException.class, () ->
KafkaSSLUtils.initSSL(consumerProps));
+
+ Assert.assertEquals(exception.getMessage(), "Kafka ConfigProvider
references are not supported for "
+ + "'ssl.truststore.password' when Pinot auto-generates Kafka SSL
stores; use a prebuilt keystore or "
+ + "truststore file instead");
+ }
+
+ @Test
+ public void testInitSSLRejectsConfigProviderKeyStorePassword()
+ throws CertificateException, NoSuchAlgorithmException,
OperatorCreationException, NoSuchProviderException {
+ Properties consumerProps = new Properties();
+ setTrustStoreProps(consumerProps);
+ setKeyStoreProps(consumerProps);
+ consumerProps.setProperty("ssl.keystore.password",
"${file:/vault/secrets/kafka.properties:password}");
+
+ IllegalArgumentException exception =
+ Assert.expectThrows(IllegalArgumentException.class, () ->
KafkaSSLUtils.initSSL(consumerProps));
+
+ Assert.assertEquals(exception.getMessage(), "Kafka ConfigProvider
references are not supported for "
+ + "'ssl.keystore.password' when Pinot auto-generates Kafka SSL stores;
use a prebuilt keystore or truststore "
+ + "file instead");
+ }
+
+ @Test
+ public void testInitSSLRejectsConfigProviderKeyPassword()
+ throws CertificateException, NoSuchAlgorithmException,
OperatorCreationException, NoSuchProviderException {
+ Properties consumerProps = new Properties();
+ setTrustStoreProps(consumerProps);
+ setKeyStoreProps(consumerProps);
+ consumerProps.setProperty("ssl.key.password",
"${file:/vault/secrets/kafka.properties:password}");
+
+ IllegalArgumentException exception =
+ Assert.expectThrows(IllegalArgumentException.class, () ->
KafkaSSLUtils.initSSL(consumerProps));
+
+ Assert.assertEquals(exception.getMessage(), "Kafka ConfigProvider
references are not supported for "
+ + "'ssl.key.password' when Pinot auto-generates Kafka SSL stores; use
a prebuilt keystore or truststore file "
+ + "instead");
+ }
+
+ @Test
+ public void testInitSSLRejectsConfigProviderBeforeRenewingExistingStore()
+ throws CertificateException, NoSuchAlgorithmException,
OperatorCreationException, NoSuchProviderException,
+ IOException {
+ Properties consumerProps = new Properties();
+ setTrustStoreProps(consumerProps);
+ KafkaSSLUtils.initSSL(consumerProps);
+ byte[] originalTrustStore = Files.readAllBytes(Paths.get(_trustStorePath));
+
+ setTrustStoreProps(consumerProps);
+ setKeyStoreProps(consumerProps);
+ consumerProps.setProperty("stream.kafka.ssl.client.key.algorithm",
+ "${file:/vault/secrets/kafka.properties:algorithm}");
+
+ IllegalArgumentException exception =
+ Assert.expectThrows(IllegalArgumentException.class, () ->
KafkaSSLUtils.initSSL(consumerProps));
+
+ Assert.assertEquals(exception.getMessage(), "Kafka ConfigProvider
references are not supported for "
+ + "'stream.kafka.ssl.client.key.algorithm' when Pinot auto-generates
Kafka SSL stores; use a prebuilt "
+ + "keystore or truststore file instead");
+
Assert.assertTrue(Arrays.equals(Files.readAllBytes(Paths.get(_trustStorePath)),
originalTrustStore));
+ }
+
@Test (expectedExceptions = java.io.FileNotFoundException.class)
public void testInitSSLKeyStoreOnly()
throws CertificateException, NoSuchAlgorithmException,
OperatorCreationException, NoSuchProviderException,
@@ -163,7 +234,6 @@ public class KafkaSSLUtilsTest {
setTrustStoreProps(consumerProps);
setKeyStoreProps(consumerProps);
KafkaSSLUtils.initSSL(consumerProps);
-
// renew the truststore and keystore
setTrustStoreProps(consumerProps);
setKeyStoreProps(consumerProps);
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java
index db24f09e41d..cfa4523477e 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java
@@ -48,7 +48,8 @@ public class ConfigUtils {
return applyConfigWithEnvVariablesAndSystemProperties(combinedMap, config);
}
- /// Apply a map of config to any given BaseJsonConfig with templates.
+ /// Apply a map of config to any given BaseJsonConfig with templates. Prefix
a template with an additional '$'
+ /// (for example, `$${name:value}`) to preserve it as a literal
`${name:value}` for downstream processing.
///
/// @return Config with the configs applied.
public static <T extends BaseJsonConfig> T
applyConfigWithEnvVariablesAndSystemProperties(
@@ -92,6 +93,9 @@ public class ConfigUtils {
break;
case STRING:
final String field = jsonNode.asText();
+ if (field.startsWith("$${") && field.endsWith("}")) {
+ return JsonNodeFactory.instance.textNode(field.substring(1));
+ }
if (field.startsWith("${") && field.endsWith("}")) {
String[] envVarSplits = field.substring(2, field.length() -
1).split(":", 2);
String envVarKey = envVarSplits[0];
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
index 586def3ca9f..7ab927c0013 100644
--- a/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
@@ -56,6 +56,35 @@ public class ConfigUtilsTest {
System.clearProperty("AWS_SECRET_KEY");
}
+ @Test
+ public void testEscapedConfigReferenceIsPreserved() {
+ IndexingConfig indexingConfig = new IndexingConfig();
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("ssl.keystore.password",
"$${file:/vault/secrets/kafka.properties:keystore.password}");
+ streamConfigMap.put("ssl.truststore.password",
"${PINOT_KAFKA_TRUSTSTORE_PASSWORD:fallback}");
+ indexingConfig.setStreamConfigs(streamConfigMap);
+
+ IndexingConfig resolvedConfig =
+ ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(Map.of(),
indexingConfig);
+
+
assertEquals(resolvedConfig.getStreamConfigs().get("ssl.keystore.password"),
+ "${file:/vault/secrets/kafka.properties:keystore.password}");
+
assertEquals(resolvedConfig.getStreamConfigs().get("ssl.truststore.password"),
"fallback");
+ }
+
+ @Test
+ public void testEscapedAndUnescapedConfigReferences() {
+ IndexingConfig indexingConfig = new IndexingConfig();
+ indexingConfig.setStreamConfigs(
+ Map.of("escaped", "$${file:fallback}", "unescaped",
"${file:fallback}"));
+
+ IndexingConfig resolvedConfig =
+ ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(Map.of(),
indexingConfig);
+
+ assertEquals(resolvedConfig.getStreamConfigs().get("escaped"),
"${file:fallback}");
+ assertEquals(resolvedConfig.getStreamConfigs().get("unescaped"),
"fallback");
+ }
+
private void testIndexingWithConfig(Map<String, String> configOverride) {
IndexingConfig indexingConfig = new IndexingConfig();
indexingConfig.setLoadMode("${LOAD_MODE}");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
