You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/02/20 00:50:38 UTC
(pinot) branch master updated: Remove kafka-2.0 plugin and migrate to kafka30 (#17602)
This is an automated email from the ASF dual-hosted git repository.
xiangfu 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 380477a6160 Remove kafka-2.0 plugin and migrate to kafka30 (#17602)
380477a6160 is described below
commit 380477a61605da2dd601921b94ea22e9d25d8f3f
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Thu Feb 19 16:50:32 2026 -0800
Remove kafka-2.0 plugin and migrate to kafka30 (#17602)
---
AGENTS.md | 1 -
.../config/feature-test-2-realtime.json | 2 +-
.../config/feature-test-2-realtime.json | 2 +-
.../config/feature-test-3-realtime.json | 2 +-
docker/images/pinot/README.md | 2 +-
.../airlineStats_realtime_table_config.json | 2 +-
.../meetupRsvp_realtime_table_config.json | 2 +-
helm/pinot/pinot-github-events-setup.yml | 2 +-
helm/pinot/pinot-realtime-quickstart.yml | 4 +-
.../java/org/apache/pinot/compat/StreamOp.java | 2 +-
.../Homepage/Operations/AddIngestionComponent.tsx | 2 +-
.../Operations/AddRealTimeIngestionComponent.tsx | 2 +-
.../Homepage/Operations/AddRealtimeTableOp.tsx | 2 +-
pinot-distribution/pinot-assembly.xml | 16 +-
.../tests/BaseClusterIntegrationTest.java | 410 +++++++++++-
.../tests/ClusterIntegrationTestUtils.java | 41 +-
.../tests/BaseRealtimeClusterIntegrationTest.java | 12 +-
.../tests/CommitTimeCompactionIntegrationTest.java | 1 +
...tlyOnceKafkaRealtimeClusterIntegrationTest.java | 15 +-
.../tests/LLCRealtimeClusterIntegrationTest.java | 4 +-
.../MergeRollupMinionClusterIntegrationTest.java | 4 +-
.../CustomDataQueryClusterIntegrationTest.java | 156 +++--
...aIncreaseDecreasePartitionsIntegrationTest.java | 7 +-
.../simpleMeetup_realtime_table_config.json | 2 +-
pinot-perf/pom.xml | 2 +-
.../pinot-input-format/pinot-protobuf/pom.xml | 15 -
pinot-plugins/pinot-stream-ingestion/README.md | 2 +-
.../pinot-kafka-2.0/README.md | 42 --
.../pinot-stream-ingestion/pinot-kafka-2.0/pom.xml | 79 ---
.../stream/kafka20/KafkaConsumerFactory.java | 70 --
.../KafkaPartitionLevelConnectionHandler.java | 234 -------
.../kafka20/KafkaPartitionLevelConsumer.java | 136 ----
.../kafka20/KafkaStreamMetadataProvider.java | 285 ---------
.../SynchronizedKafkaStreamMetadataProvider.java | 44 --
.../kafka20/server/KafkaDataServerStartable.java | 171 -----
.../kafka20/KafkaAdminClientManagerTest.java | 130 ----
.../KafkaPartitionLevelConnectionHandlerTest.java | 103 ---
...tionLevelConsumerBackwardCompatibilityTest.java | 30 -
.../kafka20/KafkaPartitionLevelConsumerTest.java | 509 ---------------
.../KafkaPartitionLevelStreamConfigTest.java | 220 -------
.../stream/kafka20/utils/MiniKafkaCluster.java | 125 ----
.../pinot-kafka-2.0/src/test/resources/log4j2.xml | 36 --
.../kafka30/KafkaStreamMetadataProvider.java | 84 ++-
.../stream/kafka30}/server/KafkaDataProducer.java | 9 +-
.../kafka30/server/KafkaServerStartable.java | 701 +++++++++++++++++++++
...g.apache.pinot.spi.stream.StreamConsumerFactory | 2 +-
.../kafka30/KafkaPartitionLevelConsumerTest.java | 26 +-
.../kafka30/server/KafkaServerStartableTest.java | 102 +++
.../stream/kafka30/utils/MiniKafkaCluster.java | 138 ----
.../kafka40/KafkaStreamMetadataProvider.java | 84 ++-
.../kafka40/KafkaPartitionLevelConsumerTest.java | 8 +-
.../stream/kafka40/utils/MiniKafkaCluster.java | 88 ++-
pinot-plugins/pinot-stream-ingestion/pom.xml | 1 -
.../org/apache/pinot/spi/plugin/PluginManager.java | 5 +-
.../spi/stream/StreamDataServerStartable.java | 23 +
.../apache/pinot/spi/plugin/PluginManagerTest.java | 5 +-
pinot-tools/pom.xml | 24 -
.../org/apache/pinot/tools/QuickStartBase.java | 75 ++-
.../LaunchSparkDataIngestionJobCommand.java | 2 +-
.../tools/admin/command/QuickStartCommand.java | 12 +
.../tools/admin/command/StartKafkaCommand.java | 39 +-
.../tools/admin/command/StopProcessCommand.java | 113 +++-
.../pinot/tools/utils/KafkaStarterUtils.java | 138 ++--
.../conf/sample_realtime_table_config.json | 2 +-
.../githubEvents_realtime_table_config.json | 2 +-
.../airlineStats_realtime_table_config.json | 2 +-
.../docker/airlineStats_realtime_table_config.json | 2 +-
.../dailySales_realtime_table_config.json | 2 +-
.../fineFoodReviews_realtime_table_config.json | 2 +-
.../githubEvents_realtime_table_config.json | 2 +-
.../meetupRsvp_realtime_table_config.json | 2 +-
...eetupRsvpComplexType_realtime_table_config.json | 2 +-
.../meetupRsvpJson_realtime_table_config.json | 2 +-
...lRequestMergedEvents_realtime_table_config.json | 2 +-
...lRequestMergedEvents_realtime_table_config.json | 2 +-
...upsertJsonMeetupRsvp_realtime_table_config.json | 2 +-
.../upsertMeetupRsvp_realtime_table_config.json | 2 +-
...ertPartialMeetupRsvp_realtime_table_config.json | 2 +-
pom.xml | 10 +-
79 files changed, 1974 insertions(+), 2673 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index 9c59eacb726..2db56d1dd9f 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -72,7 +72,6 @@ repo. It is intentionally short and focused on day-to-day work.
- pinot-batch-ingestion-standalone: standalone batch ingestion implementation.
- pinot-stream-ingestion: stream ingestion plugin family.
- pinot-kafka-base: shared Kafka ingestion base classes.
- - pinot-kafka-2.0: Kafka 2.x ingestion implementation.
- pinot-kafka-3.0: Kafka 3.x ingestion implementation.
- pinot-kinesis: AWS Kinesis ingestion implementation.
- pinot-pulsar: Apache Pulsar ingestion implementation.
diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json b/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
index 27beee9669b..4ef023e6b58 100644
--- a/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
+++ b/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
@@ -44,7 +44,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "PinotRealtimeFeatureTest2Event",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
"stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
diff --git a/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json b/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
index 27beee9669b..4ef023e6b58 100644
--- a/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
+++ b/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
@@ -44,7 +44,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "PinotRealtimeFeatureTest2Event",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
"stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
diff --git a/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json b/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
index 5319be1b507..31864393acf 100644
--- a/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
+++ b/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
@@ -26,7 +26,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "PinotRealtimeFeatureTest3Event",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
"stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
diff --git a/docker/images/pinot/README.md b/docker/images/pinot/README.md
index d57d1283cff..15778f233fa 100644
--- a/docker/images/pinot/README.md
+++ b/docker/images/pinot/README.md
@@ -147,7 +147,7 @@ docker run -p 8098:8098 pinot:release-0.1.0 StartServer -zkAddress 172.17.0.2:21
### Use docker compose to bring up Pinot stack
-Please note that this quickstart is based on Kafka 2.0.
+Please note that this quickstart is based on Kafka 3.0.
Below is a script to use docker compose to bring up zookeeper/kafka/pinot-controller/pinot-broker/pinot-server
```SHELL
diff --git a/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json b/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
index 25c210005f3..48fc7f0f8c8 100644
--- a/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
+++ b/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
@@ -18,7 +18,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.hlc.zk.connect.string": "pinot-zookeeper:2181/kafka",
"stream.kafka.zk.broker.url": "pinot-zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
diff --git a/docker/images/pinot/examples/docker/table-configs/meetupRsvp_realtime_table_config.json b/docker/images/pinot/examples/docker/table-configs/meetupRsvp_realtime_table_config.json
index 0e460f09b38..ff31b0a4668 100644
--- a/docker/images/pinot/examples/docker/table-configs/meetupRsvp_realtime_table_config.json
+++ b/docker/images/pinot/examples/docker/table-configs/meetupRsvp_realtime_table_config.json
@@ -16,7 +16,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "meetupRSVPEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "kafka:9092",
"stream.kafka.consumer.prop.auto.offset.reset": "largest",
"realtime.segment.flush.threshold.time": "12h",
diff --git a/helm/pinot/pinot-github-events-setup.yml b/helm/pinot/pinot-github-events-setup.yml
index 231c951fcc8..894ab0e9b33 100644
--- a/helm/pinot/pinot-github-events-setup.yml
+++ b/helm/pinot/pinot-github-events-setup.yml
@@ -46,7 +46,7 @@ data:
"streamType": "kafka",
"stream.kafka.topic.name": "pullRequestMergedEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "12h",
diff --git a/helm/pinot/pinot-realtime-quickstart.yml b/helm/pinot/pinot-realtime-quickstart.yml
index a7a084df12e..d2bb95b94bf 100644
--- a/helm/pinot/pinot-realtime-quickstart.yml
+++ b/helm/pinot/pinot-realtime-quickstart.yml
@@ -44,7 +44,7 @@ data:
"streamType": "kafka",
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.hlc.zk.connect.string": "kafka-zookeeper:2181",
"stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
@@ -80,7 +80,7 @@ data:
"stream.kafka.topic.name": "flights-realtime-avro",
"stream.kafka.decoder.class.name": "org.apache.pinot.core.realtime.stream.SimpleAvroMessageDecoder",
"stream.kafka.decoder.prop.schema": "{\"type\":\"record\",\"name\":\"Flight\",\"namespace\":\"pinot\",\"fields\":[{\"name\":\"DaysSinceEpoch\",\"type\":[\"int\"]},{\"name\":\"Year\",\"type\":[\"int\"]},{\"name\":\"Quarter\",\"type\":[\"int\"]},{\"name\":\"Month\",\"type\":[\"int\"]},{\"name\":\"DayofMonth\",\"type\":[\"int\"]},{\"name\":\"DayOfWeek\",\"type\":[\"int\"]},{\"name\":\"FlightDate\",\"type\":[\"string\"]},{\"name\":\"UniqueCarrier\",\"type\":[\"string\"]},{\"name\": [...]
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.hlc.zk.connect.string": "kafka-zookeeper:2181",
"stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
index 7abf0de8a71..12eaee7360b 100644
--- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
+++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
@@ -168,7 +168,7 @@ public class StreamOp extends BaseOp {
final Map<String, Object> config = new HashMap<>();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ClusterDescriptor.getInstance().getKafkaServerUrl());
- config.put(AdminClientConfig.CLIENT_ID_CONFIG, "Kafka2AdminClient-" + UUID.randomUUID());
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, "KafkaAdminClient-" + UUID.randomUUID());
config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 15000);
AdminClient adminClient = KafkaAdminClient.create(config);
NewTopic topic = new NewTopic(topicName, partitions, KAFKA_REPLICATION_FACTOR);
diff --git a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddIngestionComponent.tsx b/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddIngestionComponent.tsx
index 0d24a0740d9..e90de84f83c 100644
--- a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddIngestionComponent.tsx
+++ b/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddIngestionComponent.tsx
@@ -88,7 +88,7 @@ export default function AddIngestionComponent({
"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.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name":"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.decoder.class.name":"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
"realtime.segment.flush.threshold.rows": "0",
"realtime.segment.flush.threshold.segment.rows": "0",
diff --git a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealTimeIngestionComponent.tsx b/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealTimeIngestionComponent.tsx
index b4ae459d2c1..f89274d1c2c 100644
--- a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealTimeIngestionComponent.tsx
+++ b/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealTimeIngestionComponent.tsx
@@ -88,7 +88,7 @@ export default function AddRealTimeIngestionComponent({
"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.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name":"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.decoder.class.name":"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
"realtime.segment.flush.threshold.rows": "0",
"realtime.segment.flush.threshold.segment.rows": "0",
diff --git a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealtimeTableOp.tsx b/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealtimeTableOp.tsx
index a9851814071..176453d2e24 100644
--- a/pinot-controller/src/main/resources/app/components/Homepage/Operations/AddRealtimeTableOp.tsx
+++ b/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.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
"realtime.segment.flush.threshold.rows": "0",
"realtime.segment.flush.threshold.segment.rows": "0",
diff --git a/pinot-distribution/pinot-assembly.xml b/pinot-distribution/pinot-assembly.xml
index 078205f2131..2d839d4a87f 100644
--- a/pinot-distribution/pinot-assembly.xml
+++ b/pinot-distribution/pinot-assembly.xml
@@ -48,10 +48,10 @@
<!-- Start Include Pinot Stream Ingestion Plugins-->
<file>
<source>
- ${pinot.root}/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/target/pinot-kafka-2.0-${project.version}-shaded.jar
+ ${pinot.root}/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/target/pinot-kafka-3.0-${project.version}-shaded.jar
</source>
<destName>
- plugins/pinot-stream-ingestion/pinot-kafka-2.0/pinot-kafka-2.0-${project.version}-shaded.jar
+ plugins/pinot-stream-ingestion/pinot-kafka-3.0/pinot-kafka-3.0-${project.version}-shaded.jar
</destName>
</file>
<file>
@@ -263,18 +263,6 @@
<directory>${pinot.root}/pinot-tools/src/main/resources/conf</directory>
<outputDirectory>conf</outputDirectory>
</fileSet>
- <fileSet>
- <useDefaultExcludes>false</useDefaultExcludes>
- <directory>${pinot.root}/pinot-tools/src/main/resources/examples/stream/meetupRsvp/kafka_2.0
- </directory>
- <outputDirectory>examples/stream/meetupRsvp/</outputDirectory>
- </fileSet>
- <fileSet>
- <useDefaultExcludes>false</useDefaultExcludes>
- <directory>${pinot.root}/pinot-tools/src/main/resources/examples/stream/airlineStats/kafka_2.0
- </directory>
- <outputDirectory>examples/stream/airlineStats/</outputDirectory>
- </fileSet>
<fileSet>
<useDefaultExcludes>false</useDefaultExcludes>
<directory>${pinot.root}/pinot-tools/src/main/resources/examples</directory>
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index d65149beae7..ca59d7f33dd 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -24,6 +24,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.net.ServerSocket;
import java.net.URL;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -32,24 +33,43 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
+import org.apache.helix.model.IdealState;
import org.apache.helix.task.TaskPartitionState;
import org.apache.helix.task.TaskState;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.pinot.client.ConnectionFactory;
import org.apache.pinot.client.JsonAsyncHttpPinotClientTransportFactory;
import org.apache.pinot.client.PinotClientTransportFactory;
import org.apache.pinot.client.ResultSetGroup;
import org.apache.pinot.common.restlet.resources.ValidDocIdsMetadataInfo;
import org.apache.pinot.common.restlet.resources.ValidDocIdsType;
+import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.TarCompressionUtils;
import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.plugin.inputformat.csv.CSVMessageDecoder;
import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
+import org.apache.pinot.plugin.stream.kafka30.server.KafkaServerStartable;
import org.apache.pinot.server.starter.helix.BaseServerStarter;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
import org.apache.pinot.spi.config.table.DedupConfig;
@@ -70,6 +90,7 @@ import org.apache.pinot.spi.data.TimeBoundaryConfig;
import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamDataServerStartable;
+import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.Enablement;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.LogicalTableConfigBuilder;
@@ -78,6 +99,8 @@ import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.tools.utils.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
import org.intellij.lang.annotations.Language;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -85,6 +108,7 @@ import org.testng.Assert;
* Shared implementation details of the cluster integration tests.
*/
public abstract class BaseClusterIntegrationTest extends ClusterTest {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseClusterIntegrationTest.class);
// Default settings
protected static final String DEFAULT_TABLE_NAME = "mytable";
@@ -101,6 +125,10 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
protected static final int DEFAULT_LLC_NUM_KAFKA_BROKERS = 2;
protected static final int DEFAULT_LLC_NUM_KAFKA_PARTITIONS = 2;
protected static final int DEFAULT_MAX_NUM_KAFKA_MESSAGES_PER_BATCH = 10000;
+ private static final int KAFKA_START_MAX_ATTEMPTS = 3;
+ private static final long KAFKA_START_RETRY_WAIT_MS = 2_000L;
+ private static final long KAFKA_CLUSTER_READY_TIMEOUT_MS = 120_000L;
+ private static final long KAFKA_TOPIC_READY_TIMEOUT_MS = 120_000L;
protected static final List<String> DEFAULT_NO_DICTIONARY_COLUMNS =
Arrays.asList("ActualElapsedTime", "ArrDelay", "DepDelay", "CRSDepTime");
protected static final String DEFAULT_SORTED_COLUMN = "Carrier";
@@ -114,6 +142,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
protected final File _segmentDir = new File(_tempDir, "segmentDir");
protected final File _tarDir = new File(_tempDir, "tarDir");
protected List<StreamDataServerStartable> _kafkaStarters;
+ private List<KafkaBrokerConfig> _kafkaBrokerConfigs;
protected org.apache.pinot.client.Connection _pinotConnection;
protected org.apache.pinot.client.Connection _pinotConnectionV2;
@@ -170,10 +199,6 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
return _kafkaStarters.get(idx).getPort();
}
- protected String getKafkaZKAddress() {
- return getZkUrl() + "/kafka";
- }
-
protected int getNumKafkaPartitions() {
return DEFAULT_LLC_NUM_KAFKA_PARTITIONS;
}
@@ -377,11 +402,14 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BROKER_LIST),
- "localhost:" + _kafkaStarters.get(0).getPort());
+ getKafkaBrokerList());
if (useKafkaTransaction()) {
streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL),
KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL_READ_COMMITTED);
+ // Ensure the consumer can fetch complete transactional batches plus commit markers.
+ streamConfigMap.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
+ Integer.toString(10 * 1024 * 1024));
}
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), getStreamConsumerFactoryClassName());
@@ -398,6 +426,17 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
return streamConfigMap;
}
+ protected String getKafkaBrokerList() {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < _kafkaStarters.size(); i++) {
+ if (i > 0) {
+ builder.append(',');
+ }
+ builder.append("localhost:").append(_kafkaStarters.get(i).getPort());
+ }
+ return builder.toString();
+ }
+
/**
* Creates a new REALTIME table config.
*/
@@ -776,21 +815,327 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
}
protected void startKafkaWithoutTopic() {
- startKafkaWithoutTopic(KafkaStarterUtils.DEFAULT_KAFKA_PORT);
+ int requestedBrokers = getNumKafkaBrokers();
+ List<KafkaBrokerConfig> brokerConfigs = getOrCreateKafkaBrokerConfigs(requestedBrokers);
+ Throwable lastFailure = null;
+ for (int attempt = 1; attempt <= KAFKA_START_MAX_ATTEMPTS; attempt++) {
+ String clusterId = UUID.randomUUID().toString().replace("-", "");
+ String networkName = "pinot-it-kafka-" + UUID.randomUUID().toString().replace("-", "");
+ String quorumVoters = brokerConfigs.stream()
+ .map(config -> config._brokerId + "@" + config._containerName + ":9093")
+ .collect(Collectors.joining(","));
+
+ List<StreamDataServerStartable> kafkaStarters = new ArrayList<>(requestedBrokers);
+ try {
+ for (KafkaBrokerConfig brokerConfig : brokerConfigs) {
+ StreamDataServerStartable kafkaStarter =
+ createKafkaServerStarter(brokerConfig, clusterId, networkName, quorumVoters, requestedBrokers);
+ kafkaStarter.start();
+ kafkaStarters.add(kafkaStarter);
+ }
+ _kafkaStarters = kafkaStarters;
+ waitForKafkaClusterReady(getKafkaBrokerList(), requestedBrokers, useKafkaTransaction());
+ if (attempt > 1) {
+ LOGGER.info("Kafka startup succeeded on retry attempt {}/{}", attempt, KAFKA_START_MAX_ATTEMPTS);
+ }
+ return;
+ } catch (Throwable t) {
+ if (t instanceof Error && !(t instanceof AssertionError)) {
+ throw (Error) t;
+ }
+
+ lastFailure = t;
+ LOGGER.warn("Kafka startup attempt {}/{} failed; stopping started brokers before retry", attempt,
+ KAFKA_START_MAX_ATTEMPTS, t);
+ _kafkaStarters = kafkaStarters;
+ try {
+ stopKafka();
+ } catch (RuntimeException stopException) {
+ LOGGER.warn("Kafka cleanup failed after startup attempt {}/{}", attempt, KAFKA_START_MAX_ATTEMPTS,
+ stopException);
+ t.addSuppressed(stopException);
+ }
+
+ if (attempt < KAFKA_START_MAX_ATTEMPTS) {
+ try {
+ Thread.sleep(KAFKA_START_RETRY_WAIT_MS);
+ } catch (InterruptedException interruptedException) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting to retry Kafka startup", interruptedException);
+ }
+ }
+ }
+ }
+
+ _kafkaBrokerConfigs = null;
+ throw new RuntimeException("Failed to start Kafka cluster after " + KAFKA_START_MAX_ATTEMPTS + " attempts",
+ lastFailure);
+ }
+
+ private List<KafkaBrokerConfig> getOrCreateKafkaBrokerConfigs(int brokerCount) {
+ if (_kafkaBrokerConfigs != null && _kafkaBrokerConfigs.size() == brokerCount) {
+ return _kafkaBrokerConfigs;
+ }
+ _kafkaBrokerConfigs = createKafkaBrokerConfigs(brokerCount);
+ return _kafkaBrokerConfigs;
+ }
+
+ private StreamDataServerStartable createKafkaServerStarter(KafkaBrokerConfig brokerConfig, String clusterId,
+ String networkName, String quorumVoters, int clusterSize) {
+ Properties serverProperties = new Properties();
+ serverProperties.put("kafka.server.owner.name", getClass().getSimpleName());
+ serverProperties.put("kafka.server.bootstrap.servers", "localhost:" + brokerConfig._port);
+ serverProperties.put("kafka.server.port", Integer.toString(brokerConfig._port));
+ serverProperties.put("kafka.server.broker.id", Integer.toString(brokerConfig._brokerId));
+ serverProperties.put("kafka.server.allow.managed.for.configured.broker", "true");
+ serverProperties.put("kafka.server.container.name", brokerConfig._containerName);
+ serverProperties.put("kafka.server.network.name", networkName);
+ serverProperties.put("kafka.server.cluster.id", clusterId);
+ serverProperties.put("kafka.server.cluster.size", Integer.toString(clusterSize));
+ serverProperties.put("kafka.server.controller.quorum.voters", quorumVoters);
+ serverProperties.put("kafka.server.internal.host", brokerConfig._containerName);
+ serverProperties.put("kafka.server.skip.readiness.check", "true");
+ KafkaServerStartable kafkaServerStartable = new KafkaServerStartable();
+ kafkaServerStartable.init(serverProperties);
+ return kafkaServerStartable;
+ }
+
+ private List<KafkaBrokerConfig> createKafkaBrokerConfigs(int brokerCount) {
+ String containerPrefix = "pinot-it-kafka-" + UUID.randomUUID().toString().replace("-", "");
+ List<KafkaBrokerConfig> brokerConfigs = new ArrayList<>(brokerCount);
+ for (int i = 0; i < brokerCount; i++) {
+ int brokerId = i + 1;
+ int port = getAvailablePort();
+ String containerName = containerPrefix + "-" + brokerId;
+ brokerConfigs.add(new KafkaBrokerConfig(brokerId, port, containerName));
+ }
+ return brokerConfigs;
+ }
+
+ private static int getAvailablePort() {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to find an available port for Kafka", e);
+ }
}
- protected void startKafkaWithoutTopic(int port) {
- _kafkaStarters = KafkaStarterUtils.startServers(getNumKafkaBrokers(), port, getKafkaZKAddress(),
- KafkaStarterUtils.getDefaultKafkaConfiguration());
+ private void waitForKafkaClusterReady(String brokerList, int brokerCount, boolean requireTransactions) {
+ TestUtils.waitForCondition(aVoid -> isKafkaClusterReady(brokerList, brokerCount), 200L,
+ KAFKA_CLUSTER_READY_TIMEOUT_MS,
+ "Kafka brokers are not ready");
+ if (requireTransactions) {
+ TestUtils.waitForCondition(aVoid -> canInitTransactions(brokerList), 500L, KAFKA_CLUSTER_READY_TIMEOUT_MS,
+ "Kafka transaction coordinator is not ready");
+ }
+ }
+
+ private boolean isKafkaClusterReady(String brokerList, int brokerCount) {
+ Properties adminProps = new Properties();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+ adminProps.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ adminProps.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");
+ try (AdminClient adminClient = AdminClient.create(adminProps)) {
+ return adminClient.describeCluster().nodes().get(5, TimeUnit.SECONDS).size() == brokerCount;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private boolean canInitTransactions(String brokerList) {
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "pinot-txn-ready-" + UUID.randomUUID());
+ producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000");
+ producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps)) {
+ producer.initTransactions();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
}
protected void createKafkaTopic(String topic) {
- _kafkaStarters.get(0).createTopic(topic, KafkaStarterUtils.getTopicCreationProps(getNumKafkaPartitions()));
+ int replicationFactor = Math.max(1, Math.min(getNumKafkaBrokers(), _kafkaStarters.size()));
+ createKafkaTopic(topic, getNumKafkaPartitions(), replicationFactor);
}
- protected void stopKafka() {
+ protected void createKafkaTopic(String topic, int numPartitions) {
+ createKafkaTopic(topic, numPartitions, Math.max(1, Math.min(getNumKafkaBrokers(), _kafkaStarters.size())));
+ }
+
+ protected void createKafkaTopic(String topic, int numPartitions, int replicationFactor) {
+ int expectedReplicationFactor = Math.max(1, Math.min(replicationFactor, _kafkaStarters.size()));
+ _kafkaStarters.get(0).createTopic(
+ topic,
+ KafkaStarterUtils.getTopicCreationProps(numPartitions, expectedReplicationFactor));
+ waitForKafkaTopicReady(topic, numPartitions, expectedReplicationFactor);
+ waitForKafkaTopicMetadataReadyForConsumer(topic, numPartitions);
+ }
+
+ protected void deleteKafkaTopic(String topic) {
+ _kafkaStarters.get(0).deleteTopic(topic);
+ waitForKafkaTopicDeleted(topic);
+ }
+
+ protected void waitForKafkaTopicDeleted(String topic) {
+ TestUtils.waitForCondition(aVoid -> isKafkaTopicDeleted(topic), 200L, KAFKA_TOPIC_READY_TIMEOUT_MS,
+ "Kafka topic '" + topic + "' is not deleted");
+ }
+
+ private boolean isKafkaTopicDeleted(String topic) {
+ Properties adminProps = new Properties();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaBrokerList());
+ adminProps.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ adminProps.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");
+ try (AdminClient adminClient = AdminClient.create(adminProps)) {
+ return !adminClient.listTopics().names().get(5, TimeUnit.SECONDS).contains(topic);
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private void waitForKafkaTopicReady(String topic, int expectedPartitions, int expectedReplicationFactor) {
+ TestUtils.waitForCondition(
+ aVoid -> isKafkaTopicReady(topic, expectedPartitions, expectedReplicationFactor),
+ 200L, KAFKA_TOPIC_READY_TIMEOUT_MS, "Kafka topic '" + topic + "' is not fully ready");
+ }
+
+ private void waitForKafkaTopicMetadataReadyForConsumer(String topic, int expectedPartitions) {
+ TestUtils.waitForCondition(aVoid -> isKafkaTopicMetadataReadyForConsumer(topic, expectedPartitions), 200L,
+ KAFKA_TOPIC_READY_TIMEOUT_MS,
+ "Kafka topic '" + topic + "' metadata is not visible to consumers");
+ }
+
+ private boolean isKafkaTopicReady(String topic, int expectedPartitions, int expectedReplicationFactor) {
+ if (_kafkaStarters == null || _kafkaStarters.isEmpty()) {
+ return false;
+ }
for (StreamDataServerStartable kafkaStarter : _kafkaStarters) {
- kafkaStarter.stop();
+ if (!isKafkaTopicReadyOnBroker("localhost:" + kafkaStarter.getPort(), topic, expectedPartitions,
+ expectedReplicationFactor)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private boolean isKafkaTopicMetadataReadyForConsumer(String topic, int expectedPartitions) {
+ Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaBrokerList());
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "pinot-kafka-topic-ready-" + UUID.randomUUID());
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ consumerProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps)) {
+ List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic, Duration.ofSeconds(5));
+ return partitionInfos != null && partitionInfos.size() >= expectedPartitions;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private boolean isKafkaTopicReadyOnBroker(String brokerList, String topic, int expectedPartitions,
+ int expectedReplicationFactor) {
+ Properties adminProps = new Properties();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+ adminProps.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ adminProps.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");
+ try (AdminClient adminClient = AdminClient.create(adminProps)) {
+ TopicDescription topicDescription =
+ adminClient.describeTopics(Collections.singletonList(topic)).allTopicNames().get(5, TimeUnit.SECONDS)
+ .get(topic);
+ if (topicDescription.partitions().size() < expectedPartitions) {
+ return false;
+ }
+ for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
+ if (partitionInfo.leader() == null) {
+ return false;
+ }
+ if (partitionInfo.isr() == null || partitionInfo.isr().size() < expectedReplicationFactor) {
+ return false;
+ }
+ }
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ protected void stopKafka() {
+ if (_kafkaStarters == null || _kafkaStarters.isEmpty()) {
+ return;
+ }
+ List<StreamDataServerStartable> kafkaStarters = _kafkaStarters;
+ _kafkaStarters = null;
+
+ RuntimeException stopException = null;
+ for (int i = kafkaStarters.size() - 1; i >= 0; i--) {
+ StreamDataServerStartable kafkaStarter = kafkaStarters.get(i);
+ try {
+ kafkaStarter.stop();
+ } catch (Exception e) {
+ RuntimeException wrapped = new RuntimeException(
+ "Failed to stop Kafka broker on port: " + kafkaStarter.getPort(), e);
+ if (stopException == null) {
+ stopException = wrapped;
+ } else {
+ stopException.addSuppressed(wrapped);
+ }
+ }
+ }
+
+ for (StreamDataServerStartable kafkaStarter : kafkaStarters) {
+ try {
+ waitForKafkaBrokerStopped(kafkaStarter.getPort());
+ } catch (RuntimeException e) {
+ if (stopException == null) {
+ stopException = e;
+ } else {
+ stopException.addSuppressed(e);
+ }
+ }
+ }
+
+ if (stopException != null) {
+ throw stopException;
+ }
+ }
+
+ private void waitForKafkaBrokerStopped(int brokerPort) {
+ String brokerList = "localhost:" + brokerPort;
+ TestUtils.waitForCondition(aVoid -> !isKafkaBrokerAvailable(brokerList), 200L, 60_000L,
+ "Kafka broker is still reachable after stop: " + brokerList);
+ }
+
+ private boolean isKafkaBrokerAvailable(String brokerList) {
+ Properties adminProps = new Properties();
+ adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+ adminProps.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "2000");
+ adminProps.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "2000");
+ try (AdminClient adminClient = AdminClient.create(adminProps)) {
+ adminClient.describeCluster().nodes().get(2, TimeUnit.SECONDS);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private static final class KafkaBrokerConfig {
+ private final int _brokerId;
+ private final int _port;
+ private final String _containerName;
+
+ private KafkaBrokerConfig(int brokerId, int port, String containerName) {
+ _brokerId = brokerId;
+ _port = port;
+ _containerName = containerName;
}
}
@@ -855,7 +1200,46 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
protected void waitForDocsLoaded(long timeoutMs, boolean raiseError, String tableName) {
long countStarResult = getCountStarResult();
TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) == countStarResult, 100L, timeoutMs,
- "Failed to load " + countStarResult + " documents", raiseError, Duration.ofMillis(timeoutMs / 10));
+ "Failed to load " + countStarResult + " documents", false, Duration.ofMillis(timeoutMs / 10));
+ if (raiseError) {
+ long currentCount = getCurrentCountStarResult(tableName);
+ Assert.assertEquals(currentCount, countStarResult,
+ "Failed to load " + countStarResult + " documents; current count=" + currentCount + " for table="
+ + tableName);
+ }
+ }
+
+ protected void waitForAllRealtimePartitionsConsuming(String tableNameWithType, long timeoutMs) {
+ int expectedPartitions = getNumKafkaPartitions();
+ TestUtils.waitForCondition(() -> hasConsumingSegmentsForAllPartitions(tableNameWithType, expectedPartitions), 200L,
+ timeoutMs, "Failed to get CONSUMING segments for all " + expectedPartitions + " partitions for table "
+ + tableNameWithType, false, Duration.ofMillis(timeoutMs / 10));
+ int consumingPartitions = getNumConsumingPartitions(tableNameWithType);
+ Assert.assertEquals(consumingPartitions, expectedPartitions, "Expected CONSUMING segments for "
+ + expectedPartitions + " partitions, found " + consumingPartitions + " for table " + tableNameWithType);
+ }
+
+ private boolean hasConsumingSegmentsForAllPartitions(String tableNameWithType, int expectedPartitions) {
+ return getNumConsumingPartitions(tableNameWithType) >= expectedPartitions;
+ }
+
+ private int getNumConsumingPartitions(String tableNameWithType) {
+ IdealState idealState = _controllerStarter.getHelixResourceManager().getTableIdealState(tableNameWithType);
+ if (idealState == null) {
+ return 0;
+ }
+ Set<Integer> consumingPartitions = new HashSet<>();
+ for (String segmentName : idealState.getPartitionSet()) {
+ if (!LLCSegmentName.isLLCSegment(segmentName)) {
+ continue;
+ }
+ Map<String, String> stateMap = idealState.getInstanceStateMap(segmentName);
+ if (stateMap != null
+ && stateMap.containsValue(CommonConstants.Helix.StateModel.SegmentStateModel.CONSUMING)) {
+ consumingPartitions.add(new LLCSegmentName(segmentName).getPartitionGroupId());
+ }
+ }
+ return consumingPartitions.size();
}
protected void waitForNonZeroDocsLoaded(long timeoutMs, boolean raiseError, String tableName) {
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
index acf389656b8..f460ece6f43 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterIntegrationTestUtils.java
@@ -44,6 +44,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -61,6 +62,7 @@ import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.pinot.client.ResultSetGroup;
import org.apache.pinot.common.request.PinotQuery;
@@ -526,9 +528,12 @@ public class ClusterIntegrationTestUtils {
properties.put("bootstrap.servers", kafkaBroker);
properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- properties.put("request.required.acks", "1");
- properties.put("transactional.id", "test-transaction");
- properties.put("transaction.state.log.replication.factor", "2");
+ properties.put(ProducerConfig.ACKS_CONFIG, "all");
+ properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ properties.put(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
+ properties.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "600000");
+ properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transaction-" + UUID.randomUUID());
try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(properties)) {
pushAvroIntoKafkaWithTransaction(avroFiles, kafkaTopic, maxNumKafkaMessagesPerBatch, header, partitionColumn,
commit, producer);
@@ -542,15 +547,23 @@ public class ClusterIntegrationTestUtils {
int maxNumKafkaMessagesPerBatch, @Nullable byte[] header, @Nullable String partitionColumn, boolean commit,
KafkaProducer<byte[], byte[]> producer)
throws Exception {
+ int maxMessagesPerTransaction = maxNumKafkaMessagesPerBatch > 0 ? maxNumKafkaMessagesPerBatch : Integer.MAX_VALUE;
producer.initTransactions();
- producer.beginTransaction();
long counter = 0;
+ int recordsInTransaction = 0;
+ boolean hasOpenTransaction = false;
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream(65536)) {
for (File avroFile : avroFiles) {
try (DataFileStream<GenericRecord> reader = AvroUtils.getAvroReader(avroFile)) {
BinaryEncoder binaryEncoder = new EncoderFactory().directBinaryEncoder(outputStream, null);
GenericDatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(reader.getSchema());
for (GenericRecord genericRecord : reader) {
+ if (!hasOpenTransaction) {
+ producer.beginTransaction();
+ hasOpenTransaction = true;
+ recordsInTransaction = 0;
+ }
+
outputStream.reset();
if (header != null && 0 < header.length) {
outputStream.write(header);
@@ -563,14 +576,26 @@ public class ClusterIntegrationTestUtils {
byte[] bytes = outputStream.toByteArray();
ProducerRecord<byte[], byte[]> record = new ProducerRecord(kafkaTopic, keyBytes, bytes);
producer.send(record);
+
+ recordsInTransaction++;
+ if (recordsInTransaction >= maxMessagesPerTransaction) {
+ if (commit) {
+ producer.commitTransaction();
+ } else {
+ producer.abortTransaction();
+ }
+ hasOpenTransaction = false;
+ }
}
}
}
}
- if (commit) {
- producer.commitTransaction();
- } else {
- producer.abortTransaction();
+ if (hasOpenTransaction) {
+ if (commit) {
+ producer.commitTransaction();
+ } else {
+ producer.abortTransaction();
+ }
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
index 29657838059..f7492846fc8 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
@@ -74,6 +74,8 @@ public abstract class BaseRealtimeClusterIntegrationTest extends BaseClusterInte
addSchema(schema);
TableConfig tableConfig = createRealtimeTableConfig(avroFiles.get(0));
addTableConfig(tableConfig);
+ waitForAllRealtimePartitionsConsuming(TableNameBuilder.REALTIME.tableNameWithType(getTableName()),
+ getRealtimePartitionsReadyTimeoutMs());
// Push data into Kafka
pushAvroIntoKafka(avroFiles);
@@ -90,7 +92,15 @@ public abstract class BaseRealtimeClusterIntegrationTest extends BaseClusterInte
runValidationJob(600_000);
// Wait for all documents loaded
- waitForAllDocsLoaded(600_000L);
+ waitForAllDocsLoaded(getDocsLoadedTimeoutMs());
+ }
+
+ protected long getDocsLoadedTimeoutMs() {
+ return useKafkaTransaction() ? 900_000L : 600_000L;
+ }
+
+ protected long getRealtimePartitionsReadyTimeoutMs() {
+ return useKafkaTransaction() ? 300_000L : 120_000L;
}
protected void runValidationJob(long timeoutMs)
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CommitTimeCompactionIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CommitTimeCompactionIntegrationTest.java
index eedd5602f63..a8f219a3819 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CommitTimeCompactionIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CommitTimeCompactionIntegrationTest.java
@@ -341,6 +341,7 @@ public class CommitTimeCompactionIntegrationTest extends BaseClusterIntegrationT
"201,Player201Updated,game2,95.0,1681154300000,false,strategy;rts");
String kafkaTopicName = getKafkaTopic() + "-multivalue";
+ createKafkaTopic(kafkaTopicName);
// TABLE 1: With commit-time compaction DISABLED (baseline)
String tableNameWithoutCompaction = "gameScoresMVCompactionDisabled";
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
index 21711d08ea8..6bed1cb3b47 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
@@ -29,16 +29,27 @@ public class ExactlyOnceKafkaRealtimeClusterIntegrationTest extends BaseRealtime
return true;
}
+ @Override
+ protected int getNumKafkaBrokers() {
+ return DEFAULT_TRANSACTION_NUM_KAFKA_BROKERS;
+ }
+
+ @Override
+ protected long getDocsLoadedTimeoutMs() {
+ return 1_200_000L;
+ }
+
@Override
protected void pushAvroIntoKafka(List<File> avroFiles)
throws Exception {
+ String kafkaBrokerList = getKafkaBrokerList();
// the first transaction of kafka messages are aborted
ClusterIntegrationTestUtils
- .pushAvroIntoKafkaWithTransaction(avroFiles, "localhost:" + getKafkaPort(), getKafkaTopic(),
+ .pushAvroIntoKafkaWithTransaction(avroFiles, kafkaBrokerList, getKafkaTopic(),
getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn(), false);
// the second transaction of kafka messages are committed
ClusterIntegrationTestUtils
- .pushAvroIntoKafkaWithTransaction(avroFiles, "localhost:" + getKafkaPort(), getKafkaTopic(),
+ .pushAvroIntoKafkaWithTransaction(avroFiles, kafkaBrokerList, getKafkaTopic(),
getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn(), true);
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
index 6a5d383110d..49838cf2148 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
@@ -46,8 +46,8 @@ import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.helix.core.controllerjob.ControllerJobTypes;
import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
-import org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory;
-import org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConsumer;
+import org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory;
+import org.apache.pinot.plugin.stream.kafka30.KafkaPartitionLevelConsumer;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.spi.config.table.IndexingConfig;
import org.apache.pinot.spi.config.table.TableConfig;
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
index e7b51c60648..9b6e6d9a263 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
@@ -57,7 +57,6 @@ import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-import org.apache.pinot.tools.utils.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -157,8 +156,7 @@ public class MergeRollupMinionClusterIntegrationTest extends BaseClusterIntegrat
// create the realtime table
TableConfig tableConfig = createRealtimeTableConfig(avroFiles.get(0));
addTableConfig(tableConfig);
- _kafkaStarters.get(0)
- .createTopic(PROCESS_ALL_MODE_KAFKA_TOPIC, KafkaStarterUtils.getTopicCreationProps(getNumKafkaPartitions()));
+ createKafkaTopic(PROCESS_ALL_MODE_KAFKA_TOPIC, getNumKafkaPartitions());
schema.setSchemaName(MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE);
addSchema(schema);
TableConfig singleLevelConcatProcessAllRealtimeTableConfig =
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java
index 72f7acdc9dd..cac229e8bec 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java
@@ -21,15 +21,22 @@ package org.apache.pinot.integration.tests.custom;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
import javax.annotation.Nullable;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.commons.io.FileUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.pinot.integration.tests.BaseClusterIntegrationTest;
import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils;
import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
@@ -50,6 +57,10 @@ import org.testng.annotations.BeforeSuite;
public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterIntegrationTest {
protected static final Logger LOGGER = LoggerFactory.getLogger(CustomDataQueryClusterIntegrationTest.class);
+ private static final Object SUITE_MUTATION_LOCK = new Object();
+ private static final int REALTIME_TABLE_CONFIG_RETRY_COUNT = 5;
+ private static final long REALTIME_TABLE_CONFIG_RETRY_WAIT_MS = 1_000L;
+ private static final long KAFKA_TOPIC_METADATA_READY_TIMEOUT_MS = 30_000L;
protected static CustomDataQueryClusterIntegrationTest _sharedClusterTestSuite = null;
protected static final String TIMESTAMP_FIELD_NAME = "ts";
@@ -89,53 +100,116 @@ public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterI
@BeforeClass
public void setUp()
throws Exception {
- LOGGER.warn("Setting up integration test class: {}", getClass().getSimpleName());
- if (_controllerRequestURLBuilder == null) {
- _controllerRequestURLBuilder =
- ControllerRequestURLBuilder.baseUrl("http://localhost:" + _sharedClusterTestSuite.getControllerPort());
- }
- TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
- // create & upload schema AND table config
- Schema schema = createSchema();
- addSchema(schema);
-
- List<File> avroFiles = createAvroFiles();
- if (isRealtimeTable()) {
- // create realtime table
- TableConfig tableConfig = createRealtimeTableConfig(avroFiles.get(0));
- addTableConfig(tableConfig);
-
- // Push data into Kafka
- pushAvroIntoKafka(avroFiles);
- } else {
- // create offline table
- TableConfig tableConfig = createOfflineTableConfig();
- addTableConfig(tableConfig);
-
- // create & upload segments
- int segmentIndex = 0;
- for (File avroFile : avroFiles) {
- ClusterIntegrationTestUtils.buildSegmentFromAvro(avroFile, tableConfig, schema, segmentIndex++, _segmentDir,
- _tarDir);
- uploadSegments(getTableName(), _tarDir);
+ synchronized (SUITE_MUTATION_LOCK) {
+ LOGGER.warn("Setting up integration test class: {}", getClass().getSimpleName());
+ if (_controllerRequestURLBuilder == null) {
+ _controllerRequestURLBuilder =
+ ControllerRequestURLBuilder.baseUrl("http://localhost:" + _sharedClusterTestSuite.getControllerPort());
+ }
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
+ // create & upload schema AND table config
+ Schema schema = createSchema();
+ addSchema(schema);
+
+ List<File> avroFiles = createAvroFiles();
+ if (isRealtimeTable()) {
+ // In suite mode multiple realtime tests use different topics, so make sure
+ // this class-specific topic exists before the controller validates stream metadata.
+ _sharedClusterTestSuite.createKafkaTopic(getKafkaTopic());
+ waitForKafkaTopicMetadataReadyForConsumer(getKafkaTopic(), getNumKafkaPartitions());
+
+ // create realtime table
+ TableConfig tableConfig = createRealtimeTableConfig(avroFiles.get(0));
+ addRealtimeTableConfigWithRetry(tableConfig);
+
+ // Push data into Kafka
+ pushAvroIntoKafka(avroFiles);
+ } else {
+ // create offline table
+ TableConfig tableConfig = createOfflineTableConfig();
+ addTableConfig(tableConfig);
+
+ // create & upload segments
+ int segmentIndex = 0;
+ for (File avroFile : avroFiles) {
+ ClusterIntegrationTestUtils.buildSegmentFromAvro(avroFile, tableConfig, schema, segmentIndex++, _segmentDir,
+ _tarDir);
+ uploadSegments(getTableName(), _tarDir);
+ }
}
- }
- waitForAllDocsLoaded(60_000);
- LOGGER.warn("Finished setting up integration test class: {}", getClass().getSimpleName());
+ waitForAllDocsLoaded(60_000);
+ LOGGER.warn("Finished setting up integration test class: {}", getClass().getSimpleName());
+ }
}
@AfterClass
public void tearDown()
throws IOException {
- LOGGER.warn("Tearing down integration test class: {}", getClass().getSimpleName());
- if (isRealtimeTable()) {
- dropRealtimeTable(getTableName());
- } else {
- dropOfflineTable(getTableName());
+ synchronized (SUITE_MUTATION_LOCK) {
+ LOGGER.warn("Tearing down integration test class: {}", getClass().getSimpleName());
+ if (isRealtimeTable()) {
+ dropRealtimeTable(getTableName());
+ } else {
+ dropOfflineTable(getTableName());
+ }
+ FileUtils.deleteDirectory(_tempDir);
+ LOGGER.warn("Finished tearing down integration test class: {}", getClass().getSimpleName());
+ }
+ }
+
+ private void addRealtimeTableConfigWithRetry(TableConfig tableConfig)
+ throws Exception {
+ for (int attempt = 1; attempt <= REALTIME_TABLE_CONFIG_RETRY_COUNT; attempt++) {
+ try {
+ addTableConfig(tableConfig);
+ return;
+ } catch (IOException e) {
+ if (!isRetryableRealtimePartitionMetadataError(e) || attempt == REALTIME_TABLE_CONFIG_RETRY_COUNT) {
+ throw e;
+ }
+ LOGGER.warn("Retrying realtime table creation for topic {} after metadata propagation failure (attempt {}/{})",
+ getKafkaTopic(), attempt, REALTIME_TABLE_CONFIG_RETRY_COUNT, e);
+ waitForKafkaTopicMetadataReadyForConsumer(getKafkaTopic(), getNumKafkaPartitions());
+ Thread.sleep(REALTIME_TABLE_CONFIG_RETRY_WAIT_MS);
+ }
+ }
+ throw new IllegalStateException("Failed to create realtime table after retries for topic: " + getKafkaTopic());
+ }
+
+ private boolean isRetryableRealtimePartitionMetadataError(Throwable throwable) {
+ String errorToken = "Failed to fetch partition information for topic: " + getKafkaTopic();
+ Throwable current = throwable;
+ while (current != null) {
+ String message = current.getMessage();
+ if (message != null && message.contains(errorToken)) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private void waitForKafkaTopicMetadataReadyForConsumer(String topic, int expectedPartitions) {
+ TestUtils.waitForCondition(aVoid -> isKafkaTopicMetadataReadyForConsumer(topic, expectedPartitions), 200L,
+ KAFKA_TOPIC_METADATA_READY_TIMEOUT_MS,
+ "Kafka topic '" + topic + "' metadata is not visible to consumers in custom cluster suite");
+ }
+
+ private boolean isKafkaTopicMetadataReadyForConsumer(String topic, int expectedPartitions) {
+ Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, _sharedClusterTestSuite.getKafkaBrokerList());
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "pinot-custom-topic-ready-" + UUID.randomUUID());
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+ consumerProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps)) {
+ List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic, Duration.ofSeconds(5));
+ return partitionInfos != null && partitionInfos.size() >= expectedPartitions;
+ } catch (Exception e) {
+ return false;
}
- FileUtils.deleteDirectory(_tempDir);
- LOGGER.warn("Finished tearing down integration test class: {}", getClass().getSimpleName());
}
@Override
@@ -148,7 +222,7 @@ public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterI
protected void pushAvroIntoKafka(List<File> avroFiles)
throws Exception {
ClusterIntegrationTestUtils.pushAvroIntoKafka(avroFiles,
- "localhost:" + _sharedClusterTestSuite._kafkaStarters.get(0).getPort(), getKafkaTopic(),
+ _sharedClusterTestSuite.getKafkaBrokerList(), getKafkaTopic(),
getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn(), injectTombstones());
}
@@ -234,7 +308,7 @@ public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterI
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BROKER_LIST),
- "localhost:" + _sharedClusterTestSuite._kafkaStarters.get(0).getPort());
+ _sharedClusterTestSuite.getKafkaBrokerList());
if (useKafkaTransaction()) {
streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL),
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
index 3b40a6c4175..021eaca9718 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
@@ -24,7 +24,6 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.JsonUtils;
-import org.apache.pinot.tools.utils.KafkaStarterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;
@@ -53,16 +52,16 @@ public class KafkaIncreaseDecreasePartitionsIntegrationTest extends BaseRealtime
throws Exception {
LOGGER.info("Starting testDecreasePartitions");
LOGGER.info("Creating Kafka topic with {} partitions", NUM_PARTITIONS + 2);
- _kafkaStarters.get(0).createTopic(KAFKA_TOPIC, KafkaStarterUtils.getTopicCreationProps(NUM_PARTITIONS + 2));
+ createKafkaTopic(KAFKA_TOPIC, NUM_PARTITIONS + 2);
String tableName = createTable();
waitForNumSegmentsInDesiredStateInEV(tableName, CONSUMING, NUM_PARTITIONS + 2, TableType.REALTIME);
pauseTable(tableName);
LOGGER.info("Deleting Kafka topic");
- _kafkaStarters.get(0).deleteTopic(KAFKA_TOPIC);
+ deleteKafkaTopic(KAFKA_TOPIC);
LOGGER.info("Creating Kafka topic with {} partitions", NUM_PARTITIONS);
- _kafkaStarters.get(0).createTopic(KAFKA_TOPIC, KafkaStarterUtils.getTopicCreationProps(NUM_PARTITIONS));
+ createKafkaTopic(KAFKA_TOPIC, NUM_PARTITIONS);
resumeTable(tableName);
waitForNumSegmentsInDesiredStateInEV(tableName, CONSUMING, NUM_PARTITIONS, TableType.REALTIME);
diff --git a/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json b/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
index 1900dd198d0..903ef5d4fd1 100644
--- a/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
+++ b/pinot-integration-tests/src/test/resources/simpleMeetup_realtime_table_config.json
@@ -21,7 +21,7 @@
"stream.kafka.broker.list": "localhost:19092",
"streamType": "kafka",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"realtime.segment.flush.threshold.size": 30,
"realtime.segment.flush.threshold.rows": 30
}
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index e5144010a42..cd18ccbf943 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -51,7 +51,7 @@
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-2.0</artifactId>
+ <artifactId>pinot-kafka-3.0</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
diff --git a/pinot-plugins/pinot-input-format/pinot-protobuf/pom.xml b/pinot-plugins/pinot-input-format/pinot-protobuf/pom.xml
index cd08d24f2ff..3128499f064 100644
--- a/pinot-plugins/pinot-input-format/pinot-protobuf/pom.xml
+++ b/pinot-plugins/pinot-input-format/pinot-protobuf/pom.xml
@@ -49,21 +49,6 @@
<artifactId>protobuf-dynamic</artifactId>
<version>${protobuf-dynamic.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers-kafka</artifactId>
- <scope>test</scope>
- </dependency>
</dependencies>
<build>
<plugins>
diff --git a/pinot-plugins/pinot-stream-ingestion/README.md b/pinot-plugins/pinot-stream-ingestion/README.md
index 7f8294e2214..e804d9f9fcd 100644
--- a/pinot-plugins/pinot-stream-ingestion/README.md
+++ b/pinot-plugins/pinot-stream-ingestion/README.md
@@ -20,4 +20,4 @@
-->
# Pinot connectors module
-The pinot-connectors module is the place to write any pinot connectors to streams. For example, the pinot-connector-kafka-2.0 sub module contains the stream implementation for kafka-2.0. Dependencies to be shared across all sub modules are to be added in the pinot-connectors/pom.xml, and dependencies specific to the specific connector should go in its own pom file
+The pinot-connectors module is the place to write any pinot connectors to streams. For example, the pinot-kafka-3.0 sub module contains the stream implementation for kafka-3.0. Dependencies to be shared across all sub modules are to be added in the pinot-connectors/pom.xml, and dependencies specific to the specific connector should go in its own pom file
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/README.md b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/README.md
deleted file mode 100644
index e32826700fa..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/README.md
+++ /dev/null
@@ -1,42 +0,0 @@
-<!--
-
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing,
- software distributed under the License is distributed on an
- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- KIND, either express or implied. See the License for the
- specific language governing permissions and limitations
- under the License.
-
--->
-# Pinot connector for kafka 2.x
-
-This is an implementation of the kafka stream for kafka versions 2.x.
-
-A stream plugin for another version of kafka, or another stream, can be added in a similar fashion. Refer to documentation on [Stream Ingestion Plugin](https://docs.pinot.apache.org/developers/plugin-architecture/write-custom-plugins/write-your-stream) for the specific interfaces to implement.
-
-* How to build and release Pinot package with Kafka 2.x connector
-```$xslt
-./mvnw clean package -DskipTests -Pbin-dist
-```
-
-* How to use Kafka 2.x connector
-Below is a sample `streamConfigs` used to create a real-time table with Kafka consumer:
-```$xslt
-"streamConfigs": {
- "streamType": "kafka",
- "stream.kafka.broker.list": "localhost:19092",
- "stream.kafka.topic.name": "meetupRSVPEvents",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
- "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
-}
-```
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/pom.xml b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/pom.xml
deleted file mode 100644
index b36b5e21aba..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/pom.xml
+++ /dev/null
@@ -1,79 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing,
- software distributed under the License is distributed on an
- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- KIND, either express or implied. See the License for the
- specific language governing permissions and limitations
- under the License.
-
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <artifactId>pinot-stream-ingestion</artifactId>
- <groupId>org.apache.pinot</groupId>
- <version>1.5.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>pinot-kafka-2.0</artifactId>
- <name>Pinot Kafka 2.x</name>
- <url>https://pinot.apache.org/</url>
- <properties>
- <pinot.root>${basedir}/../../..</pinot.root>
- <shade.phase.prop>package</shade.phase.prop>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-base</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_${scala.compat.version}</artifactId>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.module</groupId>
- <artifactId>jackson-module-scala_${scala.compat.version}</artifactId>
- </dependency>
-
- <!-- Test -->
- <dependency>
- <groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-base</artifactId>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
- <profiles>
- <profile>
- <id>pinot-fastdev</id>
- <properties>
- <shade.phase.prop>none</shade.phase.prop>
- </properties>
- </profile>
- </profiles>
-</project>
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
deleted file mode 100644
index 849103c0b17..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import org.apache.pinot.plugin.stream.kafka.KafkaConfigBackwardCompatibleUtils;
-import org.apache.pinot.spi.stream.PartitionGroupConsumer;
-import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamConsumerFactory;
-import org.apache.pinot.spi.stream.StreamMetadataProvider;
-import org.apache.pinot.spi.utils.retry.RetryPolicy;
-
-
-public class KafkaConsumerFactory extends StreamConsumerFactory {
-
- @Override
- protected void init(StreamConfig streamConfig) {
- KafkaConfigBackwardCompatibleUtils.handleStreamConfig(streamConfig);
- super.init(streamConfig);
- }
-
- @Override
- public StreamMetadataProvider createPartitionMetadataProvider(String clientId, int partition) {
- return new KafkaStreamMetadataProvider(clientId, _streamConfig, partition);
- }
-
- @Override
- public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
- return new KafkaStreamMetadataProvider(clientId, _streamConfig);
- }
-
- @Override
- public StreamMetadataProvider createStreamMetadataProvider(String clientId, boolean concurrentAccessExpected) {
- if (concurrentAccessExpected) {
- return new SynchronizedKafkaStreamMetadataProvider(clientId, _streamConfig);
- } else {
- return createStreamMetadataProvider(clientId);
- }
- }
-
- @Override
- public PartitionGroupConsumer createPartitionGroupConsumer(String clientId,
- PartitionGroupConsumptionStatus partitionGroupConsumptionStatus) {
- return new KafkaPartitionLevelConsumer(clientId, _streamConfig,
- partitionGroupConsumptionStatus.getStreamPartitionGroupId());
- }
-
- @Override
- public PartitionGroupConsumer createPartitionGroupConsumer(String clientId,
- PartitionGroupConsumptionStatus partitionGroupConsumptionStatus, RetryPolicy retryPolicy) {
- return new KafkaPartitionLevelConsumer(clientId, _streamConfig,
- partitionGroupConsumptionStatus.getStreamPartitionGroupId(), retryPolicy);
- }
-}
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
deleted file mode 100644
index 3958ff77689..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java
+++ /dev/null
@@ -1,234 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import com.google.common.annotations.VisibleForTesting;
-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.concurrent.atomic.AtomicReference;
-import java.util.function.Supplier;
-import javax.annotation.Nullable;
-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;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.BytesDeserializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.utils.Bytes;
-import org.apache.pinot.plugin.stream.kafka.KafkaAdminClientManager;
-import org.apache.pinot.plugin.stream.kafka.KafkaPartitionLevelStreamConfig;
-import org.apache.pinot.plugin.stream.kafka.KafkaSSLUtils;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
-import org.apache.pinot.spi.utils.retry.RetriableOperationException;
-import org.apache.pinot.spi.utils.retry.RetryPolicy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * KafkaPartitionLevelConnectionHandler provides low level APIs to access Kafka partition level information.
- * E.g. partition counts, offsets per partition.
- *
- */
-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;
- protected final String _topic;
- protected final Consumer<String, Bytes> _consumer;
- protected final TopicPartition _topicPartition;
- protected final Properties _consumerProp;
- protected volatile KafkaAdminClientManager.AdminClientReference _sharedAdminClientRef;
-
- public KafkaPartitionLevelConnectionHandler(String clientId, StreamConfig streamConfig, int partition) {
- this(clientId, streamConfig, partition, null);
- }
-
- public KafkaPartitionLevelConnectionHandler(String clientId, StreamConfig streamConfig, int partition,
- @Nullable RetryPolicy retryPolicy) {
- _config = new KafkaPartitionLevelStreamConfig(streamConfig);
- _clientId = clientId;
- _partition = partition;
- _topic = _config.getKafkaTopicName();
- _consumerProp = buildProperties(streamConfig);
- KafkaSSLUtils.initSSL(_consumerProp);
- if (retryPolicy == null) {
- _consumer = createConsumer(_consumerProp);
- } else {
- _consumer = createConsumer(_consumerProp, retryPolicy);
- }
- _topicPartition = new TopicPartition(_topic, _partition);
- _consumer.assign(Collections.singletonList(_topicPartition));
- }
-
- private Properties buildProperties(StreamConfig streamConfig) {
- Properties consumerProp = new Properties();
- consumerProp.putAll(streamConfig.getStreamConfigsMap());
- consumerProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, _config.getBootstrapHosts());
- consumerProp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
- consumerProp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class.getName());
- if (_config.getKafkaIsolationLevel() != null) {
- consumerProp.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, _config.getKafkaIsolationLevel());
- }
- consumerProp.put(ConsumerConfig.CLIENT_ID_CONFIG, _clientId);
- 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;
- }
-
- private Consumer<String, Bytes> createConsumer(Properties consumerProp, RetryPolicy retryPolicy) {
- AtomicReference<Consumer<String, Bytes>> consumer = new AtomicReference<>();
- try {
- retryPolicy.attempt(() -> {
- try {
- consumer.set(new KafkaConsumer<>(filterKafkaProperties(consumerProp, CONSUMER_CONFIG_NAMES)));
- return true;
- } catch (Exception e) {
- LOGGER.warn("Caught exception while creating Kafka consumer, retrying.", e);
- return false;
- }
- });
- } catch (AttemptsExceededException | RetriableOperationException e) {
- LOGGER.error("Caught exception while creating Kafka consumer, giving up", e);
- throw new RuntimeException(e);
- }
- return consumer.get();
- }
-
- @VisibleForTesting
- protected Consumer<String, Bytes> createConsumer(Properties consumerProp) {
- return retry(() -> new KafkaConsumer<>(filterKafkaProperties(consumerProp, CONSUMER_CONFIG_NAMES)), 5);
- }
-
- protected AdminClient createAdminClient() {
- return retry(() -> AdminClient.create(filterKafkaProperties(_consumerProp, ADMIN_CLIENT_CONFIG_NAMES)), 5);
- }
-
- /**
- * Gets or creates a reusable admin client instance. The admin client is lazily initialized
- * and reused across multiple calls to avoid the overhead of creating new connections.
- *
- * @return the admin client instance
- */
- @Deprecated
- protected AdminClient getOrCreateAdminClient() {
- return createAdminClient();
- }
-
- /**
- * Gets or creates a shared admin client instance that can be reused across multiple
- * connection handlers connecting to the same Kafka cluster. This provides better
- * resource efficiency when multiple consumers/producers connect to the same bootstrap servers.
- *
- * @return the shared admin client instance
- */
- protected AdminClient getOrCreateSharedAdminClient() {
- return getOrCreateSharedAdminClientInternal(false);
- }
-
- private AdminClient getOrCreateSharedAdminClientInternal(boolean isRetry) {
- KafkaAdminClientManager.AdminClientReference ref = _sharedAdminClientRef;
- if (ref == null) {
- synchronized (this) {
- ref = _sharedAdminClientRef;
- if (ref == null) {
- ref = KafkaAdminClientManager.getInstance().getOrCreateAdminClient(_consumerProp);
- _sharedAdminClientRef = ref;
- }
- }
- }
- try {
- return ref.getAdminClient();
- } catch (IllegalStateException e) {
- if (isRetry) {
- throw new RuntimeException("Failed to create admin client after retry", e);
- }
- // Reference was closed, retry once
- synchronized (this) {
- _sharedAdminClientRef = null;
- }
- return getOrCreateSharedAdminClientInternal(true);
- }
- }
-
- private static <T> T retry(Supplier<T> s, int nRetries) {
- // Creation of the KafkaConsumer can fail for multiple reasons including DNS issues.
- // We arbitrarily chose 5 retries with 2 seconds sleep in between retries. 10 seconds total felt
- // like a good balance of not waiting too long for a retry, but also not retrying too many times.
- int tries = 0;
- while (true) {
- try {
- return s.get();
- } catch (KafkaException e) {
- tries++;
- if (tries >= nRetries) {
- LOGGER.error("Caught exception while creating Kafka consumer, giving up", e);
- throw e;
- }
- LOGGER.warn("Caught exception while creating Kafka consumer, retrying {}/{}", tries, nRetries, e);
- // We are choosing to sleepUniterruptibly here because other parts of the Kafka consumer code do this
- // as well. We don't want random interrupts to cause us to fail to create the consumer and have the table
- // stuck in ERROR state.
- Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
- }
- }
- }
-
- public void close()
- throws IOException {
- _consumer.close();
- if (_sharedAdminClientRef != null) {
- _sharedAdminClientRef.close();
- _sharedAdminClientRef = null;
- }
- }
-
- @VisibleForTesting
- protected KafkaPartitionLevelStreamConfig getKafkaPartitionLevelStreamConfig() {
- return _config;
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java
deleted file mode 100644
index 54c581e20f4..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.utils.Bytes;
-import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
-import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
-import org.apache.pinot.plugin.stream.kafka.KafkaStreamMessageMetadata;
-import org.apache.pinot.spi.data.readers.GenericRow;
-import org.apache.pinot.spi.stream.BytesStreamMessage;
-import org.apache.pinot.spi.stream.LongMsgOffset;
-import org.apache.pinot.spi.stream.PartitionGroupConsumer;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamMessageMetadata;
-import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
-import org.apache.pinot.spi.utils.retry.RetryPolicy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler
- implements PartitionGroupConsumer {
- private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
-
- private long _lastFetchedOffset = -1;
-
- public KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) {
- super(clientId, streamConfig, partition);
- }
-
- public KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition,
- RetryPolicy retryPolicy) {
- super(clientId, streamConfig, partition, retryPolicy);
- }
-
- @Override
- public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, int timeoutMs) {
- long startOffset = ((LongMsgOffset) startMsgOffset).getOffset();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Polling partition: {}, startOffset: {}, timeout: {}ms", _topicPartition, startOffset, timeoutMs);
- }
- if (_lastFetchedOffset < 0 || _lastFetchedOffset != startOffset - 1) {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Seeking to offset: {}", startOffset);
- }
- _consumer.seek(_topicPartition, startOffset);
- }
-
- ConsumerRecords<String, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMs));
- List<ConsumerRecord<String, Bytes>> records = consumerRecords.records(_topicPartition);
- List<BytesStreamMessage> filteredRecords = new ArrayList<>(records.size());
- long firstOffset = -1;
- long offsetOfNextBatch = startOffset;
- StreamMessageMetadata lastMessageMetadata = null;
- long batchSizeInBytes = 0;
- if (!records.isEmpty()) {
- firstOffset = records.get(0).offset();
- _lastFetchedOffset = records.get(records.size() - 1).offset();
- offsetOfNextBatch = _lastFetchedOffset + 1;
- for (ConsumerRecord<String, Bytes> record : records) {
- StreamMessageMetadata messageMetadata = extractMessageMetadata(record);
- Bytes message = record.value();
- if (message != null) {
- String key = record.key();
- byte[] keyBytes = key != null ? key.getBytes(StandardCharsets.UTF_8) : null;
- filteredRecords.add(new BytesStreamMessage(keyBytes, message.get(), messageMetadata));
- if (messageMetadata.getRecordSerializedSize() > 0) {
- batchSizeInBytes += messageMetadata.getRecordSerializedSize();
- }
- } else if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Tombstone message at offset: {}", record.offset());
- }
- lastMessageMetadata = messageMetadata;
- }
- }
-
- // In case read_committed is enabled, the messages consumed are not guaranteed to have consecutive offsets.
- // TODO: A better solution would be to fetch earliest offset from topic and see if it is greater than startOffset.
- // However, this would require and additional call to Kafka which we want to avoid.
- boolean hasDataLoss = false;
- if (_config.getKafkaIsolationLevel() == null || _config.getKafkaIsolationLevel()
- .equals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL_READ_UNCOMMITTED)) {
- hasDataLoss = firstOffset > startOffset;
- }
- return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
- hasDataLoss, batchSizeInBytes);
- }
-
- private StreamMessageMetadata extractMessageMetadata(ConsumerRecord<String, Bytes> record) {
- long timestamp = record.timestamp();
- long offset = record.offset();
-
- StreamMessageMetadata.Builder builder = new StreamMessageMetadata.Builder().setRecordIngestionTimeMs(timestamp)
- .setOffset(new LongMsgOffset(offset), new LongMsgOffset(offset + 1))
- .setSerializedValueSize(record.serializedValueSize());
- if (_config.isPopulateMetadata()) {
- Headers headers = record.headers();
- if (headers != null) {
- GenericRow headerGenericRow = new GenericRow();
- for (Header header : headers.toArray()) {
- headerGenericRow.putValue(header.key(), header.value());
- }
- builder.setHeaders(headerGenericRow);
- }
- builder.setMetadata(Map.of(KafkaStreamMessageMetadata.RECORD_TIMESTAMP_KEY, String.valueOf(timestamp),
- KafkaStreamMessageMetadata.METADATA_OFFSET_KEY, String.valueOf(offset),
- KafkaStreamMessageMetadata.METADATA_PARTITION_KEY, String.valueOf(record.partition())));
- }
- return builder.build();
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
deleted file mode 100644
index add6485d7de..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
+++ /dev/null
@@ -1,285 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Sets;
-import java.io.IOException;
-import java.time.Clock;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.ListTopicsResult;
-import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.TimeoutException;
-import org.apache.pinot.plugin.stream.kafka.KafkaConsumerPartitionLag;
-import org.apache.pinot.spi.stream.ConsumerPartitionState;
-import org.apache.pinot.spi.stream.LongMsgOffset;
-import org.apache.pinot.spi.stream.OffsetCriteria;
-import org.apache.pinot.spi.stream.PartitionLagState;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamMessageMetadata;
-import org.apache.pinot.spi.stream.StreamMetadataProvider;
-import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
-import org.apache.pinot.spi.stream.TransientConsumerException;
-import org.apache.pinot.spi.utils.TimeUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHandler
- implements StreamMetadataProvider {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
-
- public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
- this(clientId, streamConfig, Integer.MIN_VALUE);
- }
-
- public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
- super(clientId, streamConfig, partition);
- }
-
- @Override
- public int fetchPartitionCount(long timeoutMillis) {
- try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
- }
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
- } catch (TimeoutException e) {
- throw new TransientConsumerException(e);
- }
- }
-
- @Override
- public Set<Integer> fetchPartitionIds(long timeoutMillis) {
- try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
- }
- Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
- for (PartitionInfo partitionInfo : partitionInfos) {
- partitionIds.add(partitionInfo.partition());
- }
- return partitionIds;
- } catch (TimeoutException e) {
- throw new TransientConsumerException(e);
- }
- }
-
- @Override
- public Map<Integer, StreamPartitionMsgOffset> fetchLatestStreamOffset(Set<Integer> partitionIds, long timeoutMillis) {
- List<TopicPartition> topicPartitions = new ArrayList<>(partitionIds.size());
- for (Integer streamPartition: partitionIds) {
- topicPartitions.add(new TopicPartition(_topic, streamPartition));
- }
- try {
- Map<TopicPartition, Long> topicPartitionToLatestOffsetMap =
- _consumer.endOffsets(topicPartitions, Duration.ofMillis(timeoutMillis));
-
- Map<Integer, StreamPartitionMsgOffset> partitionIdToLatestOffset =
- new HashMap<>(topicPartitionToLatestOffsetMap.size());
- for (Map.Entry<TopicPartition, Long> entry : topicPartitionToLatestOffsetMap.entrySet()) {
- partitionIdToLatestOffset.put(entry.getKey().partition(), new LongMsgOffset(entry.getValue()));
- }
-
- return partitionIdToLatestOffset;
- } catch (TimeoutException e) {
- throw new TransientConsumerException(e);
- }
- }
-
- @Override
- public StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offsetCriteria, long timeoutMillis) {
- Preconditions.checkNotNull(offsetCriteria);
- long offset;
- try {
- if (offsetCriteria.isLargest()) {
- offset = _consumer.endOffsets(Collections.singletonList(_topicPartition), Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- } else if (offsetCriteria.isSmallest()) {
- offset =
- _consumer.beginningOffsets(Collections.singletonList(_topicPartition), Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- } else if (offsetCriteria.isPeriod()) {
- OffsetAndTimestamp offsetAndTimestamp = _consumer.offsetsForTimes(Collections.singletonMap(_topicPartition,
- Clock.systemUTC().millis() - TimeUtils.convertPeriodToMillis(offsetCriteria.getOffsetString())))
- .get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(Collections.singletonList(_topicPartition), Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is period and its value evaluates to null hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(), _topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
- }
- } else if (offsetCriteria.isTimestamp()) {
- OffsetAndTimestamp offsetAndTimestamp = _consumer.offsetsForTimes(Collections.singletonMap(_topicPartition,
- TimeUtils.convertTimestampToMillis(offsetCriteria.getOffsetString()))).get(_topicPartition);
- if (offsetAndTimestamp == null) {
- offset = _consumer.endOffsets(Collections.singletonList(_topicPartition), Duration.ofMillis(timeoutMillis))
- .get(_topicPartition);
- LOGGER.warn(
- "initial offset type is timestamp and its value evaluates to null hence proceeding with offset {} for "
- + "topic {} partition {}", offset, _topicPartition.topic(), _topicPartition.partition());
- } else {
- offset = offsetAndTimestamp.offset();
- }
- } else {
- throw new IllegalArgumentException("Unknown initial offset value " + offsetCriteria);
- }
- return new LongMsgOffset(offset);
- } catch (TimeoutException e) {
- throw new TransientConsumerException(e);
- }
- }
-
- @Override
- public Map<String, PartitionLagState> getCurrentPartitionLagState(
- Map<String, ConsumerPartitionState> currentPartitionStateMap) {
- Map<String, PartitionLagState> perPartitionLag = new HashMap<>();
- for (Map.Entry<String, ConsumerPartitionState> entry : currentPartitionStateMap.entrySet()) {
- ConsumerPartitionState partitionState = entry.getValue();
- // Compute records-lag
- StreamPartitionMsgOffset currentOffset = partitionState.getCurrentOffset();
- StreamPartitionMsgOffset upstreamLatest = partitionState.getUpstreamLatestOffset();
- String offsetLagString = "UNKNOWN";
-
- if (currentOffset instanceof LongMsgOffset && upstreamLatest instanceof LongMsgOffset) {
- long offsetLag = ((LongMsgOffset) upstreamLatest).getOffset() - ((LongMsgOffset) currentOffset).getOffset();
- offsetLagString = String.valueOf(offsetLag);
- }
-
- // Compute record-availability
- String availabilityLagMs = "UNKNOWN";
- StreamMessageMetadata lastProcessedMessageMetadata = partitionState.getLastProcessedRowMetadata();
- if (lastProcessedMessageMetadata != null && partitionState.getLastProcessedTimeMs() > 0) {
- long availabilityLag = partitionState.getLastProcessedTimeMs()
- - lastProcessedMessageMetadata.getRecordIngestionTimeMs();
- availabilityLagMs = String.valueOf(availabilityLag);
- }
-
- perPartitionLag.put(entry.getKey(), new KafkaConsumerPartitionLag(offsetLagString, availabilityLagMs));
- }
- return perPartitionLag;
- }
-
- @Override
- public List<TopicMetadata> getTopics() {
- try {
- AdminClient adminClient = getOrCreateSharedAdminClient();
- ListTopicsResult result = adminClient.listTopics();
- if (result == null) {
- return Collections.emptyList();
- }
- return result.names()
- .get()
- .stream()
- .map(topic -> new KafkaTopicMetadata().setName(topic))
- .collect(Collectors.toList());
- } catch (ExecutionException | InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public boolean supportsOffsetLag() {
- return true;
- }
-
- @Override
- public StreamPartitionMsgOffset getOffsetAtTimestamp(int partitionId, long timestampMillis, long timeoutMillis) {
- try {
- OffsetAndTimestamp offsetAndTimestamp = _consumer.offsetsForTimes(Map.of(_topicPartition, timestampMillis),
- Duration.ofMillis(timeoutMillis)).get(_topicPartition);
- if (offsetAndTimestamp == null) {
- return null;
- }
- return new LongMsgOffset(offsetAndTimestamp.offset());
- } catch (Exception e) {
- LOGGER.warn("Failed to get offset at timestamp {} for partition {}", timestampMillis, partitionId, e);
- return null;
- }
- }
-
- @Override
- public Map<String, StreamPartitionMsgOffset> getStreamStartOffsets() {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic);
- Map<TopicPartition, Long> startOffsets = _consumer.beginningOffsets(
- partitionInfos.stream()
- .filter(info -> info != null)
- .map(info -> new TopicPartition(_topic, info.partition()))
- .collect(Collectors.toList()));
- return startOffsets.entrySet().stream().collect(
- Collectors.toMap(
- entry -> String.valueOf(entry.getKey().partition()),
- entry -> new LongMsgOffset(entry.getValue()),
- (existingValue, newValue) -> newValue
- ));
- }
-
- @Override
- public Map<String, StreamPartitionMsgOffset> getStreamEndOffsets() {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic);
- Map<TopicPartition, Long> startOffsets = _consumer.endOffsets(
- partitionInfos.stream()
- .filter(info -> info != null)
- .map(info -> new TopicPartition(_topic, info.partition()))
- .collect(Collectors.toList()));
- return startOffsets.entrySet().stream().collect(
- Collectors.toMap(
- entry -> String.valueOf(entry.getKey().partition()),
- entry -> new LongMsgOffset(entry.getValue()),
- (existingValue, newValue) -> newValue
- ));
- }
-
- public static class KafkaTopicMetadata implements TopicMetadata {
- private String _name;
-
- public String getName() {
- return _name;
- }
-
- public KafkaTopicMetadata setName(String name) {
- _name = name;
- return this;
- }
- }
-
- @Override
- public void close()
- throws IOException {
- super.close();
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/SynchronizedKafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/SynchronizedKafkaStreamMetadataProvider.java
deleted file mode 100644
index 5549dd734fc..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/SynchronizedKafkaStreamMetadataProvider.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.util.Map;
-import java.util.Set;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
-
-/**
- * A thread-safe variant of {@link KafkaStreamMetadataProvider} that synchronizes
- * access to metadata fetch operations.
- * This is particularly useful when a shared instance of {@link KafkaStreamMetadataProvider}
- * is accessed concurrently by multiple threads.
- */
-public class SynchronizedKafkaStreamMetadataProvider extends KafkaStreamMetadataProvider {
-
- public SynchronizedKafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
- super(clientId, streamConfig);
- }
-
- @Override
- public Map<Integer, StreamPartitionMsgOffset> fetchLatestStreamOffset(Set<Integer> partitionIds, long timeoutMillis) {
- synchronized (this) {
- return super.fetchLatestStreamOffset(partitionIds, timeoutMillis);
- }
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataServerStartable.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataServerStartable.java
deleted file mode 100644
index 15e26b7f3fc..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataServerStartable.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20.server;
-
-import com.google.common.base.Function;
-import java.io.File;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import javax.annotation.Nullable;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import org.I0Itec.zkclient.ZkClient;
-import org.apache.commons.io.FileUtils;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.admin.KafkaAdminClient;
-import org.apache.kafka.clients.admin.NewPartitions;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.common.utils.Time;
-import org.apache.pinot.spi.stream.StreamDataServerStartable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.Option;
-
-
-public class KafkaDataServerStartable implements StreamDataServerStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(KafkaDataServerStartable.class);
-
- private static final String ZOOKEEPER_CONNECT = "zookeeper.connect";
- private static final String LOG_DIRS = "log.dirs";
- private static final String PORT = "port";
-
- private KafkaServer _serverStartable;
- private int _port;
- private String _zkStr;
- private String _logDirPath;
- private AdminClient _adminClient;
-
- public void init(Properties props) {
- _port = (int) props.get(PORT);
- _zkStr = props.getProperty(ZOOKEEPER_CONNECT);
- _logDirPath = props.getProperty(LOG_DIRS);
-
- // Create the ZK nodes for Kafka, if needed
- int indexOfFirstSlash = _zkStr.indexOf('/');
- if (indexOfFirstSlash != -1) {
- String bareZkUrl = _zkStr.substring(0, indexOfFirstSlash);
- String zkNodePath = _zkStr.substring(indexOfFirstSlash);
- ZkClient client = new ZkClient(bareZkUrl);
- client.createPersistent(zkNodePath, true);
- client.close();
- }
-
- File logDir = new File(_logDirPath);
- logDir.mkdirs();
-
- props.put("zookeeper.session.timeout.ms", "60000");
- _serverStartable = new KafkaServer(new KafkaConfig(props), Time.SYSTEM, Option.empty(), false);
- final Map<String, Object> config = new HashMap<>();
- config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + _port);
- config.put(AdminClientConfig.CLIENT_ID_CONFIG, "Kafka2AdminClient-" + UUID.randomUUID().toString());
- config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 15000);
- _adminClient = KafkaAdminClient.create(config);
- }
-
- @Override
- public void start() {
- _serverStartable.startup();
- }
-
- @Override
- public void stop() {
- _serverStartable.shutdown();
- FileUtils.deleteQuietly(new File(_serverStartable.config().logDirs().apply(0)));
- }
-
- @Override
- public void createTopic(String topic, Properties props) {
- int partition = (Integer) props.get("partition");
- Collection<NewTopic> topicList = Arrays.asList(new NewTopic(topic, partition, (short) 1));
- _adminClient.createTopics(topicList);
- waitForCondition(new Function<Void, Boolean>() {
- @Nullable
- @Override
- public Boolean apply(@Nullable Void aVoid) {
- try {
- return _adminClient.listTopics().names().get().contains(topic);
- } catch (Exception e) {
- LOGGER.warn("Could not fetch Kafka topics", e);
- return null;
- }
- }
- }, 1000L, 300000, "Kafka topic " + topic + " is not created yet");
- }
-
- @Override
- public void deleteTopic(String topic) {
- try {
- _adminClient.deleteTopics(Collections.singletonList(topic)).all().get();
- } catch (Exception e) {
- LOGGER.error("Could not delete topic: {}", topic);
- }
-
- waitForCondition(new Function<Void, Boolean>() {
- @Nullable
- @Override
- public Boolean apply(@Nullable Void aVoid) {
- try {
- return !_adminClient.listTopics().names().get().contains(topic);
- } catch (Exception e) {
- LOGGER.warn("Could not fetch Kafka topics", e);
- return null;
- }
- }
- }, 1000L, 300000, "Kafka topic " + topic + " is not deleted yet");
- }
-
- @Override
- public void createPartitions(String topic, int numPartitions) {
- try {
- Map<String, NewPartitions> newPartitionMap = new HashMap<>();
- newPartitionMap.put(topic, NewPartitions.increaseTo(numPartitions));
- _adminClient.createPartitions(newPartitionMap).all().get();
- } catch (Exception e) {
- LOGGER.error("Failed to create partitions for topic: {}", topic, e);
- }
- }
-
- @Override
- public int getPort() {
- return _port;
- }
-
- private static void waitForCondition(Function<Void, Boolean> condition, long checkIntervalMs, long timeoutMs,
- @Nullable String errorMessage) {
- long endTime = System.currentTimeMillis() + timeoutMs;
- String errorMessageSuffix = errorMessage != null ? ", error message: " + errorMessage : "";
- while (System.currentTimeMillis() < endTime) {
- try {
- if (Boolean.TRUE.equals(condition.apply(null))) {
- return;
- }
- Thread.sleep(checkIntervalMs);
- } catch (Exception e) {
- LOGGER.error("Caught exception while checking the condition{}", errorMessageSuffix, e);
- }
- }
- LOGGER.error("Failed to meet condition in {}ms{}", timeoutMs, errorMessageSuffix);
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaAdminClientManagerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaAdminClientManagerTest.java
deleted file mode 100644
index 3e40a374f8b..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaAdminClientManagerTest.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.util.Properties;
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.pinot.plugin.stream.kafka.KafkaAdminClientManager;
-import org.testng.annotations.Test;
-
-import static org.testng.Assert.*;
-
-
-public class KafkaAdminClientManagerTest {
-
- @Test
- public void testGetInstance() {
- KafkaAdminClientManager instance1 = KafkaAdminClientManager.getInstance();
- KafkaAdminClientManager instance2 = KafkaAdminClientManager.getInstance();
- assertSame(instance1, instance2, "Should return the same singleton instance");
- }
-
- @Test
- public void testCreateCacheKey() {
- Properties props1 = new Properties();
- props1.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
- props1.setProperty(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
-
- Properties props2 = new Properties();
- props2.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
- props2.setProperty(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
-
- Properties props3 = new Properties();
- props3.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093");
- props3.setProperty(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
-
- KafkaAdminClientManager manager = KafkaAdminClientManager.getInstance();
-
- // Use reflection to access private method for testing
- try {
- java.lang.reflect.Method createCacheKeyMethod =
- KafkaAdminClientManager.class.getDeclaredMethod("createCacheKey", Properties.class);
- createCacheKeyMethod.setAccessible(true);
-
- String key1 = (String) createCacheKeyMethod.invoke(manager, props1);
- String key2 = (String) createCacheKeyMethod.invoke(manager, props2);
- String key3 = (String) createCacheKeyMethod.invoke(manager, props3);
-
- assertEquals(key1, key2, "Same properties should produce same cache key");
- assertNotEquals(key1, key3, "Different bootstrap servers should produce different cache keys");
- } catch (Exception e) {
- fail("Failed to test createCacheKey method", e);
- }
- }
-
- @Test(expectedExceptions = IllegalArgumentException.class)
- public void testGetOrCreateAdminClientWithoutBootstrapServers() {
- Properties props = new Properties();
- // No bootstrap servers set
- KafkaAdminClientManager.getInstance().getOrCreateAdminClient(props);
- }
-
- @Test(expectedExceptions = IllegalArgumentException.class)
- public void testGetOrCreateAdminClientWithEmptyBootstrapServers() {
- Properties props = new Properties();
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "");
- KafkaAdminClientManager.getInstance().getOrCreateAdminClient(props);
- }
-
- @Test
- public void testAdminClientReferenceClosing() {
- Properties props = new Properties();
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
-
- KafkaAdminClientManager manager = KafkaAdminClientManager.getInstance();
-
- // This test focuses on the reference mechanism without actually connecting to Kafka
- // We test that multiple calls with same properties don't throw exceptions
- try {
- KafkaAdminClientManager.AdminClientReference ref1 = manager.getOrCreateAdminClient(props);
- assertNotNull(ref1, "Reference should not be null");
-
- // Closing a reference should not throw an exception
- ref1.close();
-
- // Closing again should not throw an exception
- ref1.close();
- } catch (Exception e) {
- // This is expected since we're not connecting to a real Kafka cluster
- // The test is mainly to verify the reference counting mechanism doesn't crash
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused"));
- }
- }
-
- @Test
- public void testAdminClientReferenceStateAfterClose() {
- Properties props = new Properties();
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
-
- KafkaAdminClientManager manager = KafkaAdminClientManager.getInstance();
-
- try {
- KafkaAdminClientManager.AdminClientReference ref = manager.getOrCreateAdminClient(props);
- ref.close();
-
- // Should throw IllegalStateException when trying to use closed reference
- assertThrows(IllegalStateException.class, () -> ref.getAdminClient());
- } catch (Exception e) {
- // Expected when no real Kafka cluster is available
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused"));
- }
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandlerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandlerTest.java
deleted file mode 100644
index 87d66557358..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConnectionHandlerTest.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.testng.annotations.Test;
-
-import static org.testng.Assert.assertTrue;
-
-
-public class KafkaPartitionLevelConnectionHandlerTest {
-
- private static class TestableKafkaPartitionLevelConnectionHandler extends KafkaPartitionLevelConnectionHandler {
- public TestableKafkaPartitionLevelConnectionHandler(String clientId, StreamConfig streamConfig, int partition) {
- super(clientId, streamConfig, partition);
- }
- }
-
- private StreamConfig createTestStreamConfig() {
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", "kafka");
- streamConfigMap.put("stream.kafka.topic.name", "testTopic");
- streamConfigMap.put("stream.kafka.broker.list", "localhost:9092");
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- return new StreamConfig("testTable_REALTIME", streamConfigMap);
- }
-
- @Test
- public void testSharedAdminClientReference() {
- StreamConfig streamConfig = createTestStreamConfig();
-
- try {
- TestableKafkaPartitionLevelConnectionHandler handler =
- new TestableKafkaPartitionLevelConnectionHandler("testClient", streamConfig, 0);
-
- // Test that we can call getOrCreateSharedAdminClient multiple times
- // without throwing exceptions (even though it may fail to connect)
- try {
- handler.getOrCreateSharedAdminClient();
- handler.getOrCreateSharedAdminClient(); // Should reuse the same reference
- } catch (Exception e) {
- // Expected when no real Kafka cluster is available
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused")
- || e.getCause() != null);
- }
-
- // Test that close doesn't throw exceptions
- handler.close();
- } catch (Exception e) {
- // Expected when initializing without a real Kafka cluster
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused")
- || e.getCause() != null);
- }
- }
-
- @Test
- public void testGetOrCreateAdminClientBackwardCompatibility() {
- StreamConfig streamConfig = createTestStreamConfig();
-
- try {
- TestableKafkaPartitionLevelConnectionHandler handler =
- new TestableKafkaPartitionLevelConnectionHandler("testClient", streamConfig, 0);
-
- // Test that the backward compatibility method still works
- try {
- handler.getOrCreateAdminClient();
- } catch (Exception e) {
- // Expected when no real Kafka cluster is available
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused")
- || e.getCause() != null);
- }
-
- handler.close();
- } catch (Exception e) {
- // Expected when initializing without a real Kafka cluster
- assertTrue(e.getMessage().contains("Connection") || e.getMessage().contains("Kafka")
- || e.getMessage().contains("timeout") || e.getMessage().contains("refused")
- || e.getCause() != null);
- }
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerBackwardCompatibilityTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerBackwardCompatibilityTest.java
deleted file mode 100644
index 25eec577a02..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerBackwardCompatibilityTest.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-/**
- * Tests for the KafkaPartitionLevelConsumer with old kafka consumer factory name.
- */
-public class KafkaPartitionLevelConsumerBackwardCompatibilityTest extends KafkaPartitionLevelConsumerTest {
-
- @Override
- protected String getKafkaConsumerFactoryName() {
- return "org.apache.pinot.core.realtime.impl.kafka2.KafkaConsumerFactory";
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerTest.java
deleted file mode 100644
index af36244c9e5..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerTest.java
+++ /dev/null
@@ -1,509 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.TimeoutException;
-import java.util.stream.Collectors;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.record.TimestampType;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.common.utils.Bytes;
-import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
-import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
-import org.apache.pinot.plugin.stream.kafka20.utils.MiniKafkaCluster;
-import org.apache.pinot.spi.stream.LongMsgOffset;
-import org.apache.pinot.spi.stream.MessageBatch;
-import org.apache.pinot.spi.stream.OffsetCriteria;
-import org.apache.pinot.spi.stream.PartitionGroupConsumer;
-import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamConsumerFactory;
-import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
-import org.apache.pinot.spi.stream.StreamMessage;
-import org.apache.pinot.spi.stream.StreamMessageMetadata;
-import org.apache.pinot.spi.stream.StreamMetadataProvider;
-import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
-import org.apache.pinot.spi.utils.retry.ExponentialBackoffRetryPolicy;
-import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.Test;
-
-import static org.apache.kafka.clients.consumer.ConsumerRecord.NULL_CHECKSUM;
-import static org.apache.kafka.common.record.LegacyRecord.NO_TIMESTAMP;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-
-
-/**
- * Tests for the KafkaPartitionLevelConsumer.
- */
-public class KafkaPartitionLevelConsumerTest {
- private static final long STABILIZE_SLEEP_DELAYS = 3000;
- private static final String TEST_TOPIC_1 = "foo";
- private static final String TEST_TOPIC_2 = "bar";
- private static final String TEST_TOPIC_3 = "expired";
- private static final int NUM_MSG_PRODUCED_PER_PARTITION = 1000;
- private static final long TIMESTAMP = Instant.now().toEpochMilli();
- private static final Random RANDOM = new Random();
-
- private MiniKafkaCluster _kafkaCluster;
- private String _kafkaBrokerAddress;
-
- @BeforeClass
- public void setUp()
- throws Exception {
- _kafkaCluster = new MiniKafkaCluster("0");
- _kafkaCluster.start();
- _kafkaBrokerAddress = _kafkaCluster.getKafkaServerAddress();
- _kafkaCluster.createTopic(TEST_TOPIC_1, 1, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_2, 2, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_3, 1, 1);
- Thread.sleep(STABILIZE_SLEEP_DELAYS);
- produceMsgToKafka();
- Thread.sleep(STABILIZE_SLEEP_DELAYS);
- _kafkaCluster.deleteRecordsBeforeOffset(TEST_TOPIC_3, 0, 200);
- }
-
- private void produceMsgToKafka() {
- Properties props = new Properties();
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _kafkaBrokerAddress);
- props.put(ProducerConfig.CLIENT_ID_CONFIG, "clientId");
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
- try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
- for (int i = 0; i < NUM_MSG_PRODUCED_PER_PARTITION; i++) {
- producer.send(new ProducerRecord<>(TEST_TOPIC_1, 0, TIMESTAMP + i, null, "sample_msg_" + i));
- // TEST_TOPIC_2 has 2 partitions
- producer.send(new ProducerRecord<>(TEST_TOPIC_2, 0, TIMESTAMP + i, null, "sample_msg_" + i));
- producer.send(new ProducerRecord<>(TEST_TOPIC_2, 1, TIMESTAMP + i, null, "sample_msg_" + i));
- producer.send(new ProducerRecord<>(TEST_TOPIC_3, "sample_msg_" + i));
- }
- producer.flush();
- }
- }
-
- @AfterClass
- public void tearDown()
- throws Exception {
- try {
- _kafkaCluster.deleteTopic(TEST_TOPIC_1);
- _kafkaCluster.deleteTopic(TEST_TOPIC_2);
- _kafkaCluster.deleteTopic(TEST_TOPIC_3);
- } finally {
- _kafkaCluster.close();
- }
- }
-
- @Test
- public void testBuildConsumer() {
- String streamType = "kafka";
- String streamKafkaTopicName = "theTopic";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", streamKafkaTopicName);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- streamConfigMap.put("stream.kafka.fetcher.size", "10000");
- streamConfigMap.put("stream.kafka.fetcher.minBytes", "20000");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- // test default value
- KafkaPartitionLevelConsumer kafkaSimpleStreamConsumer = createConsumer(clientId, streamConfig, 0);
- kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), 10000);
-
- assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
- kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaBufferSize());
- assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
- kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaSocketTimeout());
-
- // test parsing values
- assertEquals(10000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaFetcherSizeBytes());
- assertEquals(20000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaFetcherMinBytes());
-
- // test user defined values
- streamConfigMap.put("stream.kafka.buffer.size", "100");
- streamConfigMap.put("stream.kafka.socket.timeout", "1000");
- streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
- kafkaSimpleStreamConsumer = createConsumer(clientId, streamConfig, 0);
- kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), 10000);
- assertEquals(100, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaBufferSize());
- assertEquals(1000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaSocketTimeout());
- }
-
- @Test
- public void testGetPartitionCount() {
- String streamType = "kafka";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_1);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
- assertEquals(streamMetadataProvider.fetchPartitionCount(10000L), 1);
-
- streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_2);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
- assertEquals(streamMetadataProvider.fetchPartitionCount(10000L), 2);
- }
-
- @Test
- public void testFetchMessages() {
- String streamType = "kafka";
- String streamKafkaTopicName = "theTopic";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", streamKafkaTopicName);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- int partition = 0;
- KafkaPartitionLevelConsumer kafkaSimpleStreamConsumer = createConsumer(clientId, streamConfig, partition);
- kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), 10000);
- }
-
- @Test
- public void testFetchOffsets() {
- testFetchOffsets(TEST_TOPIC_1);
- testFetchOffsets(TEST_TOPIC_2);
- }
-
- private void testFetchOffsets(String topic) {
- String streamType = "kafka";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", topic);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- int numPartitions = new KafkaStreamMetadataProvider(clientId, streamConfig).fetchPartitionCount(10000);
- for (int partition = 0; partition < numPartitions; partition++) {
- KafkaStreamMetadataProvider kafkaStreamMetadataProvider =
- new KafkaStreamMetadataProvider(clientId, streamConfig, partition);
- assertEquals(new LongMsgOffset(0).compareTo(kafkaStreamMetadataProvider.fetchStreamPartitionOffset(
- new OffsetCriteria.OffsetCriteriaBuilder().withOffsetSmallest(), 10000)), 0);
- assertEquals(new LongMsgOffset(0).compareTo(kafkaStreamMetadataProvider.fetchStreamPartitionOffset(
- new OffsetCriteria.OffsetCriteriaBuilder().withOffsetAsPeriod("2d"), 10000)), 0);
- assertEquals(new LongMsgOffset(NUM_MSG_PRODUCED_PER_PARTITION).compareTo(
- kafkaStreamMetadataProvider.fetchStreamPartitionOffset(
- new OffsetCriteria.OffsetCriteriaBuilder().withOffsetAsTimestamp(Instant.now().toString()), 10000)), 0);
- assertEquals(new LongMsgOffset(NUM_MSG_PRODUCED_PER_PARTITION).compareTo(
- kafkaStreamMetadataProvider.fetchStreamPartitionOffset(
- new OffsetCriteria.OffsetCriteriaBuilder().withOffsetLargest(), 10000)), 0);
- }
- }
-
- @Test
- public void testConsumer()
- throws Exception {
- testConsumer(TEST_TOPIC_1);
- testConsumer(TEST_TOPIC_2);
- }
-
- private void testConsumer(String topic)
- throws TimeoutException {
- String streamType = "kafka";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", topic);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- int numPartitions = new KafkaStreamMetadataProvider(clientId, streamConfig).fetchPartitionCount(10000);
- for (int partition = 0; partition < numPartitions; partition++) {
- PartitionGroupConsumer consumer = streamConsumerFactory.createPartitionGroupConsumer(clientId,
- new PartitionGroupConsumptionStatus(partition, 0, new LongMsgOffset(0),
- new LongMsgOffset(NUM_MSG_PRODUCED_PER_PARTITION), "CONSUMING"));
-
- // Test consume a large batch, only 500 records will be returned.
- MessageBatch messageBatch = consumer.fetchMessages(new LongMsgOffset(0), 10000);
- assertEquals(messageBatch.getMessageCount(), 500);
- assertEquals(messageBatch.getUnfilteredMessageCount(), 500);
- for (int i = 0; i < 500; i++) {
- StreamMessage streamMessage = messageBatch.getStreamMessage(i);
- assertEquals(new String((byte[]) streamMessage.getValue()), "sample_msg_" + i);
- StreamMessageMetadata metadata = streamMessage.getMetadata();
- assertEquals(metadata.getRecordIngestionTimeMs(), TIMESTAMP + i);
- StreamPartitionMsgOffset offset = metadata.getOffset();
- assertTrue(offset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) offset).getOffset(), i);
- StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
- assertTrue(nextOffset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) nextOffset).getOffset(), i + 1);
- }
- assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "500");
- assertEquals(messageBatch.getFirstMessageOffset().toString(), "0");
- assertEquals(messageBatch.getLastMessageMetadata().getOffset().toString(), "499");
- assertEquals(messageBatch.getLastMessageMetadata().getNextOffset().toString(), "500");
-
- // Test second half batch
- messageBatch = consumer.fetchMessages(new LongMsgOffset(500), 10000);
- assertEquals(messageBatch.getMessageCount(), 500);
- assertEquals(messageBatch.getUnfilteredMessageCount(), 500);
- for (int i = 0; i < 500; i++) {
- StreamMessage streamMessage = messageBatch.getStreamMessage(i);
- assertEquals(new String((byte[]) streamMessage.getValue()), "sample_msg_" + (500 + i));
- StreamMessageMetadata metadata = streamMessage.getMetadata();
- assertEquals(metadata.getRecordIngestionTimeMs(), TIMESTAMP + 500 + i);
- StreamPartitionMsgOffset offset = metadata.getOffset();
- assertTrue(offset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) offset).getOffset(), 500 + i);
- StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
- assertTrue(nextOffset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) nextOffset).getOffset(), 501 + i);
- }
- assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "1000");
- assertEquals(messageBatch.getFirstMessageOffset().toString(), "500");
- assertEquals(messageBatch.getLastMessageMetadata().getOffset().toString(), "999");
- assertEquals(messageBatch.getLastMessageMetadata().getNextOffset().toString(), "1000");
-
- // Some random range
- messageBatch = consumer.fetchMessages(new LongMsgOffset(10), 10000);
- assertEquals(messageBatch.getMessageCount(), 500);
- assertEquals(messageBatch.getUnfilteredMessageCount(), 500);
- for (int i = 0; i < 500; i++) {
- StreamMessage streamMessage = messageBatch.getStreamMessage(i);
- assertEquals(new String((byte[]) streamMessage.getValue()), "sample_msg_" + (10 + i));
- StreamMessageMetadata metadata = streamMessage.getMetadata();
- assertEquals(metadata.getRecordIngestionTimeMs(), TIMESTAMP + 10 + i);
- StreamPartitionMsgOffset offset = metadata.getOffset();
- assertTrue(offset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) offset).getOffset(), 10 + i);
- StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
- assertTrue(nextOffset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) nextOffset).getOffset(), 11 + i);
- }
- assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "510");
- assertEquals(messageBatch.getFirstMessageOffset().toString(), "10");
- assertEquals(messageBatch.getLastMessageMetadata().getOffset().toString(), "509");
- assertEquals(messageBatch.getLastMessageMetadata().getNextOffset().toString(), "510");
-
- // Some random range
- messageBatch = consumer.fetchMessages(new LongMsgOffset(610), 10000);
- assertEquals(messageBatch.getMessageCount(), 390);
- assertEquals(messageBatch.getUnfilteredMessageCount(), 390);
- for (int i = 0; i < 390; i++) {
- StreamMessage streamMessage = messageBatch.getStreamMessage(i);
- assertEquals(new String((byte[]) streamMessage.getValue()), "sample_msg_" + (610 + i));
- StreamMessageMetadata metadata = streamMessage.getMetadata();
- assertEquals(metadata.getRecordIngestionTimeMs(), TIMESTAMP + 610 + i);
- StreamPartitionMsgOffset offset = metadata.getOffset();
- assertTrue(offset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) offset).getOffset(), 610 + i);
- StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
- assertTrue(nextOffset instanceof LongMsgOffset);
- assertEquals(((LongMsgOffset) nextOffset).getOffset(), 611 + i);
- }
- assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "1000");
- assertEquals(messageBatch.getFirstMessageOffset().toString(), "610");
- assertEquals(messageBatch.getLastMessageMetadata().getOffset().toString(), "999");
- assertEquals(messageBatch.getLastMessageMetadata().getNextOffset().toString(), "1000");
- }
- }
-
- protected String getKafkaConsumerFactoryName() {
- return KafkaConsumerFactory.class.getName();
- }
-
- @Test
- public void testOffsetsExpired()
- throws TimeoutException {
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", "kafka");
- streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_3);
- streamConfigMap.put("stream.kafka.broker.list", _kafkaBrokerAddress);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- streamConfigMap.put("auto.offset.reset", "earliest");
- StreamConfig streamConfig = new StreamConfig("tableName_REALTIME", streamConfigMap);
-
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- PartitionGroupConsumer consumer = streamConsumerFactory.createPartitionGroupConsumer("clientId",
- new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0),
- new LongMsgOffset(NUM_MSG_PRODUCED_PER_PARTITION), "CONSUMING"));
-
- // Start offset has expired. Automatically reset to earliest available and fetch whatever available
- MessageBatch messageBatch = consumer.fetchMessages(new LongMsgOffset(0), 10000);
- assertEquals(messageBatch.getMessageCount(), 500);
- assertEquals(messageBatch.getUnfilteredMessageCount(), 500);
- for (int i = 0; i < 500; i++) {
- assertEquals(new String((byte[]) messageBatch.getStreamMessage(i).getValue()), "sample_msg_" + (200 + i));
- }
- assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "700");
- }
-
- @Test
- public void testGetTopics() {
- String streamType = "kafka";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String clientId = "clientId";
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", "NON_EXISTING_TOPIC");
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
- StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
-
- KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
- List<StreamMetadataProvider.TopicMetadata> topics = streamMetadataProvider.getTopics();
- List<String> topicNames = topics.stream()
- .map(StreamMetadataProvider.TopicMetadata::getName)
- .collect(Collectors.toList());
- assertTrue(topicNames.containsAll(List.of(TEST_TOPIC_1, TEST_TOPIC_2, TEST_TOPIC_3)));
- }
-
- @Test
- public void testFetchLatestStreamOffset()
- throws IOException {
- StreamConfig streamConfig = getStreamConfig(TEST_TOPIC_2);
- try (KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider("clientId",
- streamConfig)) {
- Set<Integer> partitions = new HashSet<>();
- partitions.add(0);
- partitions.add(1);
- Map<Integer, StreamPartitionMsgOffset> partitionMsgOffsetMap =
- streamMetadataProvider.fetchLatestStreamOffset(partitions, 1000);
- Assert.assertEquals(((LongMsgOffset) (partitionMsgOffsetMap.get(0))).getOffset(), NUM_MSG_PRODUCED_PER_PARTITION);
- Assert.assertEquals(((LongMsgOffset) (partitionMsgOffsetMap.get(1))).getOffset(), NUM_MSG_PRODUCED_PER_PARTITION);
- }
- }
-
- @Test
- void testBatchSizeInBytesIsCalculatedCorrectly() {
- TopicPartition topicPartition = new TopicPartition("test-topic", 0);
-
- class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer {
-
- public FakeKafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) {
- super(clientId, streamConfig, partition);
- }
-
- @Override
- protected Consumer<String, Bytes> createConsumer(Properties consumerProp) {
- Consumer<String, Bytes> mockConsumer = mock(Consumer.class);
- // Mock records
- ConsumerRecord<String, Bytes> record1 =
- new ConsumerRecord<>("test-topic", 0, 0L, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, NULL_CHECKSUM, 4,
- 5, "key1", new Bytes("value1".getBytes(StandardCharsets.UTF_8)));
- ConsumerRecord<String, Bytes> record2 =
- new ConsumerRecord<>("test-topic", 0, 0L, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, NULL_CHECKSUM, 4,
- 9, "key2", new Bytes("value2".getBytes(StandardCharsets.UTF_8)));
- ConsumerRecord<String, Bytes> record3 =
- new ConsumerRecord<>("test-topic", 0, 0L, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, NULL_CHECKSUM, 4,
- -1, "key2", new Bytes("value2".getBytes(StandardCharsets.UTF_8)));
- // Mock return of poll()
- ConsumerRecords<String, Bytes> consumerRecords = new ConsumerRecords<>(
- Map.of(topicPartition, List.of(record1, record2, record3))
- );
- when(mockConsumer.poll(any(Duration.class))).thenReturn(consumerRecords);
- return mockConsumer;
- }
- }
-
- FakeKafkaPartitionLevelConsumer kafkaSimpleStreamConsumer =
- new FakeKafkaPartitionLevelConsumer("clientId-test", getStreamConfig("test-topic"), 0);
- KafkaMessageBatch kafkaMessageBatch = kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), 10000);
- Assert.assertEquals(kafkaMessageBatch.getSizeInBytes(), 14);
- }
-
- private KafkaPartitionLevelConsumer createConsumer(String clientId, StreamConfig streamConfig, int partition) {
- if (RANDOM.nextDouble() < 0.5) {
- return new KafkaPartitionLevelConsumer(clientId, streamConfig, partition);
- } else {
- return new KafkaPartitionLevelConsumer(clientId, streamConfig, partition,
- new ExponentialBackoffRetryPolicy(2, 1000, 1.1));
- }
- }
-
- private StreamConfig getStreamConfig(String topicName) {
- String streamType = "kafka";
- String streamKafkaBrokerList = _kafkaBrokerAddress;
- String tableNameWithType = "tableName_REALTIME";
-
- Map<String, String> streamConfigMap = new HashMap<>();
- streamConfigMap.put("streamType", streamType);
- streamConfigMap.put("stream.kafka.topic.name", topicName);
- streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
- streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
- streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
-
- return new StreamConfig(tableNameWithType, streamConfigMap);
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java
deleted file mode 100644
index 05c15bc0a4d..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.pinot.plugin.stream.kafka.KafkaConfigBackwardCompatibleUtils;
-import org.apache.pinot.plugin.stream.kafka.KafkaPartitionLevelStreamConfig;
-import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.stream.StreamConfigProperties;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-
-public class KafkaPartitionLevelStreamConfigTest {
- private static final String KAFKA_DECODER_CLASS_NAME =
- "org.apache.pinot.plugin.inputformat.avro.KafkaAvroMessageDecoder";
-
- private KafkaPartitionLevelStreamConfig getStreamConfig(String topic, String bootstrapHosts, String buffer,
- String socketTimeout) {
- return getStreamConfig(topic, bootstrapHosts, buffer, socketTimeout, null, null, null);
- }
-
- private KafkaPartitionLevelStreamConfig getStreamConfig(String topic, String bootstrapHosts, String buffer,
- String socketTimeout, String isolationLevel) {
- return getStreamConfig(topic, bootstrapHosts, buffer, socketTimeout, null, null, isolationLevel);
- }
-
- private KafkaPartitionLevelStreamConfig getStreamConfig(String topic, String bootstrapHosts, String buffer,
- String socketTimeout, String fetcherSize, String fetcherMinBytes, String isolationLevel) {
- return getStreamConfig(topic, bootstrapHosts, buffer, socketTimeout, fetcherSize, fetcherMinBytes, isolationLevel,
- null);
- }
-
- private KafkaPartitionLevelStreamConfig getStreamConfig(String topic, String bootstrapHosts, String buffer,
- String socketTimeout, String fetcherSize, String fetcherMinBytes, String isolationLevel,
- String populateRowMetadata) {
- return new KafkaPartitionLevelStreamConfig(
- getStreamConfig(topic, bootstrapHosts, buffer, socketTimeout, fetcherSize, fetcherMinBytes, isolationLevel,
- populateRowMetadata, "tableName_REALTIME"));
- }
-
- private StreamConfig getStreamConfig(String topic, String bootstrapHosts, String buffer,
- String socketTimeout, String fetcherSize, String fetcherMinBytes, String isolationLevel,
- String populateRowMetadata, String tableNameWithType) {
- Map<String, String> streamConfigMap = new HashMap<>();
- String streamType = "kafka";
- String consumerFactoryClassName = KafkaConsumerFactory.class.getName();
- streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
- streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
- StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClassName);
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- KAFKA_DECODER_CLASS_NAME);
- streamConfigMap.put("stream.kafka.broker.list", bootstrapHosts);
- if (buffer != null) {
- streamConfigMap.put("stream.kafka.buffer.size", buffer);
- }
- if (socketTimeout != null) {
- streamConfigMap.put("stream.kafka.socket.timeout", socketTimeout);
- }
- if (fetcherSize != null) {
- streamConfigMap.put("stream.kafka.fetcher.size", fetcherSize);
- }
- if (fetcherMinBytes != null) {
- streamConfigMap.put("stream.kafka.fetcher.minBytes", fetcherMinBytes);
- }
- if (isolationLevel != null) {
- streamConfigMap.put("stream.kafka.isolation.level", isolationLevel);
- }
- if (populateRowMetadata != null) {
- streamConfigMap.put("stream.kafka.metadata.populate", populateRowMetadata);
- }
- return new StreamConfig(tableNameWithType, streamConfigMap);
- }
-
- @Test
- public void testGetKafkaIsolationLevel() {
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "", "", "", "read_committed");
- Assert.assertEquals("read_committed", config.getKafkaIsolationLevel());
- }
-
- @Test
- public void testGetKafkaTopicName() {
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "", "", "");
- Assert.assertEquals("topic", config.getKafkaTopicName());
- }
-
- @Test
- public void testGetBootstrapHosts() {
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", "");
- Assert.assertEquals("host1", config.getBootstrapHosts());
- }
-
- @Test
- public void testGetKafkaBufferSize() {
- // test default
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", null, "");
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
- config.getKafkaBufferSize());
-
- config = getStreamConfig("topic", "host1", "", "");
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
- config.getKafkaBufferSize());
-
- config = getStreamConfig("topic", "host1", "bad value", "");
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
- config.getKafkaBufferSize());
-
- // correct config
- config = getStreamConfig("topic", "host1", "100", "");
- Assert.assertEquals(100, config.getKafkaBufferSize());
- }
-
- @Test
- public void testGetKafkaSocketTimeout() {
- // test default
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", null);
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
- config.getKafkaSocketTimeout());
-
- config = getStreamConfig("topic", "host1", "", "");
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
- config.getKafkaSocketTimeout());
-
- config = getStreamConfig("topic", "host1", "", "bad value");
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
- config.getKafkaSocketTimeout());
-
- // correct config
- config = getStreamConfig("topic", "host1", "", "100");
- Assert.assertEquals(100, config.getKafkaSocketTimeout());
- }
-
- @Test
- public void testGetFetcherSize() {
- // test default
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", "", "", null, null);
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
- config.getKafkaFetcherSizeBytes());
-
- config = getStreamConfig("topic", "host1", "100", "", "", null, null);
- Assert.assertEquals(100, config.getKafkaFetcherSizeBytes());
-
- config = getStreamConfig("topic", "host1", "100", "", "bad value", null, null);
- Assert.assertEquals(100, config.getKafkaFetcherSizeBytes());
-
- // correct config
- config = getStreamConfig("topic", "host1", "100", "", "200", null, null);
- Assert.assertEquals(200, config.getKafkaFetcherSizeBytes());
- }
-
- @Test
- public void testGetFetcherMinBytes() {
- // test default
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", "", "", null, null);
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
- config.getKafkaFetcherMinBytes());
-
- config = getStreamConfig("topic", "host1", "", "", "", "", null);
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
- config.getKafkaFetcherMinBytes());
-
- config = getStreamConfig("topic", "host1", "", "", "", "bad value", null);
- Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
- config.getKafkaFetcherMinBytes());
-
- // correct config
- config = getStreamConfig("topic", "host1", "", "", "", "100", null);
- Assert.assertEquals(100, config.getKafkaFetcherMinBytes());
- }
-
- @Test
- public void testIsPopulateRowMetadata() {
- // test default
- KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", null, null, null, null, null, null);
- Assert.assertFalse(config.isPopulateMetadata());
-
- config = getStreamConfig("topic", "host1", null, null, null, null, null, "bad value");
- Assert.assertFalse(config.isPopulateMetadata());
-
- config = getStreamConfig("topic", "host1", null, null, null, null, null, "TrUe");
- Assert.assertTrue(config.isPopulateMetadata());
- }
-
- @Test
- public void testBackwardCompatibility() {
- StreamConfig streamConfig = getStreamConfig("topic", "host1", "100", "200", "300", "400", "read_committed", "true",
- "myTable_REALTIME");
- streamConfig.getStreamConfigsMap().put("sasl.jaas.config",
- "org.apache.kafka.common.security.plain.PlainLoginModule required \n username=\"user\" \n password=\"pwd\";");
- streamConfig.getStreamConfigsMap().put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.WrongStringDeserializer");
- KafkaConfigBackwardCompatibleUtils.handleStreamConfig(streamConfig);
- Assert.assertEquals(streamConfig.getStreamConfigsMap().get("sasl.jaas.config"),
- "org.apache.kafka.common.security.plain.PlainLoginModule required \n username=\"user\" \n password=\"pwd\";");
- Assert.assertEquals(streamConfig.getStreamConfigsMap().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG),
- "org.apache.kafka.common.serialization.WrongStringDeserializer");
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/utils/MiniKafkaCluster.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/utils/MiniKafkaCluster.java
deleted file mode 100644
index 56be7c1f578..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/utils/MiniKafkaCluster.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka20.utils;
-
-import java.io.Closeable;
-import java.io.File;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import org.apache.commons.io.FileUtils;
-import org.apache.kafka.clients.admin.AdminClient;
-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.kafka.common.utils.Time;
-import org.apache.pinot.plugin.stream.kafka.utils.EmbeddedZooKeeper;
-import scala.Option;
-
-
-public final class MiniKafkaCluster implements Closeable {
- private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "MiniKafkaCluster-" + UUID.randomUUID());
-
- private final EmbeddedZooKeeper _zkServer;
- private final KafkaServer _kafkaServer;
- private final String _kafkaServerAddress;
- private final AdminClient _adminClient;
-
- public MiniKafkaCluster(String brokerId)
- throws IOException, InterruptedException {
- _zkServer = new EmbeddedZooKeeper(new File(TEMP_DIR, "zk"));
- int kafkaServerPort = getAvailablePort();
- KafkaConfig kafkaBrokerConfig = new KafkaConfig(createBrokerConfig(brokerId, kafkaServerPort));
- _kafkaServer = new KafkaServer(kafkaBrokerConfig, Time.SYSTEM, Option.empty(), false);
- _kafkaServerAddress = "localhost:" + kafkaServerPort;
- Properties kafkaClientConfig = new Properties();
- kafkaClientConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _kafkaServerAddress);
- _adminClient = AdminClient.create(kafkaClientConfig);
- }
-
- private static int getAvailablePort() {
- try {
- try (ServerSocket socket = new ServerSocket(0)) {
- return socket.getLocalPort();
- }
- } catch (IOException e) {
- throw new RuntimeException("Failed to find available port to use", e);
- }
- }
-
- private Properties createBrokerConfig(String brokerId, int port) {
- Properties props = new Properties();
- props.put("broker.id", brokerId);
- // We need to explicitly set the network interface we want to let Kafka bind to.
- // By default, it will bind to all the network interfaces, which might not be accessible always
- // in a container based environment.
- props.put("host.name", "localhost");
- props.put("port", Integer.toString(port));
- props.put("log.dir", new File(TEMP_DIR, "log").getPath());
- props.put("zookeeper.connect", _zkServer.getZkAddress());
- props.put("zookeeper.session.timeout.ms", "30000");
- props.put("controlled.shutdown.enable", "true");
- props.put("delete.topic.enable", "true");
- props.put("auto.create.topics.enable", "true");
- props.put("offsets.topic.replication.factor", "1");
- props.put("log.cleaner.dedupe.buffer.size", "2097152");
- return props;
- }
-
- public void start() {
- _kafkaServer.startup();
- }
-
- @Override
- public void close()
- throws IOException {
- _kafkaServer.shutdown();
- _zkServer.close();
- FileUtils.deleteDirectory(TEMP_DIR);
- }
-
- public String getKafkaServerAddress() {
- return _kafkaServerAddress;
- }
-
- public void createTopic(String topicName, int numPartitions, int replicationFactor)
- throws ExecutionException, InterruptedException {
- NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) replicationFactor);
- _adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
- }
-
- public void deleteTopic(String topicName)
- throws ExecutionException, InterruptedException {
- _adminClient.deleteTopics(Collections.singletonList(topicName)).all().get();
- }
-
- public void deleteRecordsBeforeOffset(String topicName, int partitionId, long offset) {
- Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
- recordsToDelete.put(new TopicPartition(topicName, partitionId), RecordsToDelete.beforeOffset(offset));
- _adminClient.deleteRecords(recordsToDelete);
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/resources/log4j2.xml b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/resources/log4j2.xml
deleted file mode 100644
index 439331f9d7b..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/resources/log4j2.xml
+++ /dev/null
@@ -1,36 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing,
- software distributed under the License is distributed on an
- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- KIND, either express or implied. See the License for the
- specific language governing permissions and limitations
- under the License.
-
--->
-<Configuration>
- <Appenders>
- <Console name="console" target="SYSTEM_OUT">
- <PatternLayout pattern="%d{HH:mm:ss.SSS} %p [%c{1}] [%t] %m%n"/>
- </Console>
- </Appenders>
- <Loggers>
- <Logger name="org.apache.pinot" level="warn" additivity="false">
- <AppenderRef ref="console"/>
- </Logger>
- <Root level="error">
- <AppenderRef ref="console"/>
- </Root>
- </Loggers>
-</Configuration>
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
index 913f5a1c969..4abc9f21919 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.admin.AdminClient;
@@ -69,11 +70,12 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(timeoutMillis);
if (CollectionUtils.isNotEmpty(partitionInfos)) {
return partitionInfos.size();
}
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ throw new TransientConsumerException(new RuntimeException(
+ String.format("Failed to fetch partition information for topic: %s", _topic)));
} catch (TimeoutException e) {
throw new TransientConsumerException(e);
}
@@ -82,9 +84,10 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
@Override
public Set<Integer> fetchPartitionIds(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(timeoutMillis);
if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ throw new TransientConsumerException(new RuntimeException(
+ String.format("Failed to fetch partition information for topic: %s", _topic)));
}
Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
for (PartitionInfo partitionInfo : partitionInfos) {
@@ -284,4 +287,77 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
throws IOException {
super.close();
}
+
+ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
+ long deadlineMs = System.currentTimeMillis() + timeoutMillis;
+ List<PartitionInfo> partitionInfos = null;
+ Exception lastError = null;
+ boolean topicMissing = false;
+ while (System.currentTimeMillis() < deadlineMs) {
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ long requestTimeoutMs = Math.min(500L, Math.max(1L, remainingMs));
+ try {
+ partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(requestTimeoutMs));
+ } catch (TimeoutException e) {
+ lastError = e;
+ }
+
+ if (CollectionUtils.isNotEmpty(partitionInfos)) {
+ return partitionInfos;
+ }
+
+ try {
+ if (!topicExists(requestTimeoutMs)) {
+ topicMissing = true;
+ lastError = new RuntimeException(String.format("Topic does not exist: %s", _topic));
+ } else {
+ topicMissing = false;
+ }
+ } catch (TransientConsumerException e) {
+ lastError = e;
+ } catch (RuntimeException e) {
+ lastError = e;
+ }
+
+ if (System.currentTimeMillis() >= deadlineMs) {
+ break;
+ }
+ try {
+ Thread.sleep(100L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (lastError != null) {
+ if (topicMissing) {
+ throw new RuntimeException(String.format("Topic does not exist: %s", _topic));
+ }
+ if (lastError instanceof TransientConsumerException) {
+ throw (TransientConsumerException) lastError;
+ }
+ if (lastError instanceof TimeoutException) {
+ throw new TransientConsumerException(lastError);
+ }
+ }
+
+ throw new TransientConsumerException(
+ new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
+ }
+
+ private boolean topicExists(long timeoutMillis) {
+ try {
+ AdminClient adminClient = getOrCreateSharedAdminClient();
+ ListTopicsResult result = adminClient.listTopics();
+ return result.names().get(timeoutMillis, TimeUnit.MILLISECONDS).contains(_topic);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new TransientConsumerException(e);
+ } catch (ExecutionException e) {
+ throw new TransientConsumerException(e);
+ } catch (java.util.concurrent.TimeoutException e) {
+ throw new TransientConsumerException(e);
+ }
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataProducer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaDataProducer.java
similarity index 92%
rename from pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataProducer.java
rename to pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaDataProducer.java
index 9e1e627b123..eee7b2b99c9 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/server/KafkaDataProducer.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaDataProducer.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.plugin.stream.kafka20.server;
+package org.apache.pinot.plugin.stream.kafka30.server;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -59,7 +59,8 @@ public class KafkaDataProducer implements StreamDataProducer {
try {
_producer = new KafkaProducer<>(props);
} catch (Exception e) {
- LOGGER.error("Failed to create a Kafka 2 Producer.", e);
+ LOGGER.error("Failed to create a Kafka Producer with properties: {}", props, e);
+ throw new IllegalStateException("Failed to initialize KafkaDataProducer; see cause for details.", e);
}
}
@@ -109,6 +110,8 @@ public class KafkaDataProducer implements StreamDataProducer {
@Override
public void close() {
- _producer.close();
+ if (_producer != null) {
+ _producer.close();
+ }
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartable.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartable.java
new file mode 100644
index 00000000000..6d2a7b6f908
--- /dev/null
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartable.java
@@ -0,0 +1,701 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.stream.kafka30.server;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+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.common.TopicPartition;
+import org.apache.pinot.spi.stream.StreamDataServerStartable;
+import org.apache.pinot.spi.utils.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Kafka 3.x implementation of {@link StreamDataServerStartable}.
+ *
+ * This class can either connect to an external broker or start a managed single-node KRaft
+ * container for local quickstart usage. It is not thread-safe.
+ */
+public class KafkaServerStartable implements StreamDataServerStartable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(KafkaServerStartable.class);
+
+ private static final String KAFKA_IMAGE = "apache/kafka:3.9.1";
+ private static final int KAFKA_STARTUP_TIMEOUT_SECONDS = 60;
+
+ private static final int DEFAULT_BROKER_ID = 0;
+ private static final int DEFAULT_KAFKA_PORT = 19092;
+ private static final String DEFAULT_KAFKA_BROKER = "localhost:" + DEFAULT_KAFKA_PORT;
+
+ private static final String KAFKA_SERVER_BOOTSTRAP_SERVERS = "kafka.server.bootstrap.servers";
+ private static final String KAFKA_SERVER_PORT = "kafka.server.port";
+ private static final String KAFKA_SERVER_BROKER_ID = "kafka.server.broker.id";
+ private static final String KAFKA_SERVER_OWNER_NAME = "kafka.server.owner.name";
+ private static final String KAFKA_SERVER_ALLOW_MANAGED_FOR_CONFIGURED_BROKER =
+ "kafka.server.allow.managed.for.configured.broker";
+ private static final String KAFKA_SERVER_CONTAINER_NAME = "kafka.server.container.name";
+ private static final String KAFKA_SERVER_NETWORK_NAME = "kafka.server.network.name";
+ private static final String KAFKA_SERVER_CLUSTER_ID = "kafka.server.cluster.id";
+ private static final String KAFKA_SERVER_CLUSTER_SIZE = "kafka.server.cluster.size";
+ private static final String KAFKA_SERVER_CONTROLLER_QUORUM_VOTERS = "kafka.server.controller.quorum.voters";
+ private static final String KAFKA_SERVER_INTERNAL_HOST = "kafka.server.internal.host";
+ private static final String KAFKA_SERVER_SKIP_READINESS_CHECK = "kafka.server.skip.readiness.check";
+
+ private static final String KAFKA_EXTERNAL_LISTENER_PORT = "9094";
+ private static final String KAFKA_BROKER_LISTENER_PORT = "9092";
+ private static final String KAFKA_CONTROLLER_LISTENER_PORT = "9093";
+ private static final int TOPIC_MUTATION_RETRIES = 5;
+ private static final int DOCKER_NETWORK_READY_RETRIES = 10;
+ private static final long DOCKER_NETWORK_READY_WAIT_MS = 500L;
+ private static final int DOCKER_NETWORK_REMOVE_RETRIES = 5;
+ private static final long DOCKER_NETWORK_REMOVE_RETRY_WAIT_MS = 500L;
+ private static final int DOCKER_PULL_RETRIES = 3;
+ private static final long DOCKER_PULL_RETRY_WAIT_MS = 2_000L;
+ private static final int CONTAINER_STOP_TIMEOUT_SECONDS = 30;
+ private static final int PORT_RELEASE_TIMEOUT_SECONDS = 30;
+ private static final long STOP_POLL_INTERVAL_MS = 200L;
+ private static final int PROCESS_TIMEOUT_SECONDS = 120;
+
+ private static final String LOCALHOST = "localhost";
+ private static final String LOOPBACK = "127.0.0.1";
+ private static final Map<String, AtomicInteger> MANAGED_NETWORK_USAGE_COUNTS = new ConcurrentHashMap<>();
+ private static final Set<String> MANAGED_NETWORKS_CREATED = ConcurrentHashMap.newKeySet();
+
+ private String _ownerName = KafkaServerStartable.class.getSimpleName();
+ private String _kafkaBrokerList = DEFAULT_KAFKA_BROKER;
+ private int _kafkaPort = DEFAULT_KAFKA_PORT;
+ private int _kafkaBrokerId = DEFAULT_BROKER_ID;
+ private boolean _allowManagedKafkaForConfiguredBroker;
+ private String _containerName;
+ private String _dockerNetworkName;
+ private String _clusterId;
+ private String _controllerQuorumVoters;
+ private String _internalHost = LOCALHOST;
+ private int _clusterSize = 1;
+ private boolean _skipReadinessCheck;
+
+ private boolean _started;
+ private String _resolvedKafkaBrokerList;
+ private String _managedKafkaContainerName;
+ private boolean _registeredManagedNetworkUsage;
+ private Thread _shutdownHookThread;
+
+ @Override
+ public void init(Properties props) {
+ _ownerName = props.getProperty(KAFKA_SERVER_OWNER_NAME, _ownerName);
+ _kafkaBrokerList = props.getProperty(KAFKA_SERVER_BOOTSTRAP_SERVERS, _kafkaBrokerList);
+ _kafkaPort = parseInt(props.getProperty(KAFKA_SERVER_PORT), parsePort(_kafkaBrokerList, DEFAULT_KAFKA_PORT));
+ _kafkaBrokerId = parseInt(props.getProperty(KAFKA_SERVER_BROKER_ID), DEFAULT_BROKER_ID);
+ _allowManagedKafkaForConfiguredBroker =
+ Boolean.parseBoolean(props.getProperty(KAFKA_SERVER_ALLOW_MANAGED_FOR_CONFIGURED_BROKER, "false"));
+ _containerName = props.getProperty(KAFKA_SERVER_CONTAINER_NAME);
+ _dockerNetworkName = props.getProperty(KAFKA_SERVER_NETWORK_NAME);
+ _clusterId = props.getProperty(KAFKA_SERVER_CLUSTER_ID);
+ _controllerQuorumVoters = props.getProperty(KAFKA_SERVER_CONTROLLER_QUORUM_VOTERS);
+ _internalHost = props.getProperty(KAFKA_SERVER_INTERNAL_HOST, LOCALHOST);
+ _clusterSize = Math.max(1, parseInt(props.getProperty(KAFKA_SERVER_CLUSTER_SIZE), 1));
+ _skipReadinessCheck = Boolean.parseBoolean(props.getProperty(KAFKA_SERVER_SKIP_READINESS_CHECK, "false"));
+ }
+
+ @Override
+ public void start() {
+ if (_started) {
+ return;
+ }
+
+ try {
+ boolean startManagedKafka = shouldStartManagedKafka(_kafkaBrokerList) && !isKafkaAvailable(_kafkaBrokerList);
+ if (startManagedKafka) {
+ ensureDockerDaemonRunning();
+ startManagedKafkaContainer();
+ }
+ if (startManagedKafka && _skipReadinessCheck) {
+ _resolvedKafkaBrokerList = _kafkaBrokerList;
+ } else {
+ _resolvedKafkaBrokerList = resolveKafkaBrokerList(_kafkaBrokerList);
+ }
+ LOGGER.info("Using Kafka at {}", _resolvedKafkaBrokerList);
+ _started = true;
+ } catch (Exception e) {
+ cleanupManagedKafkaResources();
+ throw new RuntimeException("Failed to initialize Kafka for " + _ownerName, e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ unregisterShutdownHook();
+ cleanupManagedKafkaResources();
+ _started = false;
+ }
+
+ @Override
+ public void createTopic(String topic, Properties topicProps) {
+ ensureStarted();
+ int numPartitions = parseInt(String.valueOf(topicProps.getOrDefault("partition", 1)), 1);
+ int requestedReplicationFactor =
+ parseInt(String.valueOf(topicProps.getOrDefault("replicationFactor", 1)), 1);
+ short replicationFactor = (short) Math.max(1, Math.min(_clusterSize, requestedReplicationFactor));
+ try (AdminClient adminClient = createKafkaAdminClient()) {
+ NewTopic newTopic = new NewTopic(topic, numPartitions, replicationFactor);
+ runAdminWithRetry(() -> adminClient.createTopics(Collections.singletonList(newTopic)).all().get(),
+ "create topic: " + topic);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create topic: " + topic, e);
+ }
+ }
+
+ @Override
+ public void deleteTopic(String topic) {
+ ensureStarted();
+ try (AdminClient adminClient = createKafkaAdminClient()) {
+ runAdminWithRetry(() -> adminClient.deleteTopics(Collections.singletonList(topic)).all().get(),
+ "delete topic: " + topic);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete topic: " + topic, e);
+ }
+ }
+
+ @Override
+ public void createPartitions(String topic, int numPartitions) {
+ ensureStarted();
+ try (AdminClient adminClient = createKafkaAdminClient()) {
+ runAdminWithRetry(() -> {
+ adminClient.createPartitions(Collections.singletonMap(topic, NewPartitions.increaseTo(numPartitions))).all()
+ .get();
+ return null;
+ }, "create partitions for topic: " + topic);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create partitions for topic: " + topic, e);
+ }
+ }
+
+ @Override
+ public void deleteRecordsBeforeOffset(String topic, int partition, long offset) {
+ ensureStarted();
+ TopicPartition topicPartition = new TopicPartition(topic, partition);
+ try (AdminClient adminClient = createKafkaAdminClient()) {
+ runAdminWithRetry(() -> {
+ adminClient.deleteRecords(Collections.singletonMap(topicPartition, RecordsToDelete.beforeOffset(offset))).all()
+ .get();
+ return null;
+ }, "delete records before offset for topic: " + topic + ", partition: " + partition);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete records before offset for topic: " + topic + ", partition: "
+ + partition + ", offset: " + offset, e);
+ }
+ }
+
+ @Override
+ public int getPort() {
+ return _kafkaPort;
+ }
+
+ private AdminClient createKafkaAdminClient() {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", _resolvedKafkaBrokerList);
+ return AdminClient.create(props);
+ }
+
+ String resolveKafkaBrokerList(String configuredBrokerList) {
+ if (isKafkaAvailable(configuredBrokerList)) {
+ return configuredBrokerList;
+ }
+
+ String fallback = "localhost:9092";
+ if (isKafkaAvailable(fallback)) {
+ LOGGER.warn("Configured Kafka broker list {} is not reachable; falling back to default {}", configuredBrokerList,
+ fallback);
+ return fallback;
+ }
+
+ throw new IllegalStateException("Kafka broker list is not reachable: " + configuredBrokerList
+ + ". Please start Kafka or pass -kafkaBrokerList.");
+ }
+
+ boolean shouldStartManagedKafka(String brokerList) {
+ if (_allowManagedKafkaForConfiguredBroker) {
+ return isLocalBroker(brokerList);
+ }
+ return managedKafkaBroker().equals(brokerList);
+ }
+
+ static boolean isLocalBroker(String brokerList) {
+ int separator = brokerList.lastIndexOf(':');
+ if (separator <= 0 || separator == brokerList.length() - 1) {
+ return false;
+ }
+ String host = brokerList.substring(0, separator);
+ return LOCALHOST.equalsIgnoreCase(host) || LOOPBACK.equals(host);
+ }
+
+ private static int parsePort(String brokerList, int defaultPort) {
+ int separator = brokerList.lastIndexOf(':');
+ if (separator <= 0 || separator == brokerList.length() - 1) {
+ return defaultPort;
+ }
+ return parseInt(brokerList.substring(separator + 1), defaultPort);
+ }
+
+ private static int parseInt(String value, int defaultValue) {
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return Integer.parseInt(value);
+ } catch (NumberFormatException e) {
+ return defaultValue;
+ }
+ }
+
+ boolean isKafkaAvailable(String brokerList) {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", brokerList);
+ props.put("request.timeout.ms", "2000");
+ props.put("default.api.timeout.ms", "2000");
+ try (AdminClient adminClient = AdminClient.create(props)) {
+ adminClient.describeCluster().nodes().get(2, TimeUnit.SECONDS);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private void ensureDockerDaemonRunning()
+ throws Exception {
+ LOGGER.info("Checking Docker daemon availability");
+ try {
+ runProcess(List.of("docker", "info"), false);
+ } catch (Exception e) {
+ throw new IllegalStateException("Docker daemon is not running or not reachable. Quickstart starts Kafka in "
+ + "Docker on " + managedKafkaBroker()
+ + ". Start Docker, or pass -kafkaBrokerList <host:port> to use an external Kafka broker.", e);
+ }
+ }
+
+ private void pullKafkaDockerImage()
+ throws Exception {
+ if (isKafkaDockerImagePresent()) {
+ LOGGER.info("Kafka Docker image {} already exists locally", KAFKA_IMAGE);
+ return;
+ }
+ LOGGER.info("Pulling Kafka Docker image {}", KAFKA_IMAGE);
+ Exception lastException = null;
+ for (int attempt = 1; attempt <= DOCKER_PULL_RETRIES; attempt++) {
+ try {
+ runProcess(List.of("docker", "pull", KAFKA_IMAGE), false);
+ LOGGER.info("Kafka Docker image {} is ready", KAFKA_IMAGE);
+ return;
+ } catch (Exception e) {
+ lastException = e;
+ if (!isTransientDockerPullError(e) || attempt == DOCKER_PULL_RETRIES) {
+ throw e;
+ }
+ LOGGER.warn("Failed to pull Kafka Docker image on attempt {}/{}; retrying in {}ms", attempt,
+ DOCKER_PULL_RETRIES, DOCKER_PULL_RETRY_WAIT_MS);
+ Thread.sleep(DOCKER_PULL_RETRY_WAIT_MS);
+ }
+ }
+ throw lastException;
+ }
+
+ private void startManagedKafkaContainer()
+ throws Exception {
+ _managedKafkaContainerName =
+ hasText(_containerName) ? _containerName : ("pinot-qs-kafka-" + System.currentTimeMillis());
+ pullKafkaDockerImage();
+ ensureDockerNetworkReady();
+
+ boolean singleNodeCluster = _clusterSize <= 1;
+ int topicReplicationFactor = Math.max(1, Math.min(3, _clusterSize));
+ String quorumVoters = hasText(_controllerQuorumVoters)
+ ? _controllerQuorumVoters
+ : (_kafkaBrokerId + "@"
+ + (singleNodeCluster ? LOCALHOST : _internalHost) + ":" + KAFKA_CONTROLLER_LISTENER_PORT);
+
+ List<String> runCommand = new ArrayList<>();
+ runCommand.add("docker");
+ runCommand.add("run");
+ runCommand.add("-d");
+ runCommand.add("--name");
+ runCommand.add(_managedKafkaContainerName);
+ if (hasText(_dockerNetworkName)) {
+ runCommand.add("--network");
+ runCommand.add(_dockerNetworkName);
+ }
+ runCommand.add("-p");
+ runCommand.add(_kafkaPort + ":" + (singleNodeCluster ? KAFKA_BROKER_LISTENER_PORT : KAFKA_EXTERNAL_LISTENER_PORT));
+ runCommand.add("-e");
+ runCommand.add("KAFKA_NODE_ID=" + _kafkaBrokerId);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_PROCESS_ROLES=broker,controller");
+ if (singleNodeCluster) {
+ runCommand.add("-e");
+ runCommand.add(
+ "KAFKA_LISTENERS=PLAINTEXT://:" + KAFKA_BROKER_LISTENER_PORT
+ + ",CONTROLLER://:" + KAFKA_CONTROLLER_LISTENER_PORT);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://" + managedKafkaBroker());
+ runCommand.add("-e");
+ runCommand.add("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT");
+ runCommand.add("-e");
+ runCommand.add("KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT");
+ } else {
+ runCommand.add("-e");
+ runCommand.add("KAFKA_LISTENERS=INTERNAL://:" + KAFKA_BROKER_LISTENER_PORT + ",EXTERNAL://:"
+ + KAFKA_EXTERNAL_LISTENER_PORT + ",CONTROLLER://:" + KAFKA_CONTROLLER_LISTENER_PORT);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_ADVERTISED_LISTENERS=INTERNAL://" + _internalHost + ":" + KAFKA_BROKER_LISTENER_PORT
+ + ",EXTERNAL://" + managedKafkaBroker());
+ runCommand.add("-e");
+ runCommand.add(
+ "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT");
+ runCommand.add("-e");
+ runCommand.add("KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL");
+ }
+ runCommand.add("-e");
+ runCommand.add("KAFKA_CONTROLLER_QUORUM_VOTERS=" + quorumVoters);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER");
+ runCommand.add("-e");
+ runCommand.add("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=" + topicReplicationFactor);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=" + topicReplicationFactor);
+ runCommand.add("-e");
+ runCommand.add("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=2");
+ if (hasText(_clusterId)) {
+ runCommand.add("-e");
+ runCommand.add("CLUSTER_ID=" + _clusterId);
+ }
+ runCommand.add(KAFKA_IMAGE);
+
+ runProcess(List.of("docker", "rm", "-f", _managedKafkaContainerName), true);
+ LOGGER.info("Starting Kafka container {}", _managedKafkaContainerName);
+ runKafkaContainerWithNetworkRetry(runCommand);
+ if (!_skipReadinessCheck) {
+ waitForKafkaReady();
+ }
+
+ registerShutdownHook();
+ }
+
+ private void waitForKafkaReady()
+ throws Exception {
+ String managedBroker = managedKafkaBroker();
+ LOGGER.info("Waiting for Kafka broker to become available on {}", managedBroker);
+ long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(KAFKA_STARTUP_TIMEOUT_SECONDS);
+ while (System.currentTimeMillis() < deadline) {
+ if (isKafkaAvailable(managedBroker)) {
+ LOGGER.info("Kafka broker is ready on {}", managedBroker);
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ throw new IllegalStateException("Kafka container did not become ready on " + managedBroker);
+ }
+
+ private void ensureDockerNetworkReady()
+ throws Exception {
+ if (!hasText(_dockerNetworkName)) {
+ return;
+ }
+ boolean existedBeforeCreate = isDockerNetworkReady();
+ String createOutput = "";
+ if (!existedBeforeCreate) {
+ createOutput = runProcess(List.of("docker", "network", "create", _dockerNetworkName), true);
+ if (createOutput.contains("all predefined address pools have been fully subnetted")) {
+ LOGGER.warn("Docker network address pool is exhausted. Cleaning stale Pinot networks and retrying.");
+ cleanupStalePinotDockerNetworks();
+ createOutput = runProcess(List.of("docker", "network", "create", _dockerNetworkName), true);
+ }
+ }
+ for (int attempt = 1; attempt <= DOCKER_NETWORK_READY_RETRIES; attempt++) {
+ if (isDockerNetworkReady()) {
+ if (!existedBeforeCreate) {
+ MANAGED_NETWORKS_CREATED.add(_dockerNetworkName);
+ }
+ registerManagedNetworkUsage();
+ return;
+ }
+ Thread.sleep(DOCKER_NETWORK_READY_WAIT_MS);
+ }
+ throw new IllegalStateException("Docker network '" + _dockerNetworkName + "' is not ready"
+ + (hasText(createOutput) ? ". docker network create output: " + createOutput : ""));
+ }
+
+ private boolean isDockerNetworkReady() {
+ return isDockerNetworkReady(_dockerNetworkName);
+ }
+
+ private static boolean isDockerNetworkReady(String networkName) {
+ try {
+ runProcess(List.of("docker", "network", "inspect", networkName), false);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private void registerManagedNetworkUsage() {
+ if (_registeredManagedNetworkUsage || !hasText(_dockerNetworkName)) {
+ return;
+ }
+ MANAGED_NETWORK_USAGE_COUNTS.computeIfAbsent(_dockerNetworkName, ignored -> new AtomicInteger()).incrementAndGet();
+ _registeredManagedNetworkUsage = true;
+ }
+
+ private void runKafkaContainerWithNetworkRetry(List<String> runCommand)
+ throws Exception {
+ try {
+ runProcess(runCommand, false);
+ } catch (Exception e) {
+ if (!hasText(_dockerNetworkName) || !isMissingDockerNetworkError(e)) {
+ throw e;
+ }
+ LOGGER.warn("Docker network {} was not found during Kafka container start; recreating and retrying once",
+ _dockerNetworkName);
+ ensureDockerNetworkReady();
+ runProcess(runCommand, false);
+ }
+ }
+
+ private boolean isMissingDockerNetworkError(Exception e) {
+ String message = e.getMessage();
+ if (message == null) {
+ return false;
+ }
+ return message.contains("network " + _dockerNetworkName + " not found")
+ || message.contains("No such network");
+ }
+
+ private boolean isKafkaDockerImagePresent() {
+ try {
+ runProcess(List.of("docker", "image", "inspect", KAFKA_IMAGE), false);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private boolean isTransientDockerPullError(Exception e) {
+ String message = e.getMessage();
+ if (message == null) {
+ return false;
+ }
+ return message.contains("TLS handshake timeout")
+ || message.contains("Client.Timeout exceeded")
+ || message.contains("i/o timeout");
+ }
+
+ private void cleanupStalePinotDockerNetworks()
+ throws Exception {
+ String networks = runProcess(List.of("docker", "network", "ls", "--format", "{{.Name}}"), true);
+ if (!hasText(networks)) {
+ return;
+ }
+ for (String network : networks.split("\\R")) {
+ if (network.startsWith("pinot-it-kafka-") || network.startsWith("pinot-test-network")) {
+ runProcess(List.of("docker", "network", "rm", network), true);
+ }
+ }
+ }
+
+ private void cleanupManagedKafkaResources() {
+ String managedKafkaContainerName = _managedKafkaContainerName;
+ if (managedKafkaContainerName != null) {
+ try {
+ LOGGER.info("Stopping managed Kafka container {}", managedKafkaContainerName);
+ runProcess(List.of("docker", "rm", "-f", managedKafkaContainerName), true);
+ waitForContainerToStop(managedKafkaContainerName);
+ waitForPortRelease(_kafkaPort);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to stop managed Kafka container {}", managedKafkaContainerName, e);
+ } finally {
+ _managedKafkaContainerName = null;
+ }
+ }
+ if (_registeredManagedNetworkUsage && hasText(_dockerNetworkName)) {
+ cleanupManagedKafkaNetwork();
+ }
+ }
+
+ private void registerShutdownHook() {
+ if (_shutdownHookThread != null) {
+ return;
+ }
+ _shutdownHookThread = new Thread(this::cleanupManagedKafkaResources, _ownerName + "-kafka-cleanup");
+ Runtime.getRuntime().addShutdownHook(_shutdownHookThread);
+ }
+
+ private void unregisterShutdownHook() {
+ if (_shutdownHookThread == null) {
+ return;
+ }
+ try {
+ Runtime.getRuntime().removeShutdownHook(_shutdownHookThread);
+ } catch (IllegalStateException e) {
+ // JVM is shutting down; cleanup hook is already running or about to run.
+ } finally {
+ _shutdownHookThread = null;
+ }
+ }
+
+ private void waitForContainerToStop(String containerName)
+ throws Exception {
+ long deadlineMs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(CONTAINER_STOP_TIMEOUT_SECONDS);
+ while (System.currentTimeMillis() < deadlineMs) {
+ if (!isContainerPresent(containerName)) {
+ LOGGER.info("Managed Kafka container {} is fully stopped", containerName);
+ return;
+ }
+ Thread.sleep(STOP_POLL_INTERVAL_MS);
+ }
+ throw new IllegalStateException("Managed Kafka container is still present after stop timeout: " + containerName);
+ }
+
+ private boolean isContainerPresent(String containerName) {
+ try {
+ runProcess(List.of("docker", "container", "inspect", containerName), false);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private void waitForPortRelease(int port)
+ throws Exception {
+ long deadlineMs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(PORT_RELEASE_TIMEOUT_SECONDS);
+ while (System.currentTimeMillis() < deadlineMs) {
+ if (NetUtils.available(port)) {
+ LOGGER.info("Kafka port {} is released", port);
+ return;
+ }
+ Thread.sleep(STOP_POLL_INTERVAL_MS);
+ }
+ throw new IllegalStateException("Kafka port is still in use after stop timeout: " + port);
+ }
+
+ private void cleanupManagedKafkaNetwork() {
+ _registeredManagedNetworkUsage = false;
+
+ AtomicInteger usage = MANAGED_NETWORK_USAGE_COUNTS.get(_dockerNetworkName);
+ if (usage == null) {
+ return;
+ }
+
+ int remainingUsage = usage.decrementAndGet();
+ if (remainingUsage > 0) {
+ return;
+ }
+ MANAGED_NETWORK_USAGE_COUNTS.remove(_dockerNetworkName, usage);
+
+ if (!MANAGED_NETWORKS_CREATED.remove(_dockerNetworkName)) {
+ return;
+ }
+ try {
+ removeManagedDockerNetworkWithRetry(_dockerNetworkName);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to remove managed Kafka Docker network {}", _dockerNetworkName, e);
+ }
+ }
+
+ private void removeManagedDockerNetworkWithRetry(String networkName)
+ throws Exception {
+ for (int attempt = 1; attempt <= DOCKER_NETWORK_REMOVE_RETRIES; attempt++) {
+ runProcess(List.of("docker", "network", "rm", networkName), true);
+ if (!isDockerNetworkReady(networkName)) {
+ return;
+ }
+ if (attempt < DOCKER_NETWORK_REMOVE_RETRIES) {
+ Thread.sleep(DOCKER_NETWORK_REMOVE_RETRY_WAIT_MS);
+ }
+ }
+ throw new IllegalStateException("Managed Kafka Docker network is still present: " + networkName);
+ }
+
+ private static String runProcess(List<String> command, boolean ignoreFailure)
+ throws Exception {
+ Process process = new ProcessBuilder(command).redirectErrorStream(true).start();
+ if (!process.waitFor(PROCESS_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ process.waitFor(PROCESS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ String timeoutOutput = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
+ throw new IllegalStateException("Command timed out after " + PROCESS_TIMEOUT_SECONDS + "s: "
+ + String.join(" ", command) + (timeoutOutput.isEmpty() ? "" : "\n" + timeoutOutput));
+ }
+ String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
+ int code = process.exitValue();
+ if (code != 0 && !ignoreFailure) {
+ throw new IllegalStateException("Command failed (" + code + "): " + String.join(" ", command)
+ + (output.isEmpty() ? "" : "\n" + output));
+ }
+ return output;
+ }
+
+ private void ensureStarted() {
+ if (!_started) {
+ throw new IllegalStateException("Kafka is not started for " + _ownerName
+ + ". Call start() explicitly before using Kafka operations.");
+ }
+ }
+
+ private <T> T runAdminWithRetry(AdminOperation<T> operation, String action)
+ throws Exception {
+ for (int attempt = 1; attempt <= TOPIC_MUTATION_RETRIES; attempt++) {
+ try {
+ return operation.execute();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException
+ && attempt < TOPIC_MUTATION_RETRIES) {
+ Thread.sleep(TimeUnit.SECONDS.toMillis(1));
+ continue;
+ }
+ throw e;
+ }
+ }
+ throw new IllegalStateException("Failed to " + action + " after retries");
+ }
+
+ private static boolean hasText(String value) {
+ return value != null && !value.isEmpty();
+ }
+
+ private String managedKafkaBroker() {
+ return LOCALHOST + ":" + _kafkaPort;
+ }
+
+ @FunctionalInterface
+ private interface AdminOperation<T> {
+ T execute()
+ throws Exception;
+ }
+}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory
similarity index 92%
rename from pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory
rename to pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory
index 954a1bb6be0..e61bac4cc35 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/resources/META-INF/services/org.apache.pinot.spi.stream.StreamConsumerFactory
@@ -16,4 +16,4 @@
# specific language governing permissions and limitations
# under the License.
#
-org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory
\ No newline at end of file
+org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java
index f727efa6426..d382537c6c6 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java
@@ -43,7 +43,7 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties;
-import org.apache.pinot.plugin.stream.kafka30.utils.MiniKafkaCluster;
+import org.apache.pinot.plugin.stream.kafka30.server.KafkaServerStartable;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.MessageBatch;
import org.apache.pinot.spi.stream.OffsetCriteria;
@@ -56,6 +56,7 @@ import org.apache.pinot.spi.stream.StreamMessage;
import org.apache.pinot.spi.stream.StreamMessageMetadata;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.spi.utils.retry.ExponentialBackoffRetryPolicy;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -83,18 +84,27 @@ public class KafkaPartitionLevelConsumerTest {
private static final long TIMESTAMP = Instant.now().toEpochMilli();
private static final Random RANDOM = new Random();
- private MiniKafkaCluster _kafkaCluster;
+ private KafkaServerStartable _kafkaCluster;
private String _kafkaBrokerAddress;
@BeforeClass
public void setUp()
throws Exception {
- _kafkaCluster = new MiniKafkaCluster("0");
+ int kafkaServerPort = NetUtils.findOpenPort();
+ Properties serverProperties = new Properties();
+ serverProperties.put("kafka.server.bootstrap.servers", "localhost:" + kafkaServerPort);
+ serverProperties.put("kafka.server.port", Integer.toString(kafkaServerPort));
+ serverProperties.put("kafka.server.broker.id", "0");
+ serverProperties.put("kafka.server.owner.name", getClass().getSimpleName());
+ serverProperties.put("kafka.server.allow.managed.for.configured.broker", "true");
+
+ _kafkaCluster = new KafkaServerStartable();
+ _kafkaCluster.init(serverProperties);
_kafkaCluster.start();
- _kafkaBrokerAddress = _kafkaCluster.getKafkaServerAddress();
- _kafkaCluster.createTopic(TEST_TOPIC_1, 1, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_2, 2, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_3, 1, 1);
+ _kafkaBrokerAddress = "localhost:" + _kafkaCluster.getPort();
+ _kafkaCluster.createTopic(TEST_TOPIC_1, 1);
+ _kafkaCluster.createTopic(TEST_TOPIC_2, 2);
+ _kafkaCluster.createTopic(TEST_TOPIC_3, 1);
Thread.sleep(STABILIZE_SLEEP_DELAYS);
produceMsgToKafka();
Thread.sleep(STABILIZE_SLEEP_DELAYS);
@@ -127,7 +137,7 @@ public class KafkaPartitionLevelConsumerTest {
_kafkaCluster.deleteTopic(TEST_TOPIC_2);
_kafkaCluster.deleteTopic(TEST_TOPIC_3);
} finally {
- _kafkaCluster.close();
+ _kafkaCluster.stop();
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartableTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartableTest.java
new file mode 100644
index 00000000000..5ac31b00377
--- /dev/null
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/server/KafkaServerStartableTest.java
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.stream.kafka30.server;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+import org.apache.pinot.spi.utils.NetUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class KafkaServerStartableTest {
+ @Test
+ public void testShouldStartManagedKafkaForDefaultManagedBroker() {
+ KafkaServerStartable kafkaServerStartable = new KafkaServerStartable();
+ kafkaServerStartable.init(new Properties());
+
+ Assert.assertTrue(kafkaServerStartable.shouldStartManagedKafka("localhost:19092"));
+ Assert.assertFalse(kafkaServerStartable.shouldStartManagedKafka("localhost:19093"));
+ }
+
+ @Test
+ public void testShouldStartManagedKafkaForConfiguredLocalBrokerWhenEnabled() {
+ KafkaServerStartable kafkaServerStartable = new KafkaServerStartable();
+ Properties serverProperties = new Properties();
+ serverProperties.put("kafka.server.allow.managed.for.configured.broker", "true");
+ kafkaServerStartable.init(serverProperties);
+
+ Assert.assertTrue(kafkaServerStartable.shouldStartManagedKafka("localhost:19093"));
+ Assert.assertTrue(kafkaServerStartable.shouldStartManagedKafka("127.0.0.1:19093"));
+ Assert.assertFalse(kafkaServerStartable.shouldStartManagedKafka("kafka-broker:19093"));
+ }
+
+ @Test
+ public void testResolveKafkaBrokerListFallsBackToDefault() {
+ TestableKafkaServerStartable kafkaServerStartable = new TestableKafkaServerStartable();
+ kafkaServerStartable.setAvailableBrokers("localhost:9092");
+
+ Assert.assertEquals(kafkaServerStartable.resolveKafkaBrokerList("localhost:19092"), "localhost:9092");
+ }
+
+ @Test(expectedExceptions = IllegalStateException.class,
+ expectedExceptionsMessageRegExp = "Kafka broker list is not reachable: .*")
+ public void testResolveKafkaBrokerListThrowsWhenNoBrokerAvailable() {
+ TestableKafkaServerStartable kafkaServerStartable = new TestableKafkaServerStartable();
+ kafkaServerStartable.resolveKafkaBrokerList("localhost:19092");
+ }
+
+ @Test
+ public void testStopReleasesBrokerPort()
+ throws Exception {
+ int kafkaServerPort = NetUtils.findOpenPort();
+ Properties serverProperties = new Properties();
+ serverProperties.put("kafka.server.bootstrap.servers", "localhost:" + kafkaServerPort);
+ serverProperties.put("kafka.server.port", Integer.toString(kafkaServerPort));
+ serverProperties.put("kafka.server.broker.id", "0");
+ serverProperties.put("kafka.server.owner.name", getClass().getSimpleName());
+ serverProperties.put("kafka.server.allow.managed.for.configured.broker", "true");
+
+ KafkaServerStartable kafkaServerStartable = new KafkaServerStartable();
+ kafkaServerStartable.init(serverProperties);
+ kafkaServerStartable.start();
+ try {
+ Assert.assertFalse(NetUtils.available(kafkaServerPort), "Kafka port should be in use while broker is running");
+ } finally {
+ kafkaServerStartable.stop();
+ }
+ Assert.assertTrue(NetUtils.available(kafkaServerPort), "Kafka port should be released after broker stop");
+ }
+
+ private static final class TestableKafkaServerStartable extends KafkaServerStartable {
+ private final Set<String> _availableBrokers = new HashSet<>();
+
+ void setAvailableBrokers(String... availableBrokers) {
+ _availableBrokers.clear();
+ _availableBrokers.addAll(Arrays.asList(availableBrokers));
+ }
+
+ @Override
+ boolean isKafkaAvailable(String brokerList) {
+ return _availableBrokers.contains(brokerList);
+ }
+ }
+}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/MiniKafkaCluster.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/MiniKafkaCluster.java
deleted file mode 100644
index dd5a6997e6b..00000000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/MiniKafkaCluster.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.plugin.stream.kafka30.utils;
-
-import java.io.Closeable;
-import java.io.File;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import org.apache.commons.io.FileUtils;
-import org.apache.kafka.clients.admin.AdminClient;
-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.kafka.common.utils.Time;
-import org.apache.pinot.plugin.stream.kafka.utils.EmbeddedZooKeeper;
-import scala.Option;
-
-
-public final class MiniKafkaCluster implements Closeable {
- private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "MiniKafkaCluster-" + UUID.randomUUID());
-
- private final EmbeddedZooKeeper _zkServer;
- private final KafkaServer _kafkaServer;
- private final String _kafkaServerAddress;
- private final AdminClient _adminClient;
-
- public MiniKafkaCluster(String brokerId)
- throws IOException, InterruptedException {
- _zkServer = new EmbeddedZooKeeper(new File(TEMP_DIR, "zk"));
- int kafkaServerPort = getAvailablePort();
- KafkaConfig kafkaBrokerConfig = new KafkaConfig(createBrokerConfig(brokerId, kafkaServerPort));
- _kafkaServer = new KafkaServer(kafkaBrokerConfig, Time.SYSTEM, Option.empty(), false);
- _kafkaServerAddress = "localhost:" + kafkaServerPort;
- Properties kafkaClientConfig = new Properties();
- kafkaClientConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _kafkaServerAddress);
- _adminClient = AdminClient.create(kafkaClientConfig);
- }
-
- private static int getAvailablePort() {
- try {
- try (ServerSocket socket = new ServerSocket(0)) {
- return socket.getLocalPort();
- }
- } catch (IOException e) {
- throw new RuntimeException("Failed to find available port to use", e);
- }
- }
-
- private Properties createBrokerConfig(String brokerId, int port) {
- Properties props = new Properties();
- props.put("broker.id", brokerId);
- props.put("host.name", "localhost");
- props.put("port", Integer.toString(port));
- props.put("listeners", "PLAINTEXT://localhost:" + port);
- props.put("log.dir", new File(TEMP_DIR, "log").getPath());
- props.put("zookeeper.connect", _zkServer.getZkAddress());
- props.put("zookeeper.session.timeout.ms", "30000");
- props.put("controlled.shutdown.enable", "true");
- props.put("delete.topic.enable", "true");
- props.put("auto.create.topics.enable", "true");
- props.put("offsets.topic.replication.factor", "1");
- props.put("log.cleaner.dedupe.buffer.size", "2097152");
- return props;
- }
-
- public void start() {
- _kafkaServer.startup();
- }
-
- @Override
- public void close()
- throws IOException {
- _kafkaServer.shutdown();
- _zkServer.close();
- FileUtils.deleteDirectory(TEMP_DIR);
- }
-
- public String getKafkaServerAddress() {
- return _kafkaServerAddress;
- }
-
- public void createTopic(String topicName, int numPartitions, int replicationFactor)
- throws ExecutionException, InterruptedException {
- NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) replicationFactor);
- int retries = 5;
- while (retries > 0) {
- try {
- _adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
- return;
- } catch (ExecutionException e) {
- if (e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException) {
- retries--;
- TimeUnit.SECONDS.sleep(1);
- } else {
- throw e;
- }
- }
- }
- throw new ExecutionException("Failed to create topic after retries", null);
- }
-
- public void deleteTopic(String topicName)
- throws ExecutionException, InterruptedException {
- _adminClient.deleteTopics(Collections.singletonList(topicName)).all().get();
- }
-
- public void deleteRecordsBeforeOffset(String topicName, int partitionId, long offset) {
- Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
- recordsToDelete.put(new TopicPartition(topicName, partitionId), RecordsToDelete.beforeOffset(offset));
- _adminClient.deleteRecords(recordsToDelete);
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
index 032fbf66fe2..9eb73deda63 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.admin.AdminClient;
@@ -69,11 +70,12 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(timeoutMillis);
if (CollectionUtils.isNotEmpty(partitionInfos)) {
return partitionInfos.size();
}
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ throw new TransientConsumerException(new RuntimeException(
+ String.format("Failed to fetch partition information for topic: %s", _topic)));
} catch (TimeoutException e) {
throw new TransientConsumerException(e);
}
@@ -82,9 +84,10 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
@Override
public Set<Integer> fetchPartitionIds(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(timeoutMillis);
if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ throw new TransientConsumerException(new RuntimeException(
+ String.format("Failed to fetch partition information for topic: %s", _topic)));
}
Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
for (PartitionInfo partitionInfo : partitionInfos) {
@@ -282,4 +285,77 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
throws IOException {
super.close();
}
+
+ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
+ long deadlineMs = System.currentTimeMillis() + timeoutMillis;
+ List<PartitionInfo> partitionInfos = null;
+ Exception lastError = null;
+ boolean topicMissing = false;
+ while (System.currentTimeMillis() < deadlineMs) {
+ long remainingMs = deadlineMs - System.currentTimeMillis();
+ long requestTimeoutMs = Math.min(500L, Math.max(1L, remainingMs));
+ try {
+ partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(requestTimeoutMs));
+ } catch (TimeoutException e) {
+ lastError = e;
+ }
+
+ if (CollectionUtils.isNotEmpty(partitionInfos)) {
+ return partitionInfos;
+ }
+
+ try {
+ if (!topicExists(requestTimeoutMs)) {
+ topicMissing = true;
+ lastError = new RuntimeException(String.format("Topic does not exist: %s", _topic));
+ } else {
+ topicMissing = false;
+ }
+ } catch (TransientConsumerException e) {
+ lastError = e;
+ } catch (RuntimeException e) {
+ lastError = e;
+ }
+
+ if (System.currentTimeMillis() >= deadlineMs) {
+ break;
+ }
+ try {
+ Thread.sleep(100L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (lastError != null) {
+ if (topicMissing) {
+ throw new RuntimeException(String.format("Topic does not exist: %s", _topic));
+ }
+ if (lastError instanceof TransientConsumerException) {
+ throw (TransientConsumerException) lastError;
+ }
+ if (lastError instanceof TimeoutException) {
+ throw new TransientConsumerException(lastError);
+ }
+ }
+
+ throw new TransientConsumerException(
+ new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
+ }
+
+ private boolean topicExists(long timeoutMillis) {
+ try {
+ AdminClient adminClient = getOrCreateSharedAdminClient();
+ ListTopicsResult result = adminClient.listTopics();
+ return result.names().get(timeoutMillis, TimeUnit.MILLISECONDS).contains(_topic);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new TransientConsumerException(e);
+ } catch (ExecutionException e) {
+ throw new TransientConsumerException(e);
+ } catch (java.util.concurrent.TimeoutException e) {
+ throw new TransientConsumerException(e);
+ }
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java
index f602d083963..f59c7a635a7 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java
@@ -103,9 +103,9 @@ public class KafkaPartitionLevelConsumerTest {
_kafkaCluster = new MiniKafkaCluster("0");
_kafkaCluster.start();
_kafkaBrokerAddress = _kafkaCluster.getKafkaServerAddress();
- _kafkaCluster.createTopic(TEST_TOPIC_1, 1, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_2, 2, 1);
- _kafkaCluster.createTopic(TEST_TOPIC_3, 1, 1);
+ _kafkaCluster.createTopic(TEST_TOPIC_1, 1);
+ _kafkaCluster.createTopic(TEST_TOPIC_2, 2);
+ _kafkaCluster.createTopic(TEST_TOPIC_3, 1);
Thread.sleep(STABILIZE_SLEEP_DELAYS);
produceMsgToKafka();
Thread.sleep(STABILIZE_SLEEP_DELAYS);
@@ -152,7 +152,7 @@ public class KafkaPartitionLevelConsumerTest {
_kafkaCluster.deleteTopic(TEST_TOPIC_2);
_kafkaCluster.deleteTopic(TEST_TOPIC_3);
} finally {
- _kafkaCluster.close();
+ _kafkaCluster.stop();
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/utils/MiniKafkaCluster.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/utils/MiniKafkaCluster.java
index 03535d2ee81..2ae615622da 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/utils/MiniKafkaCluster.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/utils/MiniKafkaCluster.java
@@ -27,10 +27,12 @@ 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;
@@ -39,7 +41,7 @@ import org.testcontainers.utility.DockerImageName;
* 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 Closeable {
+public final class MiniKafkaCluster implements StreamDataServerStartable, Closeable {
private static final String KAFKA_IMAGE = "apache/kafka:4.0.0";
@@ -51,15 +53,29 @@ public final class MiniKafkaCluster implements Closeable {
_kafkaContainer = new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE));
}
- public void start()
- throws Exception {
- _kafkaContainer.start();
+ @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 {
- _kafkaContainer.stop();
+ stop();
}
public String getKafkaServerAddress() {
@@ -72,8 +88,10 @@ public final class MiniKafkaCluster implements Closeable {
return AdminClient.create(kafkaClientConfig);
}
- public void createTopic(String topicName, int numPartitions, int replicationFactor)
- throws ExecutionException, InterruptedException {
+ @Override
+ public void createTopic(String topicName, Properties topicProps) {
+ int numPartitions = Integer.parseInt(String.valueOf(topicProps.getOrDefault("partition", 1)));
+ int replicationFactor = Integer.parseInt(String.valueOf(topicProps.getOrDefault("replicationFactor", 1)));
try (AdminClient adminClient = getOrCreateAdminClient()) {
NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) replicationFactor);
int retries = 5;
@@ -84,30 +102,72 @@ public final class MiniKafkaCluster implements Closeable {
} catch (ExecutionException e) {
if (e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException) {
retries--;
- TimeUnit.SECONDS.sleep(1);
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException interruptedException) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while creating topic: " + topicName, interruptedException);
+ }
} else {
- throw e;
+ throw new RuntimeException("Failed to create topic: " + topicName, e);
}
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while creating topic: " + topicName, e);
}
}
- throw new ExecutionException("Failed to create topic after retries", null);
+ throw new RuntimeException("Failed to create topic after retries: " + topicName);
}
}
- public void deleteTopic(String topicName)
- throws ExecutionException, InterruptedException {
+ @Override
+ public void deleteTopic(String topicName) {
try (AdminClient adminClient = getOrCreateAdminClient()) {
adminClient.deleteTopics(Collections.singletonList(topicName)).all().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while deleting topic: " + topicName, e);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete topic: " + topicName, e);
}
}
- public void deleteRecordsBeforeOffset(String topicName, int partitionId, long offset)
- throws ExecutionException, InterruptedException {
+ @Override
+ public void deleteRecordsBeforeOffset(String topicName, int partitionId, long offset) {
try (AdminClient adminClient = getOrCreateAdminClient()) {
Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
recordsToDelete.put(new TopicPartition(topicName, partitionId), RecordsToDelete.beforeOffset(offset));
// Wait for the deletion to complete
adminClient.deleteRecords(recordsToDelete).all().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Interrupted while deleting records for topic: " + topicName + ", partition: " + partitionId, e);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to delete records for topic: " + topicName + ", partition: " + partitionId, e);
}
}
+
+ @Override
+ public void createPartitions(String topicName, int numPartitions) {
+ try (AdminClient adminClient = getOrCreateAdminClient()) {
+ adminClient.createPartitions(Collections.singletonMap(topicName, NewPartitions.increaseTo(numPartitions))).all()
+ .get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while creating partitions for topic: " + topicName, e);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create partitions for topic: " + topicName, e);
+ }
+ }
+
+ @Override
+ public int getPort() {
+ String bootstrapServers = getKafkaServerAddress();
+ int schemeSeparator = bootstrapServers.indexOf("://");
+ String hostAndPort = schemeSeparator >= 0 ? bootstrapServers.substring(schemeSeparator + 3) : bootstrapServers;
+ int lastColon = hostAndPort.lastIndexOf(':');
+ return lastColon >= 0 ? Integer.parseInt(hostAndPort.substring(lastColon + 1)) : -1;
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pom.xml b/pinot-plugins/pinot-stream-ingestion/pom.xml
index 87e9760d4c2..b3137f27e06 100644
--- a/pinot-plugins/pinot-stream-ingestion/pom.xml
+++ b/pinot-plugins/pinot-stream-ingestion/pom.xml
@@ -37,7 +37,6 @@
<modules>
<module>pinot-kafka-base</module>
- <module>pinot-kafka-2.0</module>
<module>pinot-kafka-3.0</module>
<module>pinot-kafka-4.0</module>
<module>pinot-kinesis</module>
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java b/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
index 37abe9b03f3..4f7caa6db8c 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
@@ -96,8 +96,11 @@ public class PluginManager {
// StreamConsumerFactory
// Old-style class names mapped to current plugin packages
+ // Rename the usage of kafka 2 to kafka 3
put("org.apache.pinot.core.realtime.impl.kafka2.KafkaConsumerFactory",
- "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory");
+ "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory");
+ put("org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory");
put("org.apache.pinot.core.realtime.impl.kafka3.KafkaConsumerFactory",
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory");
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataServerStartable.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataServerStartable.java
index 9e66beeffb9..33610a510bd 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataServerStartable.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataServerStartable.java
@@ -55,6 +55,18 @@ public interface StreamDataServerStartable {
*/
void createTopic(String topic, Properties topicProps);
+ /**
+ * Create a data stream with the given partition count.
+ *
+ * @param topic topic name
+ * @param numPartitions number of partitions
+ */
+ default void createTopic(String topic, int numPartitions) {
+ Properties topicProps = new Properties();
+ topicProps.setProperty("partition", Integer.toString(numPartitions));
+ createTopic(topic, topicProps);
+ }
+
/**
* Delete a data stream (e.g Kafka topic) in the server.
*
@@ -73,6 +85,17 @@ public interface StreamDataServerStartable {
throw new NotImplementedException("createPartitions is not implemented!");
}
+ /**
+ * Delete records before a given offset for a topic partition.
+ *
+ * @param topic topic name
+ * @param partition partition id
+ * @param offset offset boundary (exclusive)
+ */
+ default void deleteRecordsBeforeOffset(String topic, int partition, long offset) {
+ throw new NotImplementedException("deleteRecordsBeforeOffset is not implemented!");
+ }
+
/**
* Get the port of the server.
*/
diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/plugin/PluginManagerTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/plugin/PluginManagerTest.java
index 83c2cc1d30e..b5f8ba2d8e6 100644
--- a/pinot-spi/src/test/java/org/apache/pinot/spi/plugin/PluginManagerTest.java
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/plugin/PluginManagerTest.java
@@ -192,7 +192,10 @@ public class PluginManagerTest {
// StreamConsumerFactory
Assert.assertEquals(PluginManager
.loadClassWithBackwardCompatibleCheck("org.apache.pinot.core.realtime.impl.kafka2.KafkaConsumerFactory"),
- "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory");
+ "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory");
+ Assert.assertEquals(PluginManager
+ .loadClassWithBackwardCompatibleCheck("org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory"),
+ "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory");
}
@DataProvider(name = "recordReaderClassNames")
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index e95a3dae456..f231cb8ded6 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -89,11 +89,6 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-thrift</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-2.0</artifactId>
- <scope>runtime</scope>
- </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-kafka-3.0</artifactId>
@@ -478,25 +473,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<executions>
- <execution>
- <id>copy-kafka-resources</id>
- <phase>process-resources</phase>
- <goals>
- <goal>copy-resources</goal>
- </goals>
- <configuration>
- <!-- this is important -->
- <overwrite>true</overwrite>
- <!-- target -->
- <outputDirectory>${basedir}/target/classes/sample_data/</outputDirectory>
- <resources>
- <resource>
- <!-- source -->
- <directory>src/main/resources/sample_data/kafka_2.0</directory>
- </resource>
- </resources>
- </configuration>
- </execution>
<execution>
<id>copy-batch-ingestion-resources</id>
<phase>process-resources</phase>
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java b/pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java
index b5edf2bd43a..cbd8cd98428 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java
@@ -37,7 +37,7 @@ import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.lang3.StringUtils;
-import org.apache.pinot.common.utils.ZkStarter;
+import org.apache.kafka.clients.admin.AdminClient;
import org.apache.pinot.spi.stream.StreamDataProducer;
import org.apache.pinot.spi.stream.StreamDataProvider;
import org.apache.pinot.spi.stream.StreamDataServerStartable;
@@ -99,7 +99,9 @@ public abstract class QuickStartBase {
protected String _zkExternalAddress;
protected String _configFilePath;
protected StreamDataServerStartable _kafkaStarter;
- protected ZkStarter.ZookeeperInstance _zookeeperInstance;
+ protected String _kafkaBrokerList = KafkaStarterUtils.DEFAULT_KAFKA_BROKER;
+ private boolean _kafkaBrokerListOverridden;
+ private String _resolvedKafkaBrokerList;
public QuickStartBase setDataDir(String dataDir) {
_dataDir = new File(dataDir);
@@ -142,6 +144,13 @@ public abstract class QuickStartBase {
return this;
}
+ public QuickStartBase setKafkaBrokerList(String kafkaBrokerList) {
+ _kafkaBrokerList = kafkaBrokerList;
+ _kafkaBrokerListOverridden = true;
+ _resolvedKafkaBrokerList = null;
+ return this;
+ }
+
public abstract List<String> types();
public void runSampleQueries(QuickstartRunner runner)
@@ -341,8 +350,9 @@ public abstract class QuickStartBase {
return DEFAULT_STREAM_TABLE_DIRECTORIES;
}
- protected static void publishStreamDataToKafka(String tableName, File dataDir)
+ protected void publishStreamDataToKafka(String tableName, File dataDir)
throws Exception {
+ resolveKafkaBrokerList();
switch (tableName) {
case "githubEvents":
publishLineSplitFileToKafka("githubEvents", new File(dataDir, "/rawdata/2021-07-21-few-hours.json"));
@@ -359,10 +369,10 @@ public abstract class QuickStartBase {
}
}
- protected static void publishLineSplitFileToKafka(String topicName, File dataFile)
+ protected void publishLineSplitFileToKafka(String topicName, File dataFile)
throws Exception {
Properties properties = new Properties();
- properties.put("metadata.broker.list", KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
+ properties.put("metadata.broker.list", resolveKafkaBrokerList());
properties.put("serializer.class", "kafka.serializer.DefaultEncoder");
properties.put("request.required.acks", "1");
StreamDataProducer producer =
@@ -399,22 +409,33 @@ public abstract class QuickStartBase {
}
}
- protected void startKafka() {
+ public final void startKafka() {
+ if (_kafkaStarter != null) {
+ return;
+ }
printStatus(Quickstart.Color.CYAN, "***** Starting Kafka *****");
- _zookeeperInstance = ZkStarter.startLocalZkServer();
try {
+ String configuredBrokerList = getConfiguredKafkaBrokerList();
+ Properties props = new Properties();
+ props.put(KafkaStarterUtils.KAFKA_SERVER_OWNER_NAME, getClass().getSimpleName());
+ props.put(KafkaStarterUtils.KAFKA_SERVER_BOOTSTRAP_SERVERS, configuredBrokerList);
+ props.put(KafkaStarterUtils.KAFKA_SERVER_PORT,
+ String.valueOf(KafkaStarterUtils.parsePort(configuredBrokerList, KafkaStarterUtils.DEFAULT_KAFKA_PORT)));
+ props.put(KafkaStarterUtils.KAFKA_SERVER_BROKER_ID, String.valueOf(KafkaStarterUtils.DEFAULT_BROKER_ID));
+ props.put(KafkaStarterUtils.KAFKA_SERVER_ALLOW_MANAGED_FOR_CONFIGURED_BROKER, "false");
_kafkaStarter = StreamDataProvider.getServerDataStartable(KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME,
- KafkaStarterUtils.getDefaultKafkaConfiguration(_zookeeperInstance));
+ props);
} catch (Exception e) {
throw new RuntimeException("Failed to start " + KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME, e);
}
_kafkaStarter.start();
+ _resolvedKafkaBrokerList = resolveKafkaBrokerList();
+ printStatus(Quickstart.Color.CYAN, "***** Using Kafka at " + _resolvedKafkaBrokerList + " *****");
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
try {
- printStatus(Quickstart.Color.GREEN, "***** Shutting down kafka and zookeeper *****");
+ printStatus(Quickstart.Color.GREEN, "***** Shutting down kafka *****");
_kafkaStarter.stop();
- ZkStarter.stopLocalZkServer(_zookeeperInstance);
} catch (Exception e) {
e.printStackTrace();
}
@@ -425,6 +446,9 @@ public abstract class QuickStartBase {
public void startAllDataStreams(StreamDataServerStartable kafkaStarter, File quickstartTmpDir)
throws Exception {
+ if (kafkaStarter == null) {
+ throw new IllegalArgumentException("Kafka starter must not be null");
+ }
for (String streamName : getDefaultStreamTableDirectories().keySet()) {
switch (streamName) {
case "airlineStats":
@@ -968,4 +992,35 @@ public abstract class QuickStartBase {
printStatus(Quickstart.Color.YELLOW, prettyPrintResponse(runner.runQuery(q7)));
printStatus(Quickstart.Color.GREEN, "***************************************************");
}
+
+ protected AdminClient createKafkaAdminClient() {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", resolveKafkaBrokerList());
+ return AdminClient.create(props);
+ }
+
+ private String resolveKafkaBrokerList() {
+ if (_resolvedKafkaBrokerList != null) {
+ return _resolvedKafkaBrokerList;
+ }
+ _resolvedKafkaBrokerList =
+ KafkaStarterUtils.resolveKafkaBrokerList(getConfiguredKafkaBrokerList(), _kafkaBrokerListOverridden);
+ return _resolvedKafkaBrokerList;
+ }
+
+ private String getConfiguredKafkaBrokerList() {
+ String envKafka = firstNonEmpty(System.getProperty("pinot.kafka.broker.list"),
+ System.getenv("PINOT_KAFKA_BROKER_LIST"),
+ System.getenv("KAFKA_BROKER_LIST"));
+ return _kafkaBrokerListOverridden ? _kafkaBrokerList : (envKafka != null ? envKafka : _kafkaBrokerList);
+ }
+
+ private static String firstNonEmpty(String... values) {
+ for (String value : values) {
+ if (value != null && !value.isBlank()) {
+ return value;
+ }
+ }
+ return null;
+ }
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java
index 3ebfa561acc..3973a8bb4a9 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/LaunchSparkDataIngestionJobCommand.java
@@ -74,7 +74,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-0.9:pinot-kafka-2.0", required =
+ @CommandLine.Option(names = {"-pluginsToExclude"}, defaultValue = "pinot-kafka-3.0", required =
false, arity = "1..*", split = ":", description =
"List " + "of plugin name separated by : to not load at runtime. e.g. pinto-s3:pinot-parquet")
private List<String> _pluginsToExclude;
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickStartCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickStartCommand.java
index e04e577a85d..db5b20a3874 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickStartCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickStartCommand.java
@@ -55,6 +55,10 @@ public class QuickStartCommand extends AbstractBaseAdminCommand implements Comma
description = "Config file path to override default pinot configs")
private String _configFilePath;
+ @CommandLine.Option(names = {"-kafkaBrokerList"}, required = false,
+ description = "Kafka broker list for streaming quickstarts (e.g. localhost:9092)")
+ private String _kafkaBrokerList;
+
@Override
public String getName() {
return "QuickStart";
@@ -109,6 +113,10 @@ public class QuickStartCommand extends AbstractBaseAdminCommand implements Comma
_configFilePath = configFilePath;
}
+ public void setKafkaBrokerList(String kafkaBrokerList) {
+ _kafkaBrokerList = kafkaBrokerList;
+ }
+
@Override
public String toString() {
return ("QuickStart -type " + _type);
@@ -166,6 +174,10 @@ public class QuickStartCommand extends AbstractBaseAdminCommand implements Comma
quickstart.setConfigFilePath(_configFilePath);
}
+ if (_kafkaBrokerList != null) {
+ quickstart.setKafkaBrokerList(_kafkaBrokerList);
+ }
+
quickstart.execute();
return true;
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartKafkaCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartKafkaCommand.java
index e4dd36070eb..041ad39d00f 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartKafkaCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartKafkaCommand.java
@@ -18,8 +18,6 @@
*/
package org.apache.pinot.tools.admin.command;
-import java.io.File;
-import java.io.IOException;
import java.util.Properties;
import org.apache.pinot.spi.stream.StreamDataProvider;
import org.apache.pinot.spi.stream.StreamDataServerStartable;
@@ -43,10 +41,6 @@ public class StartKafkaCommand extends AbstractBaseAdminCommand implements Comma
@CommandLine.Option(names = {"-brokerId"}, required = false, description = "Kafka broker ID.")
private int _brokerId = KafkaStarterUtils.DEFAULT_BROKER_ID;
- @CommandLine.Option(names = {"-zkAddress"}, required = false, description = "Address of Zookeeper.")
- private String _zkAddress = KafkaStarterUtils.getDefaultKafkaZKAddress();
- private StreamDataServerStartable _kafkaStarter;
-
@Override
public String getName() {
return "StartKafka";
@@ -54,7 +48,7 @@ public class StartKafkaCommand extends AbstractBaseAdminCommand implements Comma
@Override
public String toString() {
- return "StartKafka -port " + _port + " -brokerId " + _brokerId + " -zkAddress " + _zkAddress;
+ return "StartKafka -port " + _port + " -brokerId " + _brokerId;
}
@Override
@@ -64,20 +58,25 @@ public class StartKafkaCommand extends AbstractBaseAdminCommand implements Comma
@Override
public boolean execute()
- throws IOException {
- Properties kafkaConfiguration = KafkaStarterUtils.getDefaultKafkaConfiguration();
- kafkaConfiguration.put(KafkaStarterUtils.BROKER_ID, _brokerId);
- kafkaConfiguration.put(KafkaStarterUtils.PORT, _port);
- kafkaConfiguration.put(KafkaStarterUtils.ZOOKEEPER_CONNECT, _zkAddress);
- try {
- _kafkaStarter = StreamDataProvider
- .getServerDataStartable(KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME, kafkaConfiguration);
- } catch (Exception e) {
- throw new RuntimeException("Failed to start " + KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME, e);
+ throws Exception {
+ if (_port <= 0 || _port > 65535) {
+ throw new IllegalArgumentException("Invalid Kafka port: " + _port);
+ }
+ if (_brokerId < 0) {
+ throw new IllegalArgumentException("Invalid Kafka brokerId: " + _brokerId);
}
- _kafkaStarter.start();
- LOGGER.info("Start kafka at localhost:{} in thread {}", _port, Thread.currentThread().getName());
- savePID(System.getProperty("java.io.tmpdir") + File.separator + ".kafka.pid");
+
+ Properties props = new Properties();
+ props.put(KafkaStarterUtils.KAFKA_SERVER_OWNER_NAME, getName());
+ props.put(KafkaStarterUtils.KAFKA_SERVER_BOOTSTRAP_SERVERS, "localhost:" + _port);
+ props.put(KafkaStarterUtils.KAFKA_SERVER_PORT, String.valueOf(_port));
+ props.put(KafkaStarterUtils.KAFKA_SERVER_BROKER_ID, String.valueOf(_brokerId));
+ props.put(KafkaStarterUtils.KAFKA_SERVER_ALLOW_MANAGED_FOR_CONFIGURED_BROKER, "true");
+
+ StreamDataServerStartable kafkaStarter =
+ StreamDataProvider.getServerDataStartable(KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME, props);
+ kafkaStarter.start();
+ LOGGER.info("Kafka server started at localhost:{}", _port);
return true;
}
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StopProcessCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StopProcessCommand.java
index 4b0ac67088e..2d5db80d543 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StopProcessCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StopProcessCommand.java
@@ -22,9 +22,16 @@ import java.io.File;
import java.io.FileReader;
import java.io.FilenameFilter;
import java.io.IOException;
-import java.util.HashMap;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.tools.Command;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +47,10 @@ import picocli.CommandLine;
@CommandLine.Command(name = "StopProcess", mixinStandardHelpOptions = true)
public class StopProcessCommand extends AbstractBaseAdminCommand implements Command {
private static final Logger LOGGER = LoggerFactory.getLogger(StopProcessCommand.class);
+ private static final String QUICKSTART_KAFKA_CONTAINER_PREFIX = "pinot-qs-kafka-";
+ private static final int PORT_RELEASE_TIMEOUT_SECONDS = 30;
+ private static final long PORT_RELEASE_POLL_INTERVAL_MS = 200L;
+ private static final int PROCESS_TIMEOUT_SECONDS = 60;
@CommandLine.Option(names = {"-controller"}, required = false, description = "Stop the PinotController process.")
private boolean _controller = false;
@@ -88,15 +99,28 @@ public class StopProcessCommand extends AbstractBaseAdminCommand implements Comm
return this;
}
+ public StopProcessCommand stopKafka() {
+ _kafka = true;
+ return this;
+ }
+
@Override
public boolean execute()
throws Exception {
LOGGER.info("Executing command: {}", toString());
- Map<String, String> processes = new HashMap<String, String>();
+ Map<String, String> processes = new LinkedHashMap<>();
String prefix = System.getProperty("java.io.tmpdir") + File.separator;
File tempDir = new File(System.getProperty("java.io.tmpdir"));
+ if (_kafka) {
+ stopManagedQuickstartKafkaContainers();
+ String kafkaPidFile = prefix + ".kafka.pid";
+ if (new File(kafkaPidFile).exists()) {
+ processes.put("Kafka", kafkaPidFile);
+ }
+ }
+
if (_server) {
File[] serverFiles = tempDir.listFiles(new FilenameFilter() {
@@ -152,10 +176,6 @@ public class StopProcessCommand extends AbstractBaseAdminCommand implements Comm
processes.put("Zookeeper", prefix + ".zooKeeper.pid");
}
- if (_kafka) {
- processes.put("Kafka", prefix + ".kafka.pid");
- }
-
boolean ret = true;
for (Map.Entry<String, String> entry : processes.entrySet()) {
try {
@@ -216,4 +236,85 @@ public class StopProcessCommand extends AbstractBaseAdminCommand implements Comm
file.delete();
return true;
}
+
+ private void stopManagedQuickstartKafkaContainers() {
+ try {
+ List<String> containers = runProcess(List.of("docker", "ps", "-a",
+ "--filter", "name=" + QUICKSTART_KAFKA_CONTAINER_PREFIX,
+ "--format", "{{.Names}}"));
+ Set<Integer> publishedPorts = new HashSet<>();
+ for (String container : containers) {
+ if (!container.isBlank()) {
+ publishedPorts.addAll(getPublishedPorts(container));
+ LOGGER.info("Stopping managed quickstart Kafka container: {}", container);
+ runProcess(List.of("docker", "rm", "-f", container));
+ }
+ }
+ waitForPortsReleased(publishedPorts);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to stop managed quickstart Kafka containers", e);
+ }
+ }
+
+ private static Set<Integer> getPublishedPorts(String containerName)
+ throws Exception {
+ Set<Integer> ports = new HashSet<>();
+ for (String line : runProcess(List.of("docker", "port", containerName))) {
+ int lastColon = line.lastIndexOf(':');
+ if (lastColon < 0 || lastColon == line.length() - 1) {
+ continue;
+ }
+ String portToken = line.substring(lastColon + 1).trim();
+ if (portToken.isEmpty()) {
+ continue;
+ }
+ try {
+ ports.add(Integer.parseInt(portToken));
+ } catch (NumberFormatException ignored) {
+ // Ignore malformed output lines and continue parsing remaining mappings.
+ }
+ }
+ return ports;
+ }
+
+ private static void waitForPortsReleased(Set<Integer> ports)
+ throws Exception {
+ if (ports.isEmpty()) {
+ return;
+ }
+ long deadlineMs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(PORT_RELEASE_TIMEOUT_SECONDS);
+ while (System.currentTimeMillis() < deadlineMs) {
+ boolean allReleased = true;
+ for (int port : ports) {
+ if (!NetUtils.available(port)) {
+ allReleased = false;
+ break;
+ }
+ }
+ if (allReleased) {
+ return;
+ }
+ Thread.sleep(PORT_RELEASE_POLL_INTERVAL_MS);
+ }
+ throw new IllegalStateException("Kafka ports are still in use after stop timeout: " + ports);
+ }
+
+ private static List<String> runProcess(List<String> command)
+ throws Exception {
+ Process process = new ProcessBuilder(command).redirectErrorStream(true).start();
+ if (!process.waitFor(PROCESS_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ process.waitFor(PROCESS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ String timeoutOutput = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
+ throw new IllegalStateException("Command timed out after " + PROCESS_TIMEOUT_SECONDS + "s: "
+ + String.join(" ", command) + (timeoutOutput.isEmpty() ? "" : "\n" + timeoutOutput));
+ }
+ String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
+ int code = process.exitValue();
+ if (code != 0) {
+ throw new IllegalStateException("Command failed (" + code + "): " + String.join(" ", command)
+ + (output.isBlank() ? "" : "\n" + output.trim()));
+ }
+ return output.lines().collect(Collectors.toList());
+ }
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
index 5fa2af97b7a..e7a18b1d249 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
@@ -24,14 +24,16 @@ import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.ServiceLoader;
-import org.apache.pinot.common.utils.ZkStarter;
+import java.util.concurrent.TimeUnit;
+import org.apache.kafka.clients.admin.AdminClient;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
-import org.apache.pinot.spi.stream.StreamDataProvider;
-import org.apache.pinot.spi.stream.StreamDataServerStartable;
-import org.apache.pinot.spi.utils.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class KafkaStarterUtils {
+ private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStarterUtils.class);
+
private KafkaStarterUtils() {
}
@@ -39,18 +41,22 @@ public class KafkaStarterUtils {
public static final int DEFAULT_KAFKA_PORT = 19092;
public static final String DEFAULT_KAFKA_BROKER = "localhost:" + DEFAULT_KAFKA_PORT;
- public static final String PORT = "port";
- public static final String BROKER_ID = "broker.id";
- public static final String ZOOKEEPER_CONNECT = "zookeeper.connect";
- private static final String LOG_DIRS = "log.dirs";
+ public static final String KAFKA_SERVER_BOOTSTRAP_SERVERS = "kafka.server.bootstrap.servers";
+ public static final String KAFKA_SERVER_PORT = "kafka.server.port";
+ public static final String KAFKA_SERVER_BROKER_ID = "kafka.server.broker.id";
+ public static final String KAFKA_SERVER_OWNER_NAME = "kafka.server.owner.name";
+ public static final String KAFKA_SERVER_ALLOW_MANAGED_FOR_CONFIGURED_BROKER =
+ "kafka.server.allow.managed.for.configured.broker";
public static final String KAFKA_SERVER_STARTABLE_CLASS_NAME =
- getKafkaConnectorPackageName() + ".server.KafkaDataServerStartable";
+ getKafkaConnectorPackageName() + ".server.KafkaServerStartable";
public static final String KAFKA_PRODUCER_CLASS_NAME = getKafkaConnectorPackageName() + ".server.KafkaDataProducer";
public static final String KAFKA_STREAM_CONSUMER_FACTORY_CLASS_NAME =
getKafkaConnectorPackageName() + ".KafkaConsumerFactory";
public static final String KAFKA_STREAM_LEVEL_CONSUMER_CLASS_NAME =
getKafkaConnectorPackageName() + ".KafkaStreamLevelConsumer";
+ public static final String KAFKA_JSON_MESSAGE_DECODER_CLASS_NAME =
+ "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder";
private static String getKafkaConnectorPackageName() {
Iterator<StreamConsumerFactory> iterator = ServiceLoader.load(StreamConsumerFactory.class).iterator();
@@ -64,87 +70,67 @@ public class KafkaStarterUtils {
return streamConsumerFactoryList.get(0);
}
- public static final String KAFKA_JSON_MESSAGE_DECODER_CLASS_NAME =
- "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder";
-
- public static Properties getDefaultKafkaConfiguration() {
- final Properties configuration = new Properties();
-
- // Enable topic deletion by default for integration tests
- configureTopicDeletion(configuration, true);
-
- // set the transaction state replication factor
- configureTransactionStateLogReplicationFactor(configuration, (short) 1);
- configuration.put("transaction.state.log.min.isr", 1);
-
- // Set host name
- configureHostName(configuration, "localhost");
- configureOffsetsTopicReplicationFactor(configuration, (short) 1);
- configuration.put(PORT, DEFAULT_KAFKA_PORT);
- configuration.put(BROKER_ID, DEFAULT_BROKER_ID);
- configuration.put(ZOOKEEPER_CONNECT, getDefaultKafkaZKAddress());
- configuration.put(LOG_DIRS, "/tmp/kafka-" + Double.toHexString(Math.random()));
-
- return configuration;
- }
-
- public static void configureOffsetsTopicReplicationFactor(Properties configuration, short replicationFactor) {
- configuration.put("offsets.topic.replication.factor", replicationFactor);
+ public static Properties getTopicCreationProps(int numKafkaPartitions) {
+ return getTopicCreationProps(numKafkaPartitions, 1);
}
- public static void configureTransactionStateLogReplicationFactor(Properties configuration, short replicationFactor) {
- configuration.put("transaction.state.log.replication.factor", replicationFactor);
+ public static Properties getTopicCreationProps(int numKafkaPartitions, int replicationFactor) {
+ Properties topicProps = new Properties();
+ topicProps.put("partition", numKafkaPartitions);
+ topicProps.put("replicationFactor", Math.max(1, replicationFactor));
+ return topicProps;
}
- public static void configureTopicDeletion(Properties configuration, boolean topicDeletionEnabled) {
- configuration.put("delete.topic.enable", Boolean.toString(topicDeletionEnabled));
- }
+ public static String resolveKafkaBrokerList(String configuredBrokerList, boolean brokerListOverridden) {
+ if (isKafkaAvailable(configuredBrokerList)) {
+ return configuredBrokerList;
+ }
+ if (brokerListOverridden) {
+ throw new IllegalStateException("Kafka broker list is not reachable: " + configuredBrokerList);
+ }
- public static void configureHostName(Properties configuration, String hostName) {
- configuration.put("host.name", hostName);
- }
+ String fallback = "localhost:9092";
+ if (isKafkaAvailable(fallback)) {
+ LOGGER.warn("Configured Kafka broker list {} is not reachable; falling back to default {}", configuredBrokerList,
+ fallback);
+ return fallback;
+ }
- public static String getDefaultKafkaZKAddress() {
- return ZkStarter.getDefaultZkStr() + "/kafka";
+ throw new IllegalStateException("Kafka broker list is not reachable: " + configuredBrokerList
+ + ". Please start Kafka or pass -kafkaBrokerList.");
}
- public static Properties getTopicCreationProps(int numKafkaPartitions) {
- Properties topicProps = new Properties();
- topicProps.put("partition", numKafkaPartitions);
- return topicProps;
+ public static boolean isKafkaAvailable(String brokerList) {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", brokerList);
+ props.put("request.timeout.ms", "2000");
+ props.put("default.api.timeout.ms", "2000");
+ try (AdminClient adminClient = AdminClient.create(props)) {
+ adminClient.describeCluster().nodes().get(2, TimeUnit.SECONDS);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
}
- public static List<StreamDataServerStartable> startServers(final int brokerCount, final int port, final String zkStr,
- final Properties configuration) {
- List<StreamDataServerStartable> startables = new ArrayList<>(brokerCount);
- for (int i = 0; i < brokerCount; i++) {
- startables.add(startServer(port + i, i, zkStr, configuration));
+ public static boolean isLocalBroker(String brokerList) {
+ int separator = brokerList.lastIndexOf(':');
+ if (separator <= 0 || separator == brokerList.length() - 1) {
+ return false;
}
- return startables;
+ String host = brokerList.substring(0, separator);
+ return "localhost".equalsIgnoreCase(host) || "127.0.0.1".equals(host);
}
- public static StreamDataServerStartable startServer(final int port, final int brokerId, final String zkStr,
- final Properties baseConf) {
- StreamDataServerStartable kafkaStarter;
- Properties configuration = new Properties(baseConf);
- int kafkaPort = NetUtils.findOpenPort(port);
+ public static int parsePort(String brokerList, int defaultPort) {
+ int separator = brokerList.lastIndexOf(':');
+ if (separator <= 0 || separator == brokerList.length() - 1) {
+ return defaultPort;
+ }
try {
- configureOffsetsTopicReplicationFactor(configuration, (short) 1);
- configuration.put(KafkaStarterUtils.PORT, kafkaPort);
- configuration.put(KafkaStarterUtils.BROKER_ID, brokerId);
- configuration.put(KafkaStarterUtils.ZOOKEEPER_CONNECT, zkStr);
- configuration.put(KafkaStarterUtils.LOG_DIRS, "/tmp/kafka-" + Double.toHexString(Math.random()));
- kafkaStarter = StreamDataProvider.getServerDataStartable(KAFKA_SERVER_STARTABLE_CLASS_NAME, configuration);
- } catch (Exception e) {
- throw new RuntimeException("Failed to start " + KAFKA_SERVER_STARTABLE_CLASS_NAME, e);
+ return Integer.parseInt(brokerList.substring(separator + 1));
+ } catch (NumberFormatException e) {
+ return defaultPort;
}
- kafkaStarter.start();
- return kafkaStarter;
- }
-
- public static Properties getDefaultKafkaConfiguration(ZkStarter.ZookeeperInstance zookeeperInstance) {
- Properties kafkaConfiguration = getDefaultKafkaConfiguration();
- kafkaConfiguration.put(ZOOKEEPER_CONNECT, zookeeperInstance.getZkUrl() + "/kafka");
- return kafkaConfiguration;
}
}
diff --git a/pinot-tools/src/main/resources/conf/sample_realtime_table_config.json b/pinot-tools/src/main/resources/conf/sample_realtime_table_config.json
index 7fc0fc87ee8..af9cb7d107c 100644
--- a/pinot-tools/src/main/resources/conf/sample_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/conf/sample_realtime_table_config.json
@@ -26,7 +26,7 @@
"stream.kafka.topic.name": "kafkaTopicName",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaAvroMessageDecoder",
"stream.kafka.decoder.prop.schema.registry.rest.url": "http://localhost:2222/schemaRegistry",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "largest",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "12h",
diff --git a/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json b/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
index f077d5a6abe..3eeac1fe36d 100644
--- a/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
@@ -28,7 +28,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "githubEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "1h",
diff --git a/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
index 9d1dcd7e142..60877325156 100644
--- a/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
@@ -21,7 +21,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
index f15b7d507fa..f6f2f6b7b87 100644
--- a/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
@@ -18,7 +18,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.hlc.zk.connect.string": "zookeeper:2181/kafka",
"stream.kafka.zk.broker.url": "zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
diff --git a/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
index 2aa2bacd3a3..1cd163410fd 100644
--- a/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
@@ -25,7 +25,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "dailySales",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
index 21ea0f4d42e..84941c97859 100644
--- a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
@@ -30,7 +30,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "fineFoodReviews",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
index 563abaa7426..4c83abd311b 100644
--- a/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
@@ -19,7 +19,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "githubEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
index f911ed6558c..a799acefff6 100644
--- a/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
@@ -16,7 +16,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "meetupRSVPEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "largest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
index 16e25ac700d..d5e00c02643 100644
--- a/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
@@ -16,7 +16,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "meetupRSVPComplexTypeEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
diff --git a/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
index 56fffee5ae2..a21dac74112 100644
--- a/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
@@ -29,7 +29,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "meetupRSVPJsonEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
diff --git a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
index 3ba5a54a7b9..6ba48ea1e7c 100644
--- a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
@@ -21,7 +21,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "pullRequestMergedEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "pinot-zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
diff --git a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
index 9ba9b86d8c8..7e74a032db8 100644
--- a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
@@ -21,7 +21,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "pullRequestMergedEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
diff --git a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
index f7551a1ba05..73c0c595304 100644
--- a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
@@ -58,7 +58,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "upsertJsonMeetupRSVPEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
diff --git a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
index bbf4e7473c8..1d984b7129f 100644
--- a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
@@ -46,7 +46,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "upsertMeetupRSVPEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
diff --git a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
index 80ce012e557..27a0cf70282 100644
--- a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
+++ b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
@@ -46,7 +46,7 @@
"streamType": "kafka",
"stream.kafka.topic.name": "upsertPartialMeetupRSVPEvents",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
- "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
+ "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
diff --git a/pom.xml b/pom.xml
index ba01d4a98bd..5e6eaf5bff3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -194,7 +194,6 @@
<spark2.version>2.4.8</spark2.version>
<spark3.version>3.5.8</spark3.version>
- <kafka2.version>2.8.2</kafka2.version>
<kafka3.version>3.9.1</kafka3.version>
<kafka4.version>4.1.1</kafka4.version>
<confluent.version>7.9.5</confluent.version>
@@ -674,11 +673,6 @@
<artifactId>pinot-kafka-base</artifactId>
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.pinot</groupId>
- <artifactId>pinot-kafka-2.0</artifactId>
- <version>${project.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-kafka-3.0</artifactId>
@@ -1757,12 +1751,12 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
- <version>${kafka2.version}</version>
+ <version>${kafka3.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.compat.version}</artifactId>
- <version>${kafka2.version}</version>
+ <version>${kafka3.version}</version>
</dependency>
<dependency>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org