You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2025/10/08 18:45:56 UTC
(pinot) branch master updated: Only include kafka properties when creating kafka clients (#16965)
This is an automated email from the ASF dual-hosted git repository.
jackie 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 9ddce8d6055 Only include kafka properties when creating kafka clients (#16965)
9ddce8d6055 is described below
commit 9ddce8d6055e43542ebe8cd416269cda0c456bd6
Author: harold-kfuse <77...@users.noreply.github.com>
AuthorDate: Wed Oct 8 11:45:51 2025 -0700
Only include kafka properties when creating kafka clients (#16965)
---
.../KafkaPartitionLevelConnectionHandler.java | 26 ++++++++++++++++++++--
.../KafkaPartitionLevelConnectionHandler.java | 26 ++++++++++++++++++++--
2 files changed, 48 insertions(+), 4 deletions(-)
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java
index e63f1d83388..e3362a0577a 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java
@@ -23,9 +23,11 @@ import com.google.common.util.concurrent.Uninterruptibles;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -49,6 +51,8 @@ import org.slf4j.LoggerFactory;
public abstract class KafkaPartitionLevelConnectionHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConnectionHandler.class);
+ private static final Set<String> CONSUMER_CONFIG_NAMES = ConsumerConfig.configNames();
+ private static final Set<String> ADMIN_CLIENT_CONFIG_NAMES = AdminClientConfig.configNames();
protected final KafkaPartitionLevelStreamConfig _config;
protected final String _clientId;
protected final int _partition;
@@ -82,13 +86,31 @@ 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;
+ }
+
@VisibleForTesting
protected Consumer<String, Bytes> createConsumer(Properties consumerProp) {
- return retry(() -> new KafkaConsumer<>(consumerProp), 5);
+ return retry(() -> new KafkaConsumer<>(filterKafkaProperties(consumerProp, CONSUMER_CONFIG_NAMES)), 5);
}
protected AdminClient createAdminClient() {
- return retry(() -> AdminClient.create(_consumerProp), 5);
+ return retry(() -> AdminClient.create(filterKafkaProperties(_consumerProp, ADMIN_CLIENT_CONFIG_NAMES)), 5);
}
private static <T> T retry(Supplier<T> s, int nRetries) {
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 e24a22c6c6f..ef39b256c7f 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
@@ -23,9 +23,11 @@ import com.google.common.util.concurrent.Uninterruptibles;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -48,6 +50,8 @@ import org.slf4j.LoggerFactory;
public abstract class KafkaPartitionLevelConnectionHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConnectionHandler.class);
+ private static final Set<String> CONSUMER_CONFIG_NAMES = ConsumerConfig.configNames();
+ private static final Set<String> ADMIN_CLIENT_CONFIG_NAMES = AdminClientConfig.configNames();
protected final KafkaPartitionLevelStreamConfig _config;
protected final String _clientId;
protected final int _partition;
@@ -81,13 +85,31 @@ 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;
+ }
+
@VisibleForTesting
protected Consumer<String, Bytes> createConsumer(Properties consumerProp) {
- return retry(() -> new KafkaConsumer<>(consumerProp), 5);
+ return retry(() -> new KafkaConsumer<>(filterKafkaProperties(consumerProp, CONSUMER_CONFIG_NAMES)), 5);
}
protected AdminClient createAdminClient() {
- return retry(() -> AdminClient.create(_consumerProp), 5);
+ return retry(() -> AdminClient.create(filterKafkaProperties(_consumerProp, ADMIN_CLIENT_CONFIG_NAMES)), 5);
}
private static <T> T retry(Supplier<T> s, int nRetries) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org