xiangfu0 commented on code in PR #20112:
URL: https://github.com/apache/pinot/pull/20112#discussion_r4486119019
##########
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:
Fixed in 10fe73e086. Removed the three direct Jupiter API declarations and
adjacent embedded-broker comments. Dependency trees for integration-test-base,
integration-tests and the Flink connector all retain Jupiter API 6.1.3 in test
scope through the Kafka 4 testkit. The existing transitive test graph remains
intact.
##########
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:
Fixed in 10fe73e086. Deleted MiniKafkaCluster. Both callers now construct
EmbeddedKafkaCluster directly and use bootstrapServers(), retaining their
explicit stop() teardown. The 13 consumer cases and three data-loss cases pass
against the real embedded broker.
##########
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:
Fixed in 10fe73e086. Replaced the accumulated expression with the requested
early returns: selected Spark plugin first, explicit includes second, then the
default/exclusion filter. Kept the small Kafka alias matcher. Both existing
Spark-selection tests pass.
--
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]
