xiangfu0 commented on code in PR #19202:
URL: https://github.com/apache/pinot/pull/19202#discussion_r3762991347
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java:
##########
@@ -92,6 +93,9 @@ private static JsonNode
applyConfigWithEnvVariablesAndSystemProperties(Map<Strin
break;
case STRING:
final String field = jsonNode.asText();
+ if (field.startsWith("$${") && field.endsWith("}")) {
+ return JsonNodeFactory.instance.textNode(field.substring(1));
Review Comment:
This is the right shape — connector-neutral, five lines, and `pinot-spi` no
longer knows Kafka exists. It also gives every downstream system a way to pass
a literal `${...}` through, which is generally useful beyond this feature.
One small note for the docs: the escape isn't composable. Expressing a
literal `$${x}` would need `$$${x}`, which matches neither branch and passes
through unchanged. Fine in practice, just worth a sentence so nobody discovers
it the hard way.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -98,28 +99,13 @@ private Properties buildProperties(StreamConfig
streamConfig) {
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) {
AtomicReference<Consumer<Bytes, Bytes>> consumer = new AtomicReference<>();
try {
retryPolicy.attempt(() -> {
try {
- consumer.set(new KafkaConsumer<>(filterKafkaProperties(consumerProp,
CONSUMER_CONFIG_NAMES)));
+ consumer.set(new
KafkaConsumer<>(KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp,
Review Comment:
**The retry wrapper swallows the validation error this revision just added.**
`ConfigException extends KafkaException extends RuntimeException` — I
confirmed the hierarchy against 3.9.2. Validation runs *inside* the retry
lambda here, so the `catch (Exception e)` on line 110 logs it at WARN, returns
`false`, and the method ends up throwing
`RuntimeException(AttemptsExceededException)`. The operator sees "attempts
exceeded" — never "Kafka ConfigProvider alias 'file' ... is not listed in
'config.providers'".
`retry(Supplier, 5)` on lines 124 and 129 has the same problem in milder
form: it catches `KafkaException`, so a deterministic config error burns 5
attempts and ~8s of sleeps before surfacing. The message does survive there.
This undercuts the whole point of the new validation. Hoisting
`filterAndValidateKafkaProperties` out of the retry — validate once in
`buildProperties` or the constructor, then retry only the client construction —
fixes both paths and gets the operator a clear message immediately.
##########
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()) {
Review Comment:
Worth a line in the description: this is stricter than Kafka itself. Kafka
leaves an unresolvable reference untouched; here any `${a:b}`-shaped
**substring** with an undeclared alias throws.
A whole-value reference can't reach this point — `ConfigUtils` would already
have consumed it — so this only affects embedded ones, e.g. a literal `${x:y}`
inside `sasl.jaas.config`. Those previously passed through to Kafka untouched
and now fail the table, and the new `$${...}` escape can't express them because
it only matches whole values.
Narrow, and I think failing loudly is the right default here. But it is a
behavior change with no escape hatch, so it should be stated rather than
discovered.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaSSLUtils.java:
##########
@@ -232,6 +266,103 @@ private static Path getKeyStorePath(Properties
consumerProps) {
return Paths.get(keyStoreLocation);
}
+ private static void writeKeyStoreAtomically(Path storePath, KeyStore
keyStore, String password)
Review Comment:
**This and `validatePrivateKeyMatchesPublicKey` below are unrelated to
config providers — I'd split them into their own PR.**
Rejecting provider references under auto-SSL (line 134) is in scope and is
what I asked for. These two are not:
- `writeKeyStoreAtomically` replaces delete → create → write with temp file
+ `ATOMIC_MOVE`. Genuine improvement, and it closes a real torn-file window —
but it's a separate bug fix, and it drops the existing
`FileAlreadyExistsException` → warn-and-skip path, which is a behavior change
for current auto-SSL users.
- `validatePrivateKeyMatchesPublicKey` is ~100 new lines spanning RSA / EC /
EdDSA / DSA / RSASSA-PSS with PSS parameter derivation and a signature
round-trip, now on the `initKeyStore` path of every existing auto-SSL user.
`validateAutoSslMaterial` likewise parses certs and keys eagerly and wraps any
exception in `IllegalArgumentException`.
Both look like improvements to me. But neither is exercised by the
config-provider feature, both can newly hard-fail a setup that works today
(unusual key type, provider-specific algorithm naming, non-default keystore
type), and security-sensitive shared code deserves review on its own merits
rather than as a rider on a feature PR. Splitting also keeps this PR's blast
radius to something a reviewer can reason about.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
