xiangfu0 commented on code in PR #20112:
URL: https://github.com/apache/pinot/pull/20112#discussion_r4436867549
##########
pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealtimeTableOp.tsx:
##########
@@ -100,7 +100,7 @@ const defaultTableObj = {
"stream.kafka.topic.name": "",
"stream.kafka.broker.list": "",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka40.KafkaConsumerFactory",
Review Comment:
Removed the explicit factory key from all three UI defaults and the three
already-updated Helm examples. New configs use each component's StreamConfig
default during a rolling upgrade. The migration guide now recommends leaving
the key unset and waiting until every consuming component has Kafka 4 before
explicitly writing kafka40. Fixed in 72007a4136.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java:
##########
@@ -526,6 +533,26 @@ public void registerRecordReaderClass(String inputFormat,
String recordReaderCla
}
}
+ private String getBackwardCompatiblePluginName(String pluginName) {
+ return _backwardCompatiblePluginNames.getOrDefault(pluginName, pluginName);
+ }
+
+ /// Registers a direct alias for a renamed plugin, including
`plugins.include` and plugin-qualified class names.
+ /// Call during bootstrap, before creating plugin instances. If eager
initialization already skipped the replacement
+ /// because only the old name was included, loads that replacement now.
Repeated registration is harmless.
+ public synchronized void registerBackwardCompatiblePluginName(String
oldPluginName, String newPluginName) {
Review Comment:
Kept the registration API because StarTree uses it during production
bootstrap to register startree-kafka-2.0/3.0 aliases. Changed the mutable alias
registry to ConcurrentHashMap so unsynchronized class-loading reads are safe
while aliases are registered. The existing plugin-manager regression covers
late registration and repeated registration. Fixed in 72007a4136.
##########
pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java:
##########
@@ -71,7 +71,7 @@ public class LaunchSparkDataIngestionJobCommand extends
AbstractBaseAdminCommand
// Kafka plugins need to be excluded as they contain scala dependencies
which cause
// NoSuchMethodErrors with runtime spark.
// It is also fine to exclude Kafka plugins as they are not going to be used
in batch ingestion in any case
- @CommandLine.Option(names = {"-pluginsToExclude"}, defaultValue =
"pinot-kafka-3.0", required =
+ @CommandLine.Option(names = {"-pluginsToExclude"}, defaultValue =
"pinot-kafka-4.0", required =
Review Comment:
Explicit Spark plugin include/exclude lists now recognize pinot-kafka-2.0
and pinot-kafka-3.0 as aliases for the shipped Kafka 4 plugin. Replaced the
stale Scala comment. Extended the existing command test to exercise both flags
with Kafka 2/3/4 names and verify unrelated plugin selection is preserved.
Fixed in 72007a4136.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtils.java:
##########
@@ -48,15 +59,92 @@ private KafkaConfigUtils() {
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)
+ if (validConfigNames.contains(key) ||
(key.equals(AUTO_INCLUDE_JMX_REPORTER_CONFIG)
+ &&
validConfigNames.contains(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG))
+ || key.equals(AbstractConfig.CONFIG_PROVIDERS_CONFIG)
|| key.startsWith(CONFIG_PROVIDERS_PREFIX)) {
filteredProperties.put(key, properties.get(key));
}
}
validateConfigProviderReferences(filteredProperties);
+ normalizeLegacyProperties(filteredProperties);
return filteredProperties;
}
+ private static void normalizeLegacyProperties(Properties properties) {
+ boolean hasLegacyJmxSetting =
properties.containsKey(AUTO_INCLUDE_JMX_REPORTER_CONFIG);
Review Comment:
An omitted legacy flag now preserves Kafka 3 automatic JMX inclusion
whenever metric.reporters is explicitly configured, including empty lists,
subclasses, and ConfigProvider-backed reporters. Explicit false still preserves
only the configured reporters. Four cases in the existing KafkaConfigUtilsTest
failed before this fix; the full 28-case class passes after it, with real
Consumer/Admin reporter construction and unchanged input properties. Fixed in
72007a4136.
--
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]
