You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "abhioncbr (via GitHub)" <gi...@apache.org> on 2024/08/26 18:43:45 UTC
[PR] Kafka3 support [pinot]
abhioncbr opened a new pull request, #13891:
URL: https://github.com/apache/pinot/pull/13891
TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on PR #13891:
URL: https://github.com/apache/pinot/pull/13891#issuecomment-2323295401
More refactoring can be done in the common code, but opening the PR for early feedback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1751633490
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
Review Comment:
change the property to `kafka3.lib.version` and move it to root pom?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1742841062
##########
pom.xml:
##########
@@ -1579,17 +1580,6 @@
<version>${scala.version}</version>
</dependency>
- <dependency>
Review Comment:
Suggest keeping it as is and only override the version in kafka3 module. This way we can ensure existing modules are unchanged
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1757350921
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/EmbeddedZooKeeper.java:
##########
@@ -0,0 +1,54 @@
+/**
+ * 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.InetSocketAddress;
+import org.apache.commons.io.FileUtils;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+
+public class EmbeddedZooKeeper implements Closeable {
Review Comment:
Moved to kafka-base as per the suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1757352328
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/MiniKafkaCluster.java:
##########
@@ -0,0 +1,138 @@
+/**
+ * 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 scala.Option;
+
+public final class MiniKafkaCluster implements Closeable {
Review Comment:
This starts the Kafka server based on Kafka 3. In my understanding, all others are running the Kafka2 server.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1744689447
##########
pom.xml:
##########
@@ -1579,17 +1580,6 @@
<version>${scala.version}</version>
</dependency>
- <dependency>
Review Comment:
reverted. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1744689214
##########
pinot-plugins/pinot-input-format/pinot-confluent-avro/pom.xml:
##########
@@ -49,10 +49,17 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
+ <version>${kafka2.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-client</artifactId>
+ <exclusions>
Review Comment:
Reverted, as per the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1750826051
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/EmbeddedZooKeeper.java:
##########
@@ -0,0 +1,54 @@
+/**
+ * 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.InetSocketAddress;
+import org.apache.commons.io.FileUtils;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+
+public class EmbeddedZooKeeper implements Closeable {
Review Comment:
Is this existed somewhere already?
If not, shall we move this to kafka-base or even pinot-common?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
Review Comment:
no need to exclude `kafka-clients` here?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.compat.version}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.101tec</groupId>
+ <artifactId>zkclient</artifactId>
+ </dependency>
+
+
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
Review Comment:
maybe directly put kafka-base test-jar here to not specify those test scope dependencies.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/utils/MiniKafkaCluster.java:
##########
@@ -0,0 +1,138 @@
+/**
+ * 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 scala.Option;
+
+public final class MiniKafkaCluster implements Closeable {
Review Comment:
I thought we have `KafkaDataServerStartable` for quickstart and integration tests.
Can you please check ?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.compat.version}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.101tec</groupId>
Review Comment:
zkclient is not used I think?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
Review Comment:
change the property to `kafka3.lib.version` and move it to root pom?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1739940748
##########
pom.xml:
##########
@@ -1579,17 +1580,6 @@
<version>${scala.version}</version>
</dependency>
- <dependency>
Review Comment:
It needs to be removed to support both Kafka2 and Kafka3 version-based streaming. In the future, when we use only Kafka3, we can bring the Kafka client dependency here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #13891:
URL: https://github.com/apache/pinot/pull/13891#issuecomment-2334910543
@swaminathanmanish @xiangfu0 Can you help review this PR? Once we verify Kafka3 consumer is backward compatible with Kafka2 producer, we can clean up Kafka2 code since it exposes some vulnerabilities.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1759750780
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeKafka3ClusterIntegrationTest.java:
##########
@@ -0,0 +1,528 @@
+/**
+ * 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.integration.tests;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.FileUtils;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.common.utils.HashUtil;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
+import org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory;
+import org.apache.pinot.plugin.stream.kafka30.KafkaPartitionLevelConsumer;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+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.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.junit.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Integration test for low-level Kafka3 consumer.
+ */
+public class LLCRealtimeKafka3ClusterIntegrationTest extends BaseRealtimeClusterIntegrationTest {
+ private static final String CONSUMER_DIRECTORY = "/tmp/consumer-test";
+ private static final long RANDOM_SEED = System.currentTimeMillis();
+ private static final Random RANDOM = new Random(RANDOM_SEED);
+
+ private final boolean _isDirectAlloc = RANDOM.nextBoolean();
+ private final boolean _isConsumerDirConfigured = RANDOM.nextBoolean();
+ private final boolean _enableLeadControllerResource = RANDOM.nextBoolean();
+ private final long _startTime = System.currentTimeMillis();
+
+ @Override
+ protected boolean injectTombstones() {
+ return true;
+ }
+
+ @Override
+ protected String getLoadMode() {
+ return ReadMode.mmap.name();
+ }
+
+ @Override
+ public void startController()
+ throws Exception {
+ super.startController();
+ enableResourceConfigForLeadControllerResource(_enableLeadControllerResource);
+ }
+
+ @Override
+ protected void overrideServerConf(PinotConfiguration configuration) {
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_REALTIME_OFFHEAP_ALLOCATION, true);
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_REALTIME_OFFHEAP_DIRECT_ALLOCATION, _isDirectAlloc);
+ if (_isConsumerDirConfigured) {
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_CONSUMER_DIR, CONSUMER_DIRECTORY);
+ }
+ }
+
+ @Override
+ protected void overrideControllerConf(Map<String, Object> properties) {
+ // Make sure the realtime segment validation manager does not run by itself, only when we invoke it.
+ properties.put(ControllerConf.ControllerPeriodicTasksConf.REALTIME_SEGMENT_VALIDATION_FREQUENCY_PERIOD, "2h");
+ properties.put(ControllerConf.ControllerPeriodicTasksConf.REALTIME_SEGMENT_VALIDATION_INITIAL_DELAY_IN_SECONDS,
+ 3600);
+ }
+
+ @Override
+ protected void runValidationJob(long timeoutMs)
+ throws Exception {
+ final int partition = ExceptingKafka3ConsumerFactory.PARTITION_FOR_EXCEPTIONS;
+ if (partition < 0) {
+ return;
+ }
+ int[] seqNumbers = {
+ ExceptingKafka3ConsumerFactory.SEQ_NUM_FOR_CREATE_EXCEPTION,
+ ExceptingKafka3ConsumerFactory.SEQ_NUM_FOR_CONSUME_EXCEPTION};
+ Arrays.sort(seqNumbers);
+ for (int seqNum : seqNumbers) {
+ if (seqNum < 0) {
+ continue;
+ }
+ TestUtils.waitForCondition(() -> isOffline(partition, seqNum), 5000L, timeoutMs,
+ "Failed to find offline segment in partition " + partition + " seqNum ", true,
+ Duration.ofMillis(timeoutMs / 10));
+ getControllerRequestClient().runPeriodicTask("RealtimeSegmentValidationManager");
+ }
+ }
+
+ private boolean isOffline(int partition, int seqNum) {
+ ExternalView ev = _helixAdmin.getResourceExternalView(getHelixClusterName(),
+ TableNameBuilder.REALTIME.tableNameWithType(getTableName()));
+
+ boolean isOffline = false;
+ for (String segmentNameStr : ev.getPartitionSet()) {
+ if (LLCSegmentName.isLLCSegment(segmentNameStr)) {
+ LLCSegmentName segmentName = new LLCSegmentName(segmentNameStr);
+ if (segmentName.getSequenceNumber() == seqNum && segmentName.getPartitionGroupId() == partition
+ && ev.getStateMap(segmentNameStr).values().contains(
+ CommonConstants.Helix.StateModel.SegmentStateModel.OFFLINE)) {
+ isOffline = true;
+ }
+ }
+ }
+ return isOffline;
+ }
+
+ @Override
+ protected Map<String, String> getStreamConfigMap() {
+ Map<String, String> streamConfigMap = super.getStreamConfigMap();
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(
+ streamConfigMap.get(StreamConfigProperties.STREAM_TYPE),
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), ExceptingKafka3ConsumerFactory.class.getName());
+ ExceptingKafka3ConsumerFactory.init(getHelixClusterName(), _helixAdmin, getTableName());
+ return streamConfigMap;
+ }
+ @Override
Review Comment:
Nit: newline
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeKafka3ClusterIntegrationTest.java:
##########
@@ -0,0 +1,528 @@
+/**
+ * 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.integration.tests;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.FileUtils;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.common.utils.HashUtil;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch;
+import org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory;
+import org.apache.pinot.plugin.stream.kafka30.KafkaPartitionLevelConsumer;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+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.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.junit.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Integration test for low-level Kafka3 consumer.
+ */
+public class LLCRealtimeKafka3ClusterIntegrationTest extends BaseRealtimeClusterIntegrationTest {
+ private static final String CONSUMER_DIRECTORY = "/tmp/consumer-test";
+ private static final long RANDOM_SEED = System.currentTimeMillis();
+ private static final Random RANDOM = new Random(RANDOM_SEED);
+
+ private final boolean _isDirectAlloc = RANDOM.nextBoolean();
+ private final boolean _isConsumerDirConfigured = RANDOM.nextBoolean();
+ private final boolean _enableLeadControllerResource = RANDOM.nextBoolean();
+ private final long _startTime = System.currentTimeMillis();
+
+ @Override
+ protected boolean injectTombstones() {
+ return true;
+ }
+
+ @Override
+ protected String getLoadMode() {
+ return ReadMode.mmap.name();
+ }
+
+ @Override
+ public void startController()
+ throws Exception {
+ super.startController();
+ enableResourceConfigForLeadControllerResource(_enableLeadControllerResource);
+ }
+
+ @Override
+ protected void overrideServerConf(PinotConfiguration configuration) {
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_REALTIME_OFFHEAP_ALLOCATION, true);
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_REALTIME_OFFHEAP_DIRECT_ALLOCATION, _isDirectAlloc);
+ if (_isConsumerDirConfigured) {
+ configuration.setProperty(CommonConstants.Server.CONFIG_OF_CONSUMER_DIR, CONSUMER_DIRECTORY);
+ }
+ }
+
+ @Override
+ protected void overrideControllerConf(Map<String, Object> properties) {
+ // Make sure the realtime segment validation manager does not run by itself, only when we invoke it.
+ properties.put(ControllerConf.ControllerPeriodicTasksConf.REALTIME_SEGMENT_VALIDATION_FREQUENCY_PERIOD, "2h");
+ properties.put(ControllerConf.ControllerPeriodicTasksConf.REALTIME_SEGMENT_VALIDATION_INITIAL_DELAY_IN_SECONDS,
+ 3600);
+ }
+
+ @Override
+ protected void runValidationJob(long timeoutMs)
+ throws Exception {
+ final int partition = ExceptingKafka3ConsumerFactory.PARTITION_FOR_EXCEPTIONS;
+ if (partition < 0) {
+ return;
+ }
+ int[] seqNumbers = {
+ ExceptingKafka3ConsumerFactory.SEQ_NUM_FOR_CREATE_EXCEPTION,
+ ExceptingKafka3ConsumerFactory.SEQ_NUM_FOR_CONSUME_EXCEPTION};
+ Arrays.sort(seqNumbers);
+ for (int seqNum : seqNumbers) {
+ if (seqNum < 0) {
+ continue;
+ }
+ TestUtils.waitForCondition(() -> isOffline(partition, seqNum), 5000L, timeoutMs,
+ "Failed to find offline segment in partition " + partition + " seqNum ", true,
+ Duration.ofMillis(timeoutMs / 10));
+ getControllerRequestClient().runPeriodicTask("RealtimeSegmentValidationManager");
+ }
+ }
+
+ private boolean isOffline(int partition, int seqNum) {
+ ExternalView ev = _helixAdmin.getResourceExternalView(getHelixClusterName(),
+ TableNameBuilder.REALTIME.tableNameWithType(getTableName()));
+
+ boolean isOffline = false;
+ for (String segmentNameStr : ev.getPartitionSet()) {
+ if (LLCSegmentName.isLLCSegment(segmentNameStr)) {
+ LLCSegmentName segmentName = new LLCSegmentName(segmentNameStr);
+ if (segmentName.getSequenceNumber() == seqNum && segmentName.getPartitionGroupId() == partition
+ && ev.getStateMap(segmentNameStr).values().contains(
+ CommonConstants.Helix.StateModel.SegmentStateModel.OFFLINE)) {
+ isOffline = true;
+ }
+ }
+ }
+ return isOffline;
+ }
+
+ @Override
+ protected Map<String, String> getStreamConfigMap() {
+ Map<String, String> streamConfigMap = super.getStreamConfigMap();
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(
+ streamConfigMap.get(StreamConfigProperties.STREAM_TYPE),
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), ExceptingKafka3ConsumerFactory.class.getName());
+ ExceptingKafka3ConsumerFactory.init(getHelixClusterName(), _helixAdmin, getTableName());
+ return streamConfigMap;
+ }
+ @Override
+ protected IngestionConfig getIngestionConfig() {
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ ingestionConfig.setStreamIngestionConfig(
+ new StreamIngestionConfig(Collections.singletonList(getStreamConfigMap())));
+ return ingestionConfig;
+ }
+
+ @Override
+ protected Map<String, String> getStreamConfigs() {
+ return null;
+ }
+
+ @Override
+ protected void createSegmentsAndUpload(List<File> avroFiles, Schema schema, TableConfig tableConfig)
+ throws Exception {
+ if (!_tarDir.exists()) {
+ _tarDir.mkdir();
+ }
+ if (!_segmentDir.exists()) {
+ _segmentDir.mkdir();
+ }
+
+ // create segments out of the avro files (segments will be placed in _tarDir)
+ List<File> copyOfAvroFiles = new ArrayList<>(avroFiles);
+ ClusterIntegrationTestUtils.buildSegmentsFromAvro(copyOfAvroFiles, tableConfig, schema, 0, _segmentDir, _tarDir);
+
+ // upload segments to controller
+ uploadSegmentsToController(getTableName(), _tarDir, false, false);
+
+ // upload the first segment again to verify refresh
+ uploadSegmentsToController(getTableName(), _tarDir, true, false);
+
+ // upload the first segment again to verify refresh with different segment crc
+ uploadSegmentsToController(getTableName(), _tarDir, true, true);
+
+ // add avro files to the original list so H2 will have the uploaded data as well
+ avroFiles.addAll(copyOfAvroFiles);
+ }
+
+ private void uploadSegmentsToController(String tableName, File tarDir, boolean onlyFirstSegment, boolean changeCrc)
+ throws Exception {
+ File[] segmentTarFiles = tarDir.listFiles();
+ assertNotNull(segmentTarFiles);
+ int numSegments = segmentTarFiles.length;
+ assertTrue(numSegments > 0);
+ if (onlyFirstSegment) {
+ numSegments = 1;
+ }
+ URI uploadSegmentHttpURI = URI.create(getControllerRequestURLBuilder().forSegmentUpload());
+ try (FileUploadDownloadClient fileUploadDownloadClient = new FileUploadDownloadClient()) {
+ if (numSegments == 1) {
+ File segmentTarFile = segmentTarFiles[0];
+ if (changeCrc) {
+ changeCrcInSegmentZKMetadata(tableName, segmentTarFile.toString());
+ }
+ assertEquals(
+ fileUploadDownloadClient.uploadSegment(uploadSegmentHttpURI, segmentTarFile.getName(), segmentTarFile,
+ tableName, TableType.REALTIME).getStatusCode(), HttpStatus.SC_OK);
+ } else {
+ // Upload segments in parallel
+ ExecutorService executorService = Executors.newFixedThreadPool(numSegments);
+ List<Future<Integer>> futures = new ArrayList<>(numSegments);
+ for (File segmentTarFile : segmentTarFiles) {
+ futures.add(executorService.submit(
+ () -> fileUploadDownloadClient.uploadSegment(uploadSegmentHttpURI, segmentTarFile.getName(),
+ segmentTarFile, tableName, TableType.REALTIME).getStatusCode()));
+ }
+ executorService.shutdown();
+ for (Future<Integer> future : futures) {
+ assertEquals((int) future.get(), HttpStatus.SC_OK);
+ }
+ }
+ }
+ }
+
+ private void changeCrcInSegmentZKMetadata(String tableName, String segmentFilePath) {
+ int startIdx = segmentFilePath.indexOf("mytable_");
+ int endIdx = segmentFilePath.indexOf(".tar.gz");
+ String segmentName = segmentFilePath.substring(startIdx, endIdx);
+ String tableNameWithType = TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(tableName);
+ SegmentZKMetadata segmentZKMetadata = _helixResourceManager.getSegmentZKMetadata(tableNameWithType, segmentName);
+ segmentZKMetadata.setCrc(111L);
+ _helixResourceManager.updateZkMetadata(tableNameWithType, segmentZKMetadata);
+ }
+
+ @Override
+ protected long getCountStarResult() {
+ // all the data that was ingested from Kafka also got uploaded via the controller's upload endpoint
+ return super.getCountStarResult() * 2;
+ }
+
+ @BeforeClass
+ @Override
+ public void setUp()
+ throws Exception {
+ System.out.println(String.format(
+ "Using random seed: %s, isDirectAlloc: %s, isConsumerDirConfigured: %s, enableLeadControllerResource: %s",
+ RANDOM_SEED, _isDirectAlloc, _isConsumerDirConfigured, _enableLeadControllerResource));
Review Comment:
Are those configs per server basis or table basis? If possible we can create 8 tables to cover the situation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #13891:
URL: https://github.com/apache/pinot/pull/13891
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1739941946
##########
pinot-plugins/pinot-input-format/pinot-confluent-avro/pom.xml:
##########
@@ -49,10 +49,17 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
+ <version>${kafka2.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-client</artifactId>
+ <exclusions>
Review Comment:
For fixing the dependency convergence issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1757351396
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.compat.version}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.101tec</groupId>
+ <artifactId>zkclient</artifactId>
+ </dependency>
+
+
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
Review Comment:
Changed, as per suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1757348853
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.compat.version}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.101tec</groupId>
Review Comment:
It was getting used in `EmbeddedZooKeeper`; since it's now in the Kafka-base module, we don't need it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on code in PR #13891:
URL: https://github.com/apache/pinot/pull/13891#discussion_r1757350141
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/pom.xml:
##########
@@ -0,0 +1,92 @@
+<?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.3.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>pinot-kafka-3.0</artifactId>
+ <name>Pinot Kafka 3.x</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../../..</pinot.root>
+ <shade.phase.prop>package</shade.phase.prop>
+ <kafka.lib.version>3.8.0</kafka.lib.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-kafka-base</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.lib.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.compat.version}</artifactId>
+ <version>${kafka.lib.version}</version>
+ <exclusions>
+ <exclusion>
Review Comment:
We need to exclude otherwise we are getting dependency conflict with Kafka version 2.8.0 coming from root pom.xml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #13891:
URL: https://github.com/apache/pinot/pull/13891#issuecomment-2310908121
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/13891?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `0%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 0.00%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`d7894f1`)](https://app.codecov.io/gh/apache/pinot/commit/d7894f1ad5b5cc4414dad32ceb12765f5f9d880c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 937 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/pinot/pull/13891?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ava/org/apache/pinot/spi/plugin/PluginManager.java](https://app.codecov.io/gh/apache/pinot/pull/13891?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fplugin%2FPluginManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvcGx1Z2luL1BsdWdpbk1hbmFnZXIuamF2YQ==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/13891?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (d7894f1). Click for more details.
>
> <details><summary>HEAD has 53 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (d7894f1) |
>|------|------|------|
>|integration|7|1|
>|integration2|3|1|
>|temurin|12|1|
>|java-21|7|1|
>|skip-bytebuffers-true|3|0|
>|skip-bytebuffers-false|7|1|
>|unittests|5|0|
>|unittests1|2|0|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #13891 +/- ##
=============================================
- Coverage 61.75% 0.00% -61.76%
=============================================
Files 2436 2505 +69
Lines 133233 138776 +5543
Branches 20636 21561 +925
=============================================
- Hits 82274 0 -82274
- Misses 44911 138776 +93865
+ Partials 6048 0 -6048
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-0.01%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (ø)` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-61.63%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-61.75%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-61.76%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/13891/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/13891?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Kafka3 support [pinot]
Posted by "abhioncbr (via GitHub)" <gi...@apache.org>.
abhioncbr commented on PR #13891:
URL: https://github.com/apache/pinot/pull/13891#issuecomment-2334255952
> This is awesome! Is Kafka 3 consumer backward compatible with Kafka 2 producer? Wondering how to migrate from Kafka 2 to 3. Can you add an integration test on Kafka 3?
I added the integration test based on Kafka3 consumers. Based on the integration test, migration from Kafka 2 to 3 should be easy and look like a drop-in replacement type, but I can't confirm it. Let me know if you want some tests around migration as well. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org