This is an automated email from the ASF dual-hosted git repository.
stoty pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new dd0be98 PHOENIX-6935 Remove Phoenix Kafka connector
dd0be98 is described below
commit dd0be98983bf3e6bb9ec47f5bebe2dee203d88cd
Author: Aron Meszaros <meszaros.aron.att...@gmail.com>
AuthorDate: Mon Oct 9 09:52:40 2023 +0200
PHOENIX-6935 Remove Phoenix Kafka connector
---
README.md | 2 +-
phoenix5-connectors-assembly/pom.xml | 24 --
.../src/build/components/phoenix5-jars.xml | 8 -
phoenix5-kafka/pom.xml | 236 -----------------
.../apache/phoenix/kafka/PhoenixConsumerIT.java | 293 ---------------------
phoenix5-kafka/src/it/resources/consumer.props | 32 ---
phoenix5-kafka/src/it/resources/producer.props | 24 --
.../org/apache/phoenix/kafka/KafkaConstants.java | 52 ----
.../phoenix/kafka/consumer/PhoenixConsumer.java | 292 --------------------
.../kafka/consumer/PhoenixConsumerTool.java | 107 --------
pom.xml | 9 -
11 files changed, 1 insertion(+), 1078 deletions(-)
diff --git a/README.md b/README.md
index c7487ff..fec24b6 100644
--- a/README.md
+++ b/README.md
@@ -22,4 +22,4 @@ limitations under the License.
Copyright ©2019 [Apache Software Foundation](http://www.apache.org/). All
Rights Reserved.
## Introduction
-This repo contains the Flume, Kafka, Spark and Hive connectors for Phoenix.
\ No newline at end of file
+This repo contains the Flume, Spark and Hive connectors for Phoenix.
\ No newline at end of file
diff --git a/phoenix5-connectors-assembly/pom.xml
b/phoenix5-connectors-assembly/pom.xml
index c6c4253..917ec5e 100644
--- a/phoenix5-connectors-assembly/pom.xml
+++ b/phoenix5-connectors-assembly/pom.xml
@@ -49,10 +49,6 @@
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix5-hive-shaded</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix5-kafka</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix5-spark</artifactId>
@@ -153,26 +149,6 @@
</arguments>
</configuration>
</execution>
- <execution>
- <id>kafka without version</id>
- <phase>package</phase>
- <goals>
- <goal>exec</goal>
- </goals>
- <configuration>
- <executable>ln</executable>
-
<workingDirectory>${project.basedir}/../phoenix5-kafka/target</workingDirectory>
- <arguments>
- <argument>-fnsv</argument>
- <argument>
- phoenix5-kafka-${project.version}.jar
- </argument>
- <argument>
- phoenix5-kafka.jar
- </argument>
- </arguments>
- </configuration>
- </execution>
</executions>
</plugin>
<plugin>
diff --git
a/phoenix5-connectors-assembly/src/build/components/phoenix5-jars.xml
b/phoenix5-connectors-assembly/src/build/components/phoenix5-jars.xml
index b507a93..589c5d8 100644
--- a/phoenix5-connectors-assembly/src/build/components/phoenix5-jars.xml
+++ b/phoenix5-connectors-assembly/src/build/components/phoenix5-jars.xml
@@ -31,14 +31,6 @@
<include>phoenix5-flume.jar</include>
</includes>
</fileSet>
- <fileSet>
- <directory>${project.basedir}/../phoenix5-kafka/target</directory>
- <outputDirectory>/</outputDirectory>
- <includes>
- <include>phoenix5-kafka-${project.version}.jar</include>
- <include>phoenix5-kafka.jar</include>
- </includes>
- </fileSet>
<fileSet>
<directory>${project.basedir}/../phoenix5-spark-shaded/target</directory>
<outputDirectory>/</outputDirectory>
diff --git a/phoenix5-kafka/pom.xml b/phoenix5-kafka/pom.xml
deleted file mode 100644
index ecf9c7e..0000000
--- a/phoenix5-kafka/pom.xml
+++ /dev/null
@@ -1,236 +0,0 @@
-<?xml version='1.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.
-
--->
-
-<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>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix-connectors</artifactId>
- <version>6.0.0-SNAPSHOT</version>
- </parent>
- <artifactId>phoenix5-kafka</artifactId>
- <name>Phoenix Kafka Connector for Phoenix 5</name>
-
- <properties>
- <top.dir>${project.basedir}/..</top.dir>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-common</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
- <!-- To work with kafka with phoenix -->
- <dependency>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix-hbase-compat-${hbase.compat.version}</artifactId>
- <scope>runtime</scope>
- <optional>true</optional>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>${kafka.version}</version>
- </dependency>
- <dependency>
- <groupId>commons-cli</groupId>
- <artifactId>commons-cli</artifactId>
- <version>${commons-cli.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix5-flume</artifactId>
- </dependency>
- <!-- Test dependencies -->
- <dependency>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix-core</artifactId>
- <classifier>tests</classifier>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.phoenix.thirdparty</groupId>
- <artifactId>phoenix-shaded-guava</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.11</artifactId>
- <version>${kafka.version}</version>
- <classifier>test</classifier>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-it</artifactId>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-minicluster</artifactId>
- <scope>test</scope>
- </dependency>
-
- <!-- Flume dependencies, as the Kafka connectors extends Flume. -->
-<!-- <dependency> -->
-<!-- <groupId>org.apache.flume</groupId> -->
-<!-- <artifactId>flume-ng-core</artifactId> -->
-<!-- </dependency> -->
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-sdk</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-configuration</artifactId>
- </dependency>
- </dependencies>
-
- <build>
- <pluginManagement>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-dependency-plugin</artifactId>
- <configuration>
- <ignoreNonCompile>true</ignoreNonCompile>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <version>3.0.0</version>
- <executions>
- <execution>
- <id>add-parent-source</id>
- <phase>generate-sources</phase>
- <goals>
- <goal>add-source</goal>
- </goals>
- <configuration>
- <sources>
- <source>${project.parent.basedir}/src/main/java</source>
- </sources>
- </configuration>
- </execution>
- <execution>
- <id>add-parent-test-source</id>
- <phase>generate-sources</phase>
- <goals>
- <goal>add-test-source</goal>
- </goals>
- <configuration>
- <sources>
- <source>${project.parent.basedir}/src/it/java</source>
- <source>${project.parent.basedir}/src/it/resources</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-failsafe-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-site-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-resources-plugin</artifactId>
- <executions>
- <execution>
- <id>copy-resources</id>
- <phase>generate-resources</phase>
- <goals>
- <goal>copy-resources</goal>
- </goals>
- <configuration>
- <outputDirectory>${project.build.directory}/test-classes
- </outputDirectory>
- <overwrite>true</overwrite>
- <resources>
- <resource>
-
<directory>${project.parent.basedir}/src/it/resources</directory>
- </resource>
- </resources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <configuration>
- <shadedArtifactAttached>true</shadedArtifactAttached>
- <promoteTransitiveDependencies>true</promoteTransitiveDependencies>
- <shadedClassifierName>shaded-minimal</shadedClassifierName>
- <shadeTestJar>false</shadeTestJar>
- <artifactSet>
- <includes>
- <include>org.apache.phoenix:phoenix5-kafka</include>
- <include>org.apache.kafka:kafka-clients</include>
- <include>org.apache.phoenix:phoenix5-flume</include>
- </includes>
- </artifactSet>
- </configuration>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </pluginManagement>
- <plugins>
- <plugin>
- <!-- Allows us to get the apache-ds bundle artifacts -->
- <groupId>org.apache.felix</groupId>
- <artifactId>maven-bundle-plugin</artifactId>
- <extensions>true</extensions>
- <inherited>true</inherited>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git
a/phoenix5-kafka/src/it/java/org/apache/phoenix/kafka/PhoenixConsumerIT.java
b/phoenix5-kafka/src/it/java/org/apache/phoenix/kafka/PhoenixConsumerIT.java
deleted file mode 100644
index cb41008..0000000
--- a/phoenix5-kafka/src/it/java/org/apache/phoenix/kafka/PhoenixConsumerIT.java
+++ /dev/null
@@ -1,293 +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.phoenix.kafka;
-
-import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.Map;
-import java.util.Properties;
-
-import org.I0Itec.zkclient.ZkClient;
-import org.apache.flume.Context;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
-import org.apache.phoenix.flume.DefaultKeyGenerator;
-import org.apache.phoenix.flume.FlumeConstants;
-import org.apache.phoenix.flume.serializer.EventSerializers;
-import org.apache.phoenix.kafka.consumer.PhoenixConsumer;
-import org.apache.phoenix.util.PropertiesUtil;
-import org.apache.phoenix.query.BaseTest;
-import org.apache.phoenix.query.QueryServices;
-import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.util.ReadOnlyProps;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-
-import kafka.admin.AdminUtils;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import kafka.utils.MockTime;
-import kafka.utils.TestUtils;
-import kafka.utils.Time;
-import kafka.utils.ZKStringSerializer$;
-import kafka.utils.ZkUtils;
-import kafka.zk.EmbeddedZookeeper;
-
-@Category(NeedsOwnMiniClusterTest.class)
-public class PhoenixConsumerIT extends BaseTest {
- private static final String ZKHOST = "127.0.0.1";
- private static final String BROKERHOST = "127.0.0.1";
- private static final String BROKERPORT = "9092";
- private static final String TOPIC = "topic1";
- private KafkaServer kafkaServer;
- private PhoenixConsumer pConsumer;
- private EmbeddedZookeeper zkServer;
- private ZkClient zkClient;
- private Connection conn;
-
- @BeforeClass
- public static void doSetup() throws Exception {
- Map<String,String> props = Maps.newHashMapWithExpectedSize(3);
- props.put(QueryServices.EXTRA_JDBC_ARGUMENTS_ATTRIB,
QueryServicesOptions.DEFAULT_EXTRA_JDBC_ARGUMENTS);
- // Must update config before starting server
- setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
- }
-
- @Before
- public void setUp() throws Exception {
- setUpTestDriver(ReadOnlyProps.EMPTY_PROPS);
- // setup Zookeeper
- zkServer = new EmbeddedZookeeper();
- String zkConnect = ZKHOST + ":" + zkServer.port();
- zkClient = new ZkClient(zkConnect, 30000, 30000,
ZKStringSerializer$.MODULE$);
- ZkUtils zkUtils = ZkUtils.apply(zkClient, false);
-
- // setup Broker
- Properties brokerProps = new Properties();
- brokerProps.setProperty("zookeeper.connect", zkConnect);
- brokerProps.setProperty("broker.id", "0");
- brokerProps.setProperty("log.dirs",
- Files.createTempDirectory("kafka-").toAbsolutePath().toString());
- brokerProps.setProperty("listeners", "PLAINTEXT://" + BROKERHOST + ":"
+ BROKERPORT);
- KafkaConfig config = new KafkaConfig(brokerProps);
- Time mock = new MockTime();
- kafkaServer = TestUtils.createServer(config, mock);
- kafkaServer.startup();
-
- // create topic
- AdminUtils.createTopic(zkUtils, TOPIC, 1, 1, new Properties());
-
- pConsumer = new PhoenixConsumer();
-
- Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
- conn = DriverManager.getConnection(getUrl(), props);
- }
-
- @Test
- public void testPhoenixConsumerWithFile() throws SQLException {
- String consumerPath = "consumer.props";
- PhoenixConsumerThread pConsumerThread = new
PhoenixConsumerThread(pConsumer, consumerPath);
- pConsumerThread.properties.setProperty(FlumeConstants.CONFIG_JDBC_URL,
getUrl());
- Thread phoenixConsumer = new Thread(pConsumerThread);
-
- String producerPath = "producer.props";
- KafkaProducerThread kProducerThread = new
KafkaProducerThread(producerPath, TOPIC);
- Thread kafkaProducer = new Thread(kProducerThread);
-
- phoenixConsumer.start();
-
- try {
- phoenixConsumer.join(10000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- kafkaProducer.start();
-
- try {
- kafkaProducer.join();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- if (!kafkaProducer.isAlive()) {
- System.out.println("kafka producer is not alive");
- pConsumer.stop();
- }
-
- // Verify our serializer wrote out data
- ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM
SAMPLE1");
- assertTrue(rs.next());
- assertTrue(rs.getFetchSize() > 0);
- rs.close();
- }
-
- @Test
- public void testPhoenixConsumerWithProperties() throws SQLException {
-
- final String fullTableName = "SAMPLE2";
- final String ddl = "CREATE TABLE IF NOT EXISTS SAMPLE2(uid VARCHAR NOT
NULL,c1 VARCHAR,c2 VARCHAR,c3 VARCHAR CONSTRAINT pk PRIMARY KEY(uid))\n";
-
- Properties consumerProperties = new Properties();
- consumerProperties.setProperty(FlumeConstants.CONFIG_TABLE,
fullTableName);
- consumerProperties.setProperty(FlumeConstants.CONFIG_JDBC_URL,
getUrl());
-
consumerProperties.setProperty(FlumeConstants.CONFIG_SERIALIZER,EventSerializers.REGEX.name());
- consumerProperties.setProperty(FlumeConstants.CONFIG_TABLE_DDL, ddl);
- consumerProperties.setProperty(FlumeConstants.CONFIG_SERIALIZER_PREFIX
+ FlumeConstants.CONFIG_REGULAR_EXPRESSION,"([^\\,]*),([^\\,]*),([^\\,]*)");
- consumerProperties.setProperty(FlumeConstants.CONFIG_SERIALIZER_PREFIX
+ FlumeConstants.CONFIG_COLUMN_NAMES,"c1,c2,c3");
- consumerProperties.setProperty(FlumeConstants.CONFIG_SERIALIZER_PREFIX
+ FlumeConstants.CONFIG_ROWKEY_TYPE_GENERATOR, DefaultKeyGenerator.UUID.name());
- consumerProperties.setProperty(KafkaConstants.BOOTSTRAP_SERVERS,
"localhost:9092");
- consumerProperties.setProperty(KafkaConstants.TOPICS, "topic1,topic2");
- consumerProperties.setProperty(KafkaConstants.TIMEOUT, "100");
-
- PhoenixConsumerThread pConsumerThread = new
PhoenixConsumerThread(pConsumer, consumerProperties);
- Thread phoenixConsumer = new Thread(pConsumerThread);
-
- Properties producerProperties = new Properties();
- producerProperties.setProperty(KafkaConstants.BOOTSTRAP_SERVERS,
"localhost:9092");
- producerProperties.setProperty(KafkaConstants.KEY_SERIALIZER,
KafkaConstants.DEFAULT_KEY_SERIALIZER);
- producerProperties.setProperty(KafkaConstants.VALUE_SERIALIZER,
KafkaConstants.DEFAULT_VALUE_SERIALIZER);
- producerProperties.setProperty("auto.commit.interval.ms", "1000");
-
- KafkaProducerThread kProducerThread = new
KafkaProducerThread(producerProperties, TOPIC);
- Thread kafkaProducer = new Thread(kProducerThread);
-
- phoenixConsumer.start();
-
- try {
- phoenixConsumer.join(10000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- kafkaProducer.start();
-
- try {
- kafkaProducer.join();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- if (!kafkaProducer.isAlive()) {
- System.out.println("kafka producer is not alive");
- pConsumer.stop();
- }
-
- // Verify our serializer wrote out data
- ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM
SAMPLE2");
- assertTrue(rs.next());
- assertTrue(rs.getFetchSize() > 0);
- rs.close();
- }
-
- @After
- public void cleanUp() throws Exception {
- kafkaServer.shutdown();
- zkClient.close();
- zkServer.shutdown();
- conn.close();
- }
-
- class PhoenixConsumerThread implements Runnable {
- PhoenixConsumer pConsumer;
- Properties properties;
-
- PhoenixConsumerThread(PhoenixConsumer pConsumer, String path) {
- this.pConsumer = pConsumer;
- try (InputStream props =
getClass().getClassLoader().getResourceAsStream(path)) {
- Properties properties = new Properties();
- properties.load(props);
- this.properties = properties;
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- PhoenixConsumerThread(PhoenixConsumer pConsumer, Properties
properties) {
- this.pConsumer = pConsumer;
- this.properties = properties;
- }
-
- @Override
- public void run() {
- // intialize the kafka
- pConsumer.intializeKafka(properties);
-
- // configure the phoenix
- Context context = pConsumer.prepareContext();
- pConsumer.configure(context);
-
- // start the kafka consumer
- pConsumer.start();
-
- // process kafka messages
- pConsumer.process();
- }
- }
-
- class KafkaProducerThread implements Runnable {
- KafkaProducer<String, String> producer;
- String topic;
-
- KafkaProducerThread(String path, String topic) {
- this.topic = topic;
- try (InputStream props =
getClass().getClassLoader().getResourceAsStream(path)) {
- Properties properties = new Properties();
- properties.load(props);
- producer = new KafkaProducer<>(properties);
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- KafkaProducerThread(Properties properties, String topic) {
- this.topic = topic;
- producer = new KafkaProducer<>(properties);
- }
-
- @Override
- public void run() {
- try {
- for (int i = 1; i <= 10; i++) {
- String message = String.format("%s,%.3f,%d", "msg" + i, i
* 2000f, i);
- producer.send(new ProducerRecord<String, String>(topic,
message));
- producer.flush();
- Thread.sleep(100);
- }
- } catch (Throwable throwable) {
- System.out.printf("%s", throwable.fillInStackTrace());
- } finally {
- producer.close();
- }
- }
- }
-}
diff --git a/phoenix5-kafka/src/it/resources/consumer.props
b/phoenix5-kafka/src/it/resources/consumer.props
deleted file mode 100644
index 703fd7c..0000000
--- a/phoenix5-kafka/src/it/resources/consumer.props
+++ /dev/null
@@ -1,32 +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.
-#
-############################################################################
-
-serializer=regex
-serializer.rowkeyType=uuid
-serializer.regex=([^\,]*),([^\,]*),([^\,]*)
-serializer.columns=c1,c2,c3
-
-jdbcUrl=jdbc:phoenix:localhost
-table=SAMPLE1
-ddl=CREATE TABLE IF NOT EXISTS SAMPLE1(uid VARCHAR NOT NULL,c1 VARCHAR,c2
VARCHAR,c3 VARCHAR CONSTRAINT pk PRIMARY KEY(uid))
-
-bootstrap.servers=localhost:9092
-topics=topic1,topic2
-poll.timeout.ms=100
diff --git a/phoenix5-kafka/src/it/resources/producer.props
b/phoenix5-kafka/src/it/resources/producer.props
deleted file mode 100644
index 4c3cd2f..0000000
--- a/phoenix5-kafka/src/it/resources/producer.props
+++ /dev/null
@@ -1,24 +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.
-#
-############################################################################
-
-bootstrap.servers=localhost:9092
-auto.commit.interval.ms=1000
-key.serializer=org.apache.kafka.common.serialization.StringSerializer
-value.serializer=org.apache.kafka.common.serialization.StringSerializer
diff --git
a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/KafkaConstants.java
b/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/KafkaConstants.java
deleted file mode 100644
index cc1aa61..0000000
--- a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/KafkaConstants.java
+++ /dev/null
@@ -1,52 +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.phoenix.kafka;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.serialization.StringSerializer;
-
-public final class KafkaConstants {
-
- public static final String BOOTSTRAP_SERVERS =
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
-
- public static final String KEY_SERIALIZER =
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
-
- public static final String VALUE_SERIALIZER =
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
-
- public static final String DEFAULT_KEY_SERIALIZER =
StringSerializer.class.getName();
-
- public static final String DEFAULT_VALUE_SERIALIZER =
StringSerializer.class.getName();
-
- public static final String KEY_DESERIALIZER =
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
-
- public static final String VALUE_DESERIALIZER =
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
-
- public static final String DEFAULT_KEY_DESERIALIZER =
StringDeserializer.class.getName();
-
- public static final String DEFAULT_VALUE_DESERIALIZER =
StringDeserializer.class.getName();
-
- public static final String TOPICS = "topics";
-
- public static final String GROUP_ID = ConsumerConfig.GROUP_ID_CONFIG;
-
- public static final String TIMEOUT = "poll.timeout.ms";
-
- public static final long DEFAULT_TIMEOUT = 100;
-}
diff --git
a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumer.java
b/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumer.java
deleted file mode 100644
index 6551386..0000000
---
a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumer.java
+++ /dev/null
@@ -1,292 +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.phoenix.kafka.consumer;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Random;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.event.EventBuilder;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.phoenix.flume.FlumeConstants;
-import org.apache.phoenix.flume.serializer.EventSerializer;
-import org.apache.phoenix.flume.serializer.EventSerializers;
-import org.apache.phoenix.kafka.KafkaConstants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class PhoenixConsumer {
- private static final Logger logger =
LoggerFactory.getLogger(PhoenixConsumer.class);
-
- private KafkaConsumer<String, String> consumer = null;
- private Properties properties = new Properties();
- private Integer batchSize;
- private long timeout;
- private EventSerializer serializer;
- private Boolean process = true;
-
- public PhoenixConsumer() {
-
- }
-
- public PhoenixConsumer(Configuration conf) throws IOException {
- // intialize the kafka
- intializeKafka(conf);
-
- // configure the phoenix
- Context context = prepareContext();
- configure(context);
-
- // start the kafka consumer
- start();
-
- // process kafka messages
- process();
- }
-
- /**
- * Initializes the kafka with properties file.
- * @param conf
- * @throws IOException
- */
- public void intializeKafka(Configuration conf) throws IOException {
- // get the kafka consumer file
- String file = conf.get("kafka.consumer.file");
- if(file==null){
- throw new NullPointerException("File path cannot be empty, please
specify in the arguments");
- }
-
- Path path = new Path(file);
- FileSystem fs = FileSystem.get(conf);
- try (InputStream props = fs.open(path)) {
- properties.load(props);
- } catch (IOException e) {
- e.printStackTrace();
- }
-
- intializeKafka(properties);
- }
-
- /**
- * Initializes the kafka with properties.
- * @param properties
- */
- public void intializeKafka(Properties properties) {
- this.properties = properties;
-
- String servers =
properties.getProperty(KafkaConstants.BOOTSTRAP_SERVERS);
- if(servers ==null){
- throw new NullPointerException("Bootstrap Servers cannot be empty,
please specify in the configuration file");
- }
- properties.setProperty(KafkaConstants.BOOTSTRAP_SERVERS, servers);
-
- if (properties.getProperty(KafkaConstants.GROUP_ID) == null) {
- properties.setProperty(KafkaConstants.GROUP_ID, "group-" + new
Random().nextInt(100000));
- }
-
- if (properties.getProperty(KafkaConstants.TIMEOUT) == null) {
- properties.setProperty(KafkaConstants.TIMEOUT,
String.valueOf(KafkaConstants.DEFAULT_TIMEOUT));
- }
-
- String topics = properties.getProperty(KafkaConstants.TOPICS);
-
- if (topics == null) {
- throw new NullPointerException("Topics cannot be empty, please
specify in the configuration file");
- }
-
- properties.setProperty(KafkaConstants.KEY_DESERIALIZER,
KafkaConstants.DEFAULT_KEY_DESERIALIZER);
-
- properties.setProperty(KafkaConstants.VALUE_DESERIALIZER,
KafkaConstants.DEFAULT_VALUE_DESERIALIZER);
-
- this.consumer = new KafkaConsumer<>(properties);
- consumer.subscribe(Arrays.asList(topics.split(",")));
- }
-
- /**
- * Convert the properties to context
- */
- public Context prepareContext() {
- Map<String, String> map = new HashMap<String, String>();
- for (Entry<Object, Object> entry : properties.entrySet()) {
- map.put((String) entry.getKey(), (String) entry.getValue());
- }
- return new Context(map);
- }
-
- /**
- * Configure the context
- */
- public void configure(Context context){
- this.timeout = context.getLong(KafkaConstants.TIMEOUT,
KafkaConstants.DEFAULT_TIMEOUT);
- this.batchSize = context.getInteger(FlumeConstants.CONFIG_BATCHSIZE,
FlumeConstants.DEFAULT_BATCH_SIZE);
- final String eventSerializerType =
context.getString(FlumeConstants.CONFIG_SERIALIZER);
- if (eventSerializerType ==null){
- throw new NullPointerException("Event serializer cannot be empty,
please specify in the configuration file");
- }
- initializeSerializer(context,eventSerializerType);
- }
-
- /**
- * Process the kafka messages
- */
- public void process() {
- int timeouts = 0;
- // noinspection InfiniteLoopStatement
- while (process) {
- // read records with a short timeout.
- // If we time out, we don't really care.
- // Assuming only key & value text data
- ConsumerRecords<String, String> records =
consumer.poll(this.timeout);
- if (records.count() == 0) {
- timeouts++;
- } else {
- System.out.printf("Got %d records after %d timeouts\n",
records.count(), timeouts);
- timeouts = 0;
- }
-
- if (!records.isEmpty()) {
- List<Event> events = new ArrayList<>(records.count());
- for (ConsumerRecord<String, String> record : records) {
- Event event =
EventBuilder.withBody(Bytes.toBytes(record.value()));
- events.add(event);
- }
- // save to Hbase
- try {
- serializer.upsertEvents(events);
- } catch (SQLException e) {
- e.printStackTrace();
- }
- }
- }
- }
-
- /**
- * start the serializer
- */
- public void start() {
- logger.info("Starting consumer {} ", this.getClass());
- try {
- serializer.initialize();
- } catch (Exception ex) {
- logger.error("Error {} in initializing the serializer.",
ex.getMessage());
- if (ex instanceof RuntimeException){
- throw RuntimeException.class.cast(ex);
- }
- else {
- throw new RuntimeException(ex);
- }
- }
- }
-
- /**
- * stop the consumer and serializer
- */
- public void stop() {
- this.close();
- consumer.close();
- try {
- serializer.close();
- } catch (SQLException e) {
- logger.error(" Error while closing connection {} for consumer.",
e.getMessage());
- }
- }
-
- /**
- * make the changes to stop in gracefully
- */
- public void close(){
- this.process = false;
- try {
- Thread.sleep(30000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
-
- /**
- * Initializes the serializer for kafka messages.
- * @param context
- * @param eventSerializerType
- */
- private void initializeSerializer(final Context context, final String
eventSerializerType) {
- String serializerClazz = null;
- EventSerializers eventSerializer = null;
-
- try {
- eventSerializer =
EventSerializers.valueOf(eventSerializerType.toUpperCase());
- } catch (IllegalArgumentException iae) {
- serializerClazz = eventSerializerType;
- }
-
- final Context serializerContext = new Context();
-
serializerContext.putAll(context.getSubProperties(FlumeConstants.CONFIG_SERIALIZER_PREFIX));
- copyPropertiesToSerializerContext(context,serializerContext);
-
- try {
- @SuppressWarnings("unchecked")
- Class<? extends EventSerializer> clazz = null;
- if (serializerClazz == null) {
- clazz = (Class<? extends EventSerializer>)
Class.forName(eventSerializer.getClassName());
- } else {
- clazz = (Class<? extends EventSerializer>)
Class.forName(serializerClazz);
- }
-
- serializer = clazz.newInstance();
- serializer.configure(serializerContext);
- } catch (Exception e) {
- logger.error("Could not instantiate event serializer.", e);
- if (e instanceof RuntimeException){
- throw (RuntimeException)e;
- }
- else {
- throw new RuntimeException(e);
- }
- }
- }
-
- /**
- * Copy properties to serializer context.
- * @param context
- * @param serializerContext
- */
- private void copyPropertiesToSerializerContext(Context context, Context
serializerContext) {
-
serializerContext.put(FlumeConstants.CONFIG_TABLE_DDL,context.getString(FlumeConstants.CONFIG_TABLE_DDL));
-
serializerContext.put(FlumeConstants.CONFIG_TABLE,context.getString(FlumeConstants.CONFIG_TABLE));
-
serializerContext.put(FlumeConstants.CONFIG_ZK_QUORUM,context.getString(FlumeConstants.CONFIG_ZK_QUORUM));
-
serializerContext.put(FlumeConstants.CONFIG_JDBC_URL,context.getString(FlumeConstants.CONFIG_JDBC_URL));
-
serializerContext.put(FlumeConstants.CONFIG_BATCHSIZE,context.getString(FlumeConstants.CONFIG_BATCHSIZE));
- }
-
-}
diff --git
a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumerTool.java
b/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumerTool.java
deleted file mode 100644
index 8c10aa5..0000000
---
a/phoenix5-kafka/src/main/java/org/apache/phoenix/kafka/consumer/PhoenixConsumerTool.java
+++ /dev/null
@@ -1,107 +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.phoenix.kafka.consumer;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.commons.cli.PosixParser;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PhoenixConsumerTool extends Configured implements Tool {
- private static final Logger logger =
LoggerFactory.getLogger(PhoenixConsumerTool.class);
- static final Option FILE_PATH_OPT = new Option("f", "file", true, "input
file path");
- static final Option HELP_OPT = new Option("h", "help", false, "Show this
help and quit");
-
- public static Options getOptions() {
- Options options = new Options();
- options.addOption(FILE_PATH_OPT);
- options.addOption(HELP_OPT);
- return options;
- }
-
- public static CommandLine parseOptions(String[] args) {
-
- Options options = getOptions();
-
- CommandLineParser parser = new PosixParser();
- CommandLine cmdLine = null;
- try {
- cmdLine = parser.parse(options, args);
- } catch (ParseException e) {
- printHelpAndExit("Error parsing command line options: " +
e.getMessage(), options);
- }
-
- if (cmdLine.hasOption(HELP_OPT.getOpt())) {
- printHelpAndExit(options, 0);
- }
-
- if (!cmdLine.hasOption(FILE_PATH_OPT.getOpt())) {
- throw new IllegalStateException(FILE_PATH_OPT.getLongOpt() + " is
a mandatory " + "parameter");
- }
-
- if (!cmdLine.getArgList().isEmpty()) {
- throw new IllegalStateException("Got unexpected extra parameters:
" + cmdLine.getArgList());
- }
-
- return cmdLine;
- }
-
- public static void printHelpAndExit(String errorMessage, Options options) {
- System.err.println(errorMessage);
- printHelpAndExit(options, 1);
- }
-
- public static void printHelpAndExit(Options options, int exitCode) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("help", options);
- System.exit(exitCode);
- }
-
- @Override
- public int run(String[] args) throws Exception {
- Configuration conf = HBaseConfiguration.create(getConf());
-
- CommandLine cmdLine = null;
- try {
- cmdLine = parseOptions(args);
- } catch (IllegalStateException e) {
- printHelpAndExit(e.getMessage(), getOptions());
- }
-
- String path = cmdLine.getOptionValue(FILE_PATH_OPT.getOpt());
- conf.set("kafka.consumer.file", path);
- new PhoenixConsumer(conf);
-
- return 1;
- }
-
- public static void main(String[] args) throws Exception {
- int exitStatus = ToolRunner.run(new PhoenixConsumerTool(), args);
- System.exit(exitStatus);
- }
-}
diff --git a/pom.xml b/pom.xml
index b4c4cb8..b1a590a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,7 +46,6 @@
<module>phoenix5-hive</module>
<module>phoenix5-hive-shaded</module>
<module>phoenix5-flume</module>
- <module>phoenix5-kafka</module>
<module>phoenix5-spark</module>
<module>phoenix5-spark-shaded</module>
<module>phoenix5-spark3</module>
@@ -100,7 +99,6 @@
<hive3-storage.version>2.7.0</hive3-storage.version>
<hive-storage.version>${hive3-storage.version}</hive-storage.version>
<flume.version>1.4.0</flume.version>
- <kafka.version>0.9.0.0</kafka.version>
<spark.version>2.4.0</spark.version>
<spark3.version>3.0.3</spark3.version>
<scala.version>2.11.12</scala.version>
@@ -118,8 +116,6 @@
<jodatime.version>2.10.5</jodatime.version>
<commons-cli.version>1.4</commons-cli.version>
<commons-compress.version>1.9</commons-compress.version>
- <!-- For Kafka -->
- <com-101tek-zkclient.version>0.7</com-101tek-zkclient.version>
<!-- For hive -->
<commons-io.version>2.11.0</commons-io.version>
@@ -515,11 +511,6 @@
<artifactId>phoenix5-flume</artifactId>
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.phoenix</groupId>
- <artifactId>phoenix5-kafka</artifactId>
- <version>${project.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix5-spark</artifactId>
