Jackie-Jiang commented on code in PR #20112:
URL: https://github.com/apache/pinot/pull/20112#discussion_r4485473750
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/utils/MiniKafkaCluster.java:
##########
@@ -19,153 +19,20 @@
package org.apache.pinot.plugin.stream.kafka40.utils;
import java.io.Closeable;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.NewPartitions;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.admin.RecordsToDelete;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.pinot.spi.stream.StreamDataServerStartable;
-import org.testcontainers.kafka.KafkaContainer;
-import org.testcontainers.utility.DockerImageName;
+import org.apache.pinot.plugin.stream.kafka40.server.EmbeddedKafkaCluster;
-/// MiniKafkaCluster for Kafka 4.x using Testcontainers with KRaft mode (no
ZooKeeper).
-/// Uses the apache/kafka-native image for fast startup.
-public final class MiniKafkaCluster implements StreamDataServerStartable,
Closeable {
-
- private static final String KAFKA_IMAGE = "apache/kafka:4.0.0";
-
- private final KafkaContainer _kafkaContainer;
-
- public MiniKafkaCluster(String brokerId)
- throws Exception {
- // Use apache/kafka image which supports KRaft mode
- _kafkaContainer = new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE));
- }
-
- @Override
- public void init(Properties props) {
- // No-op; test cluster is self-contained.
- }
-
- @Override
- public void start() {
- try {
- _kafkaContainer.start();
- } catch (Exception e) {
- throw new RuntimeException("Failed to start Kafka test container", e);
- }
- }
-
- @Override
- public void stop() {
- _kafkaContainer.stop();
- }
-
- @Override
- public void close()
- throws IOException {
- stop();
+/// Single-node, in-process Kafka 4 KRaft cluster used by the consumer tests.
Not thread-safe.
+public final class MiniKafkaCluster extends EmbeddedKafkaCluster implements
Closeable {
+ public MiniKafkaCluster(String brokerId) {
Review Comment:
MINOR: Use EmbeddedKafkaCluster directly
This wrapper now only accepts an ignored brokerId, aliases
bootstrapServers() through getKafkaServerAddress(), and provides an unused
close(). Its only two callers, KafkaPartitionLevelConsumerTest and
KafkaPartitionLevelConsumerDataLossIntegrationTest, already stop the cluster
explicitly. They can instantiate EmbeddedKafkaCluster directly and use
bootstrapServers(), allowing this test-only wrapper to be deleted.
##########
pom.xml:
##########
@@ -660,6 +660,14 @@
<dependencyManagement>
<dependencies>
+ <!-- Keep Kafka's testkit and the TestNG platform provider on the same
JUnit infrastructure. -->
+ <dependency>
+ <groupId>org.junit</groupId>
+ <artifactId>junit-bom</artifactId>
Review Comment:
MINOR: Consolidate JUnit dependency management
Move this junit-bom import into the Test Libraries section beside the
Mockito/Testcontainers/AssertJ BOMs, and remove the now-redundant managed
junit-jupiter-api and junit-platform-suite entries. The BOM manages both
Jupiter and Platform versions without adding their jars. All six current direct
declarations of those two artifacts explicitly set scope=test, so removing
their managed scope defaults preserves the current scopes. Retain the
explanation that Pinot tests use TestNG and the Platform supports execution;
remove the obsolete comment claiming the embedded Kafka broker's readiness wait
asserts through JUnit.
##########
pinot-integration-test-base/pom.xml:
##########
@@ -92,32 +92,13 @@
<dependency>
<groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-3.0</artifactId>
+ <artifactId>pinot-kafka-4.0</artifactId>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
- <artifactId>kafka_${scala.compat.version}</artifactId>
- <classifier>test</classifier>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <classifier>test</classifier>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-server-common</artifactId>
- <classifier>test</classifier>
+ <artifactId>kafka-test-common-runtime</artifactId>
Review Comment:
MINOR: Remove redundant direct Jupiter API declarations
With the Kafka 4 testkit dependency, the direct test-scoped
junit-jupiter-api declarations in pinot-integration-test-base,
pinot-integration-tests, and pinot-connectors/pinot-flink-connector are
redundant: the current testkit graph already supplies Jupiter API transitively,
and these modules do not import Jupiter APIs. Remove those three declarations
and their adjacent embedded-broker comments. This leaves Jupiter API
transitively test-scoped; excluding Jupiter engines or other jars would be a
separate change requiring validation.
##########
pinot-plugins/pinot-input-format/pinot-json/README.md:
##########
@@ -96,7 +96,7 @@ All examples use a Kafka real-time table; substitute your
stream type (`stream.<
"streamType": "kafka",
"stream.kafka.topic.name": "events",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
- "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:
MINOR: Omit the optional Kafka factory from ordinary examples
Remove stream.kafka.consumer.factory.class.name from both examples here
(lines 99 and 124). StreamConfig already defaults to the Kafka 4 factory in
this PR, and leaving the key unset lets each component use its own default
during a rolling upgrade, consistent with the UI/Helm changes. Keep the
configurable override for custom factories and the intentional optional-value
example in the stream-ingestion migration README.
##########
pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java:
##########
@@ -267,14 +266,20 @@ private void addDepsJarToDistributedCache(SparkLauncher
sparkLauncher, String de
}
}
- private boolean shouldLoadPlugin(String parentDir) {
- boolean shouldLoadPlugin = _pluginsToLoad == null &&
!_pluginsToExclude.contains(parentDir)
+ @VisibleForTesting
+ boolean shouldLoadPlugin(String parentDir) {
+ boolean shouldLoadPlugin = _pluginsToLoad == null &&
!containsPlugin(_pluginsToExclude, parentDir)
Review Comment:
MINOR: Make plugin-selection precedence explicit
Keep the small Kafka alias matcher, but replace the accumulated Boolean
expression with early returns so the existing precedence is visible: the
selected Spark plugin is always included; an explicit include list takes
precedence over exclusions; otherwise exclude the configured plugins and other
Spark plugins.
```java
if (_sparkVersion.getPluginName().equals(parentDir)) {
return true;
}
if (_pluginsToLoad != null) {
return containsPlugin(_pluginsToLoad, parentDir);
}
return !parentDir.contains("spark")
&& !containsPlugin(_pluginsToExclude, parentDir);
```
This preserves the old Kafka include/exclude aliases without adding an
option-normalization lifecycle.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java:
##########
@@ -526,6 +534,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) {
+ String previousName = _backwardCompatiblePluginNames.put(oldPluginName,
newPluginName);
+ if (newPluginName.equals(previousName) ||
StringUtils.isEmpty(_pluginsDirectories)
Review Comment:
MINOR: Separate alias registration from replacement loading
Keep the public registration API, concurrent maps, and late-loading behavior
needed by StarTree bootstrap. For readability, extract
loadReplacementPluginIfNeeded() and split the long compound return condition
into clear guard groups: unchanged mapping, no applicable configured include,
and replacement already loaded. Eager initialization may have skipped the
replacement before registration, so preserve the subsequent scan/load path and
the existing synchronization.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaConfigUtils.java:
##########
@@ -48,15 +59,94 @@ 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 hasJmxSettings =
properties.containsKey(AUTO_INCLUDE_JMX_REPORTER_CONFIG)
+ ||
properties.containsKey(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG);
+ List<String> compatibilityKeys = new ArrayList<>();
Review Comment:
MINOR: Simplify the compatibility normalization flow
Use one fixed immutable list of the three compatibility keys instead of
building a conditional list on each call; resolveCompatibilityReferences
already ignores absent properties. Keep the hasJmxSettings guard for the actual
JMX rewrite, and extract the independent JMX and OAuth translations into named
helpers so normalizeLegacyProperties reads as resolve references, normalize
JMX, then normalize OAuth. Preserve provider validation, restricted reference
resolution, and the current JMX default/explicit-false behavior.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java:
##########
@@ -228,7 +233,8 @@ public HashMap<String, File> getPluginsToLoad(String
pluginsDirectories, String
FileUtils.listFiles(new File(pluginsDirectory), new
String[]{JAR_FILE_EXTENSION}, true);
List<String> pluginsToLoad = null;
if (!StringUtils.isEmpty(pluginsInclude)) {
Review Comment:
MINOR: Resolve the include list once per scan
The same pluginsInclude string is split and its aliases resolved for every
entry in pluginsDirectories. Compute the normalized include list once before
the directory loop, then reuse it when filtering each directory. This keeps the
existing alias behavior while avoiding repeated parsing and resolution.
--
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]
