xiangfu0 commented on code in PR #19202:
URL: https://github.com/apache/pinot/pull/19202#discussion_r3754559660
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java:
##########
@@ -33,6 +35,7 @@ private ConfigUtils() {
}
private static final Map<String, String> ENVIRONMENT_VARIABLES =
System.getenv();
+ private static final String CONFIG_PROVIDERS = "config.providers";
Review 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.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -166,7 +169,8 @@ private AdminClient
getOrCreateSharedAdminClientInternal(boolean isRetry) {
synchronized (this) {
ref = _sharedAdminClientRef;
if (ref == null) {
- ref =
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(_consumerProp);
+ ref = KafkaAdminClientManager.getInstance()
+ .getOrCreateAdminClient(filterKafkaProperties(_consumerProp,
ADMIN_CLIENT_CONFIG_NAMES));
Review 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.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -98,16 +99,18 @@ 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.
+ /// 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.
///
/// @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) {
+ @VisibleForTesting
+ static Properties filterKafkaProperties(Properties props, Set<String>
validConfigNames) {
Properties filteredProps = new Properties();
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 + ".")) {
Review 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.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/ConfigUtils.java:
##########
@@ -86,13 +96,17 @@ private static JsonNode
applyConfigWithEnvVariablesAndSystemProperties(Map<Strin
ArrayNode arrayNode = (ArrayNode) jsonNode;
for (int i = 0; i < arrayNode.size(); i++) {
JsonNode arrayElement = arrayNode.get(i);
- arrayNode.set(i,
applyConfigWithEnvVariablesAndSystemProperties(configValues, arrayElement));
+ arrayNode.set(i,
+ applyConfigWithEnvVariablesAndSystemProperties(configValues,
arrayElement, inheritedConfigProviders));
}
}
break;
case STRING:
final String field = jsonNode.asText();
- if (field.startsWith("${") && field.endsWith("}")) {
+ // 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)) {
Review 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.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -98,16 +99,18 @@ 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.
+ /// 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.
Review 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.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandler.java:
##########
@@ -98,16 +99,18 @@ 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.
+ /// 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.
///
/// @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) {
+ @VisibleForTesting
+ static Properties filterKafkaProperties(Properties props, Set<String>
validConfigNames) {
Properties filteredProps = new Properties();
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 + ".")) {
Review 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.
##########
pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java:
##########
@@ -56,6 +58,46 @@ public void testIndexingWithSystemProperties() {
System.clearProperty("AWS_SECRET_KEY");
}
+ @Test
+ public void testKafkaConfigProviderReferencesArePreserved() {
+ IndexingConfig indexingConfig = new IndexingConfig();
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("config.providers", "file");
+ 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 testConfigProviderReferencesAreScopedToDeclaringObject() {
Review 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.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConnectionHandlerTest.java:
##########
@@ -44,6 +61,63 @@ private StreamConfig createTestStreamConfig() {
return new StreamConfig("testTable_REALTIME", streamConfigMap);
}
+ @Test
+ public void testConfigProviderReferencesReachKafkaClients()
+ 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 =
+
KafkaPartitionLevelConnectionHandler.filterKafkaProperties(properties,
ConsumerConfig.configNames());
+
assertEquals(consumerProperties.getProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG),
"file");
+
assertEquals(consumerProperties.getProperty("config.providers.file.class"),
FileConfigProvider.class.getName());
+
assertEquals(consumerProperties.getProperty("config.providers.file.param.allowed.paths"),
+ providerFile.getParent().toString());
+ assertFalse(consumerProperties.containsKey("streamType"));
+ assertEquals(new
ConsumerConfig(consumerProperties).getPassword(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).value(),
+ "test-password");
+ try (KafkaConsumer<Bytes, Bytes> consumer = new
KafkaConsumer<>(consumerProperties)) {
+ assertNotNull(consumer);
+ }
+
+ Properties adminProperties =
+
KafkaPartitionLevelConnectionHandler.filterKafkaProperties(properties,
AdminClientConfig.configNames());
+
assertEquals(adminProperties.getProperty(AbstractConfig.CONFIG_PROVIDERS_CONFIG),
"file");
+ assertEquals(adminProperties.getProperty("config.providers.file.class"),
FileConfigProvider.class.getName());
+
assertEquals(adminProperties.getProperty("config.providers.file.param.allowed.paths"),
+ providerFile.getParent().toString());
+ assertFalse(adminProperties.containsKey("streamType"));
+ assertEquals(new
AdminClientConfig(adminProperties).getPassword(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).value(),
+ "test-password");
+ try (KafkaAdminClientManager.AdminClientReference adminClientReference =
+
KafkaAdminClientManager.getInstance().getOrCreateAdminClient(adminProperties)) {
Review 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.
--
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]
