goutamadwant commented on code in PR #19202:
URL: https://github.com/apache/pinot/pull/19202#discussion_r3755334140
##########
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:
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.
--
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]
