Support Kafka config providers in realtime table configs - #19202
Support Kafka config providers in realtime table configs#19202goutamadwant wants to merge 4 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19202 +/- ##
============================================
+ Coverage 66.63% 66.95% +0.32%
Complexity 1423 1423
============================================
Files 3443 3453 +10
Lines 218663 218582 -81
Branches 34801 34752 -49
============================================
+ Hits 145705 146355 +650
+ Misses 61230 60547 -683
+ Partials 11728 11680 -48
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Jackie-Jiang
added
kafka
xiangfu0
left a comment
xiangfu0
left a comment
There was a problem hiding this comment.
Thanks for picking this up — the diagnosis is right and the fix is in the right places.
I verified the core mechanism locally against kafka-clients 3.9.2: filtered properties carrying config.providers, config.providers.file.class and config.providers.file.param.allowed.paths do resolve ${file:/tmp/x:keystore.password} to the real value inside ConsumerConfig. AbstractConfig.CONFIG_PROVIDERS_CONFIG is public in both 3.9.2 and 4.2.0, so the constant reference is safe on both connectors.
Three points I'd like resolved before merge (inline), then two things that fall outside the diff:
KafkaSSLUtils.initSSL is incompatible with provider references. KafkaSSLUtils.java:98 reads ssl.keystore.password raw from _consumerProp, before any Kafka client is constructed. If the auto-keystore-generation feature is combined with ${file:...} passwords, the keystore gets written with the literal reference as its password while Kafka reads it back with the resolved one. It's opt-in (stream.kafka.ssl.server.certificate), so a collision is unlikely — but it should be documented or guarded rather than left to be discovered.
Rotation doesn't reach the shared AdminClient. KafkaAdminClientManager.createCacheKey (KafkaAdminClientManager.java:84-99) keys on bootstrap servers plus SSL locations, not passwords. A rotated secret won't take effect for the admin client until the refcount drops to zero. The "recreate the consumer and it rereads the file" story in the description is consumer-only; worth stating that limitation explicitly.
One correction worth folding in: the config example in #19184 uses stream.kafka.consumer.prop.ssl.keystore.password. You already flagged in the description that this prefix isn't a generic Kafka namespace, and that's correct — KafkaStreamConfigProperties.KAFKA_CONSUMER_PROP_PREFIX is declared at line 54 and referenced nowhere in the repo, and buildProperties does a raw putAll(streamConfig.getStreamConfigsMap()) with no stripping. Since the issue's acceptance criteria are written against the prefixed form, it's worth correcting there so the reporter's config actually works.
Acceptance criterion 4 (docs) is also still open. Given the allowed.paths point below, I'd treat the pinot-contrib/pinot-docs PR as part of shipping this rather than a follow-up.
| } | ||
|
|
||
| private static final Map<String, String> ENVIRONMENT_VARIABLES = System.getenv(); | ||
| private static final String CONFIG_PROVIDERS = "config.providers"; |
There was a problem hiding this comment.
Kafka-specific knowledge in pinot-spi's generic resolver.
ConfigUtils is Pinot's universal ${...} substitution for every BaseJsonConfig — it runs on every table-config read (ZKMetadataProvider:588) and every create/update validation (TableConfigValidationUtils:88). Hardcoding "config.providers" here means whether ${x:y} gets substituted now depends on the presence of a sibling key in the same JSON object. That's action at a distance, in a very generic and very hot path. Pulsar and Kinesis will hit the same ${...} collision eventually, with no equivalent signal to key off.
A generic escape — $${...} resolving to a literal ${...} — would keep pinot-spi connector-agnostic and put the intent at the point of use.
To be fair to the current design: it needs no syntax change from operators, and it keys off exactly the same signal Kafka itself uses to decide what is a provider reference. That's a real advantage for the rotation workflow in #19184. So I read this as a trade-off rather than a defect — but it's a pinot-spi API-shaped decision, and I'd rather a committer make that call explicitly than have it arrive as a side effect of a Kafka fix.
There was a problem hiding this comment.
@xiangfu0 Agreed. I replaced the provider-aware traversal in ConfigUtils with a connector-neutral escape. A whole value beginning with $${...} becomes the literal ${...} for downstream processing, so ConfigUtils no longer knows about Kafka or config.providers. Normal ${name:default} behavior is unchanged.
This is an upgrade-gated feature: all controllers and servers must be upgraded before a provider-backed table config is applied. Older Pinot versions cannot consume either form correctly, so the companion docs state that requirement explicitly.
| // Kafka ConfigProvider references share Pinot's ${name:value} syntax. Keep references for providers declared | ||
| // in the containing config object so Kafka can resolve them when constructing the client. | ||
| if (field.startsWith("${") && field.endsWith("}") | ||
| && !isConfigProviderReference(field, inheritedConfigProviders)) { |
There was a problem hiding this comment.
The silent-corruption path survives when config.providers is omitted.
With the declaration missing, ${file:/vault/secrets/kafka.properties:keystore.password} still falls through to split(":", 2) → key file, default /vault/secrets/kafka.properties:keystore.password. The password silently becomes that literal path string, and it surfaces later as an SSL handshake error pointing nowhere near the real cause.
That's the original bug's worst symptom, and it's the likeliest misconfiguration once this is documented. Worth failing loudly: when a value has three or more colon-separated segments (${alias:path:key}) and alias isn't a declared provider, that's far more likely a provider reference with a missing declaration than an env var with a colon in its default. A warn or a throw here would save a lot of support time.
There was a problem hiding this comment.
Addressed in the shared Kafka helper. After Pinot unescapes $${...}, client property preparation validates every Kafka provider reference and throws ConfigException if the alias is not listed in config.providers or its class is missing. I added negative tests for both cases, so an escaped but incomplete provider setup fails during client creation instead of becoming a path-like password.
| for (String key : props.stringPropertyNames()) { | ||
| if (validConfigNames.contains(key)) { | ||
| if (validConfigNames.contains(key) || key.equals(AbstractConfig.CONFIG_PROVIDERS_CONFIG) | ||
| || key.startsWith(AbstractConfig.CONFIG_PROVIDERS_CONFIG + ".")) { |
There was a problem hiding this comment.
This opens a new arbitrary-file-read primitive reachable from table config.
FileConfigProvider defaults to unrestricted paths. After this change, anyone who can write a table config can read any file readable by the server or controller process into a Kafka config value — a service-account token into client.id, say, which then goes over the wire to a broker they control.
Arbitrary class instantiation from table config already exists today (key.deserializer, SASL login callback handlers), so that half isn't new. File-contents-to-config-value exfiltration is. #19184 carries the security label, so I think this needs to be explicit rather than implicit.
Minimum bar: the docs lead with config.providers.file.param.allowed.paths as required, not optional. The test already sets it, which is the right instinct. For multi-tenant clusters a cluster-level opt-in is worth discussing separately.
There was a problem hiding this comment.
@xiangfu0 Addressed in code and documentation. A referenced FileConfigProvider now requires a non-empty config.providers.<alias>.param.allowed.paths. The companion docs also require a narrow path, the JVM provider-class allowlist, consistent mounts on controllers and servers, and trusted table-config writers.
I kept cluster-controlled root policy separate because table-level allowed.paths is useful defense in depth but is not a hard multi-tenant boundary. Docs compare: https://github.com/pinot-contrib/pinot-docs/compare/latest...goutamadwant:pinot-docs:support-kafka-config-providers-docs?expand=1
| /// Filter properties to only include the specified Kafka configurations. | ||
| /// This prevents "was supplied but isn't a known config" warnings from Kafka clients. | ||
| /// Filter properties to include the specified Kafka configurations and the dynamic config-provider namespace. | ||
| /// This prevents "was supplied but isn't a known config" warnings without dropping config-provider settings. |
There was a problem hiding this comment.
Small accuracy point on the javadoc: I confirmed the config.providers* keys remain in AbstractConfig.originals() after resolution, so Kafka will still log them as "supplied but isn't a known config".
That's unavoidable — Kafka needs them in originals to do the resolution at all — but "without dropping config-provider settings" now slightly oversells it. Worth a sentence noting the provider keys themselves will still be reported as unknown.
There was a problem hiding this comment.
Updated. The shared helper documentation now says that the provider namespace is retained and explicitly notes that Kafka may still report the provider keys themselves as unknown after using them for resolution.
| if (ref == null) { | ||
| ref = KafkaAdminClientManager.getInstance().getOrCreateAdminClient(_consumerProp); | ||
| ref = KafkaAdminClientManager.getInstance() | ||
| .getOrCreateAdminClient(filterKafkaProperties(_consumerProp, ADMIN_CLIENT_CONFIG_NAMES)); |
There was a problem hiding this comment.
Good catch folding this in — it makes the shared path consistent with createAdminClient() above.
Worth confirming in the description that this doesn't change admin-client sharing: createCacheKey reads bootstrap.servers, security.protocol, sasl.mechanism, sasl.jaas.config, ssl.keystore.location and ssl.truststore.location, and all six are in AdminClientConfig.configNames(), so filtering leaves the cache key identical. I checked and it does hold — just call it out, since silently re-keying a shared client cache would be a nasty regression.
There was a problem hiding this comment.
@xiangfu0 confirmed and also called this out in the PR description. The shared path filters through AdminClientConfig.configNames() plus the provider namespace. All six cache-key fields remain included, so cache identity is unchanged.
I also documented the separate rotation limitation: an existing shared AdminClient keeps its current credentials until its references drop to zero and the client is recreated.
| for (String key : props.stringPropertyNames()) { | ||
| if (validConfigNames.contains(key)) { | ||
| if (validConfigNames.contains(key) || key.equals(AbstractConfig.CONFIG_PROVIDERS_CONFIG) | ||
| || key.startsWith(AbstractConfig.CONFIG_PROVIDERS_CONFIG + ".")) { |
There was a problem hiding this comment.
filterKafkaProperties is now byte-identical across pinot-kafka-3.0 and pinot-kafka-4.0, and so is the ~65-line test. Nothing in it is version-specific — it could live in pinot-kafka-base alongside KafkaAdminClientManager and KafkaSSLUtils.
Pre-existing pattern, so I won't block on it, but this PR doubles the surface that has to stay in sync.
There was a problem hiding this comment.
done. Filtering and validation now live in pinot-kafka-base as KafkaConfigUtils, and the shared resolution setup lives in KafkaConfigProviderTestUtils. The Kafka 3.x and 4.x tests now contain only the small version-specific calls.
| } | ||
|
|
||
| @Test | ||
| public void testConfigProviderReferencesAreScopedToDeclaringObject() { |
There was a problem hiding this comment.
The name promises descendant scoping, but this case actually verifies sibling isolation — two entries in streamConfigMaps, where the second never inherits from the first.
The implementation inherits the provider set downward into all descendants of the declaring object, which is a different property. Either rename to match what's asserted, or add a case that pins the descendant behavior (a provider declared at an outer level reaching a nested object).
Also missing: a negative case where a provider-shaped reference appears with no config.providers declared. That's the silent-corruption path I flagged in ConfigUtils, and it's the one most likely to bite operators.
There was a problem hiding this comment.
The provider-scoping design is gone with the Kafka-aware traversal. ConfigUtils tests now cover generic escaped and unescaped substitution directly. Kafka-side negative coverage now includes an undeclared alias, a missing provider class, and FileConfigProvider without allowed.paths.
| assertEquals(new AdminClientConfig(adminProperties).getPassword(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).value(), | ||
| "test-password"); | ||
| try (KafkaAdminClientManager.AdminClientReference adminClientReference = | ||
| KafkaAdminClientManager.getInstance().getOrCreateAdminClient(adminProperties)) { |
There was a problem hiding this comment.
This constructs a real AdminClient inside the process-wide KafkaAdminClientManager singleton, which spawns connection threads against localhost:9092 and leaves state in a static that other tests share.
The assertion above it — new AdminClientConfig(adminProperties).getPassword(...) — already proves the thing under test, which is that the provider reference survives filtering and resolves. Dropping the live client would give the same coverage without the singleton mutation.
There was a problem hiding this comment.
removed the live KafkaConsumer and AdminClient construction from this regression. It now verifies provider resolution through ConsumerConfig and AdminClientConfig only, so it does not mutate the process-wide singleton or create connection threads.
xiangfu0
left a comment
xiangfu0
left a comment
There was a problem hiding this comment.
Re-reviewed at f59403d. All eight of my earlier comments are addressed, and the ConfigUtils rework is better than what I asked for — a five-line connector-neutral escape with zero Kafka awareness in pinot-spi is exactly the right shape.
I verified independently:
- GET returns the raw stored template (
replaceVariables=false,PinotTableRestletResource:585), so a GET → edit → PUT round-trip doesn't destroy the escape.validateEnvironmentVariablesalso discards its resolved copy, so nothing unescaped is persisted back to ZK. FileConfigProvider.ALLOWED_PATHS_CONFIGandConfigTransformer.DEFAULT_PATTERNboth exist in 3.9.2 and 4.2.0, andkafka-clientsisprovidedscope inpinot-kafka-base— so the shared helper is binary-safe against both connectors.- CI is green, 12/12.
Two things I'd still like resolved, both inline. The first one matters because it defeats the fail-loudly behavior this revision added.
One doc note as well: the description frames this as upgrade-gated, which is right. Worth also saying what a rollback does — on an older Pinot, $${file:...} is left untouched and config.providers is stripped, so Kafka never resolves the reference and the client receives the literal string. That's a handshake failure rather than a plausible-looking wrong password, which is the good failure mode, but operators planning a rollback should be told.
| case STRING: | ||
| final String field = jsonNode.asText(); | ||
| if (field.startsWith("$${") && field.endsWith("}")) { | ||
| return JsonNodeFactory.instance.textNode(field.substring(1)); |
There was a problem hiding this 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.
There was a problem hiding this comment.
Documented in the companion docs branch at 63ca1f2e97. The escape is now explicitly described as whole-value only and non-recursive/non-composable. Embedded provider-shaped substrings do not have a literal escape.
| retryPolicy.attempt(() -> { | ||
| try { | ||
| consumer.set(new KafkaConsumer<>(filterKafkaProperties(consumerProp, CONSUMER_CONFIG_NAMES))); | ||
| consumer.set(new KafkaConsumer<>(KafkaConfigUtils.filterAndValidateKafkaProperties(consumerProp, |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed this. Filtering and validation now run once before entering the retry logic for Kafka 3.x and 4.x consumer creation, direct AdminClient creation, and shared AdminClient initialization. Retries now cover client construction only, so deterministic ConfigException failures surface immediately with their original message.
| Set<String> configuredProviders = getConfiguredProviders(properties); | ||
| for (String key : properties.stringPropertyNames()) { | ||
| Matcher matcher = ConfigTransformer.DEFAULT_PATTERN.matcher(properties.getProperty(key)); | ||
| while (matcher.find()) { |
There was a problem hiding this 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.
There was a problem hiding this comment.
Documented the stricter behavior in the companion docs branch. Embedded ${alias:...} substrings require a declared provider alias and there is no escape for preserving them literally. This behavior now fails explicitly instead of allowing unresolved provider-shaped content to pass through.
| return Paths.get(keyStoreLocation); | ||
| } | ||
|
|
||
| private static void writeKeyStoreAtomically(Path storePath, KeyStore keyStore, String password) |
There was a problem hiding this 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:
writeKeyStoreAtomicallyreplaces 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 existingFileAlreadyExistsException→ warn-and-skip path, which is a behavior change for current auto-SSL users.validatePrivateKeyMatchesPublicKeyis ~100 new lines spanning RSA / EC / EdDSA / DSA / RSASSA-PSS with PSS parameter derivation and a signature round-trip, now on theinitKeyStorepath of every existing auto-SSL user.validateAutoSslMateriallikewise parses certs and keys eagerly and wraps any exception inIllegalArgumentException.
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.
There was a problem hiding this comment.
Addressed this as well. I removed the unrelated atomic store replacement, eager certificate/key parsing, and key-pair proof changes from this PR. The remaining auto-SSL change is limited to ConfigProvider-reference preflight before any renewal, with focused coverage for truststore, keystore, and key passwords and rejection before mutation.
4 participants
Summary
config.providersandconfig.providers.<alias>.*through Kafka consumer and AdminClient property filteringWhy
Two independent transformations currently prevent Kafka ConfigProviders from working in realtime table configs:
${file:/vault/secrets/kafka.properties:keystore.password}as its own${name:default}expression while reading the table config. Kafka never receives the reference.ConsumerConfig.configNames()orAdminClientConfig.configNames(). Kafka's dynamicconfig.providers.*namespace is not part of those static name sets, so the provider configuration is removed before client construction.This change preserves a
${provider:...}reference only when that provider alias is declared byconfig.providersin the containing config object. Normal Pinot environment and system-property substitution remains unchanged, including across separate stream config maps.The Kafka client filter continues to reject unrelated Pinot stream settings and admits only the target client's known properties plus Kafka's reserved config-provider namespace. The shared AdminClient path now applies the same filtering.
Provider values are resolved when a Kafka client is constructed. Recreating a consumer therefore rereads the mounted provider file; hot reloading an already-running Kafka client is outside this change.
Kafka client properties should use their unprefixed names in
streamConfigs, for examplessl.keystore.password. Thestream.kafka.consumer.prop.*prefix is not a generic Kafka property namespace.Testing
./mvnw -pl pinot-spi test(787 tests passed)./mvnw -pl pinot-spi -Dtest=ConfigUtilsTest test(5 tests passed)./mvnw -pl pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0,pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0 -am -Dtest=KafkaPartitionLevelConnectionHandlerTest -Dsurefire.failIfNoSpecifiedTests=false test(3 tests passed for each connector)./mvnw -pl pinot-spi,pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0,pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0 spotless:check checkstyle:check license:checkThe connector regression creates a temporary FileConfigProvider file, carries the unresolved password reference through Pinot table-config resolution and Kafka property filtering, forwards
allowed.paths, verifies the resolved password, and constructs both a KafkaConsumer and the shared AdminClient. It runs against Kafka 3.9.2 and Kafka 4.2.0.User documentation lives in
pinot-contrib/pinot-docsand should be updated through its separate documentation workflow after this source change.Addresses #19184