You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by rz...@apache.org on 2024/12/16 14:36:51 UTC
(storm) branch STORM-4132---Remove-storm-hive created (now 6bbb90c4e)
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a change to branch STORM-4132---Remove-storm-hive
in repository https://gitbox.apache.org/repos/asf/storm.git
at 6bbb90c4e STORM-4132 - Remove "storm-hive"
This branch includes the following new commits:
new 6bbb90c4e STORM-4132 - Remove "storm-hive"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(storm) 01/01: STORM-4132 - Remove "storm-hive"
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch STORM-4132---Remove-storm-hive
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 6bbb90c4e84b98cff3afe720f4dbaed8b2fd5796
Author: Richard Zowalla <rz...@apache.org>
AuthorDate: Mon Dec 16 15:36:41 2024 +0100
STORM-4132 - Remove "storm-hive"
---
docs/index.md | 1 -
examples/storm-hive-examples/pom.xml | 106 -----
.../storm/hive/bolt/BucketTestHiveTopology.java | 182 -------
.../org/apache/storm/hive/bolt/HiveTopology.java | 147 ------
.../storm/hive/bolt/HiveTopologyPartitioned.java | 148 ------
.../storm/hive/trident/TridentHiveTopology.java | 187 --------
external/storm-autocreds/pom.xml | 48 --
.../org/apache/storm/hive/security/AutoHive.java | 49 --
.../storm/hive/security/AutoHiveCommand.java | 68 ---
.../apache/storm/hive/security/AutoHiveNimbus.java | 260 ----------
.../storm/hive/security/HiveSecurityUtil.java | 34 --
external/storm-hive/README.md | 189 --------
external/storm-hive/pom.xml | 200 --------
.../java/org/apache/storm/hive/bolt/HiveBolt.java | 334 -------------
.../bolt/mapper/DelimitedRecordHiveMapper.java | 142 ------
.../apache/storm/hive/bolt/mapper/HiveMapper.java | 62 ---
.../hive/bolt/mapper/JsonRecordHiveMapper.java | 123 -----
.../org/apache/storm/hive/common/HiveOptions.java | 154 ------
.../org/apache/storm/hive/common/HiveUtils.java | 115 -----
.../org/apache/storm/hive/common/HiveWriter.java | 479 -------------------
.../org/apache/storm/hive/trident/HiveState.java | 304 ------------
.../storm/hive/trident/HiveStateFactory.java | 49 --
.../org/apache/storm/hive/trident/HiveUpdater.java | 26 -
.../org/apache/storm/hive/bolt/HiveSetupUtil.java | 214 ---------
.../org/apache/storm/hive/bolt/TestHiveBolt.java | 527 ---------------------
.../apache/storm/hive/common/TestHiveWriter.java | 207 --------
pom.xml | 3 -
.../final-package/src/main/assembly/binary.xml | 7 -
28 files changed, 4365 deletions(-)
diff --git a/docs/index.md b/docs/index.md
index 7cdc90789..4dd603afb 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -95,7 +95,6 @@ But small change will not affect the user experience. We will notify the user wh
* [Apache Kafka Integration](storm-kafka-client.html)
* [Apache HBase Integration](storm-hbase.html)
* [Apache HDFS Integration](storm-hdfs.html)
-* [Apache Hive Integration](storm-hive.html)
* [JDBC Integration](storm-jdbc.html)
* [JMS Integration](storm-jms.html)
* [Redis Integration](storm-redis.html)
diff --git a/examples/storm-hive-examples/pom.xml b/examples/storm-hive-examples/pom.xml
deleted file mode 100644
index 866d3dae3..000000000
--- a/examples/storm-hive-examples/pom.xml
+++ /dev/null
@@ -1,106 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <artifactId>storm</artifactId>
- <groupId>org.apache.storm</groupId>
- <version>2.8.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <artifactId>storm-hive-examples</artifactId>
-
- <properties>
- <!-- Required downgrade by hive-hcatalog-core 2.3.4 -->
- <guava.version>33.3.1-jre</guava.version>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <scope>${provided.scope}</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-hive</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <configuration>
- <createDependencyReducedPom>true</createDependencyReducedPom>
- <filters>
- <filter>
- <artifact>*:*</artifact>
- <excludes>
- <exclude>META-INF/*.SF</exclude>
- <exclude>META-INF/*.sf</exclude>
- <exclude>META-INF/*.DSA</exclude>
- <exclude>META-INF/*.dsa</exclude>
- <exclude>META-INF/*.RSA</exclude>
- <exclude>META-INF/*.rsa</exclude>
- <exclude>META-INF/*.EC</exclude>
- <exclude>META-INF/*.ec</exclude>
- <exclude>META-INF/MSFTSIG.SF</exclude>
- <exclude>META-INF/MSFTSIG.RSA</exclude>
- </excludes>
- </filter>
- </filters>
- </configuration>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
- <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- </transformer>
- </transformers>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- <!--Note - the version would be inherited-->
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-pmd-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/BucketTestHiveTopology.java b/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/BucketTestHiveTopology.java
deleted file mode 100644
index 8432c532f..000000000
--- a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/BucketTestHiveTopology.java
+++ /dev/null
@@ -1,182 +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.storm.hive.bolt;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-
-
-public class BucketTestHiveTopology {
- static final String USER_SPOUT_ID = "user-spout";
- static final String BOLT_ID = "my-hive-bolt";
- static final String TOPOLOGY_NAME = "hive-test-topology1";
-
- public static void main(String[] args) throws Exception {
- if ((args == null) || (args.length < 7)) {
- System.out.println("Usage: BucketTestHiveTopology metastoreURI "
- + "dbName tableName dataFileLocation hiveBatchSize "
- + "hiveTickTupl]eIntervalSecs workers [topologyNamey] [keytab file]"
- + " [principal name] ");
- System.exit(1);
- }
- String metaStoreUri = args[0];
- String dbName = args[1];
- String tblName = args[2];
- Integer hiveBatchSize = Integer.parseInt(args[4]);
- Integer hiveTickTupleIntervalSecs = Integer.parseInt(args[5]);
- Integer workers = Integer.parseInt(args[6]);
- String[] colNames = {
- "ss_sold_date_sk", "ss_sold_time_sk", "ss_item_sk",
- "ss_customer_sk", "ss_cdemo_sk", "ss_hdemo_sk", "ss_addr_sk",
- "ss_store_sk", "ss_promo_sk", "ss_ticket_number", "ss_quantity",
- "ss_wholesale_cost", "ss_list_price", "ss_sales_price",
- "ss_ext_discount_amt", "ss_ext_sales_price",
- "ss_ext_wholesale_cost", "ss_ext_list_price", "ss_ext_tax",
- "ss_coupon_amt", "ss_net_paid", "ss_net_paid_inc_tax",
- "ss_net_profit"
- };
- Config config = new Config();
- config.setNumWorkers(workers);
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames)).withTimeAsPartitionField("yyyy/MM/dd");
- HiveOptions hiveOptions;
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(hiveBatchSize);
- // doing below because its affecting storm metrics most likely
- // had to make tick tuple a mandatory argument since its positional
- if (hiveTickTupleIntervalSecs > 0) {
- hiveOptions.withTickTupleInterval(hiveTickTupleIntervalSecs);
- }
- if (args.length == 10) {
- hiveOptions.withKerberosKeytab(args[8]).withKerberosPrincipal(args[9]);
- }
- HiveBolt hiveBolt = new HiveBolt(hiveOptions);
- TopologyBuilder builder = new TopologyBuilder();
- String sourceFileLocation = args[3];
- UserDataSpout spout = new UserDataSpout().withDataFile(sourceFileLocation);
- builder.setSpout(USER_SPOUT_ID, spout, 1);
- // SentenceSpout --> MyBolt
- builder.setBolt(BOLT_ID, hiveBolt, 14)
- .shuffleGrouping(USER_SPOUT_ID);
- String topoName = TOPOLOGY_NAME;
- if (args.length > 6) {
- topoName = args[7];
- }
- StormSubmitter.submitTopology(args[7], config, builder.createTopology());
- }
-
- public static class UserDataSpout extends BaseRichSpout {
- private ConcurrentHashMap<UUID, Values> pending;
- private SpoutOutputCollector collector;
- private String filePath;
- private BufferedReader br;
- private int count = 0;
- private long total = 0L;
- private String[] outputFields = {
- "ss_sold_date_sk", "ss_sold_time_sk",
- "ss_item_sk", "ss_customer_sk", "ss_cdemo_sk", "ss_hdemo_sk",
- "ss_addr_sk", "ss_store_sk", "ss_promo_sk", "ss_ticket_number",
- "ss_quantity", "ss_wholesale_cost", "ss_list_price",
- "ss_sales_price", "ss_ext_discount_amt", "ss_ext_sales_price",
- "ss_ext_wholesale_cost", "ss_ext_list_price", "ss_ext_tax",
- "ss_coupon_amt", "ss_net_paid", "ss_net_paid_inc_tax",
- "ss_net_profit"
- };
-
- public UserDataSpout withDataFile(String filePath) {
- this.filePath = filePath;
- return this;
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields(this.outputFields));
- }
-
- @Override
- public void open(Map<String, Object> config, TopologyContext context,
- SpoutOutputCollector collector) {
- this.collector = collector;
- this.pending = new ConcurrentHashMap<UUID, Values>();
- try {
- this.br = new BufferedReader(new FileReader(new File(this
- .filePath)));
- } catch (Exception ex) {
- ex.printStackTrace();
- }
- }
-
- @Override
- public void nextTuple() {
- String line;
- try {
- if ((line = br.readLine()) != null) {
- System.out.println("*********" + line);
- String[] values = line.split("\\|", -1);
- // above gives an extra empty string at the end. below
- // removes that
- values = Arrays.copyOfRange(values, 0,
- this.outputFields.length);
- Values tupleValues = new Values(values);
- UUID msgId = UUID.randomUUID();
- this.pending.put(msgId, tupleValues);
- this.collector.emit(tupleValues, msgId);
- count++;
- total++;
- if (count > 1000) {
- count = 0;
- System.out.println("Pending count: " + this.pending.size() + ", total: " + this.total);
- }
- }
- } catch (IOException ex) {
- ex.printStackTrace();
- }
- }
-
- @Override
- public void ack(Object msgId) {
- this.pending.remove(msgId);
- }
-
- @Override
- public void fail(Object msgId) {
- System.out.println("**** RESENDING FAILED TUPLE");
- this.collector.emit(this.pending.get(msgId), msgId);
- }
- }
-}
diff --git a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopology.java b/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopology.java
deleted file mode 100644
index 4cfd3e49f..000000000
--- a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopology.java
+++ /dev/null
@@ -1,147 +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.storm.hive.bolt;
-
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-
-
-public class HiveTopology {
- static final String USER_SPOUT_ID = "user-spout";
- static final String BOLT_ID = "my-hive-bolt";
- static final String TOPOLOGY_NAME = "hive-test-topology1";
-
- public static void main(String[] args) throws Exception {
- String metaStoreUri = args[0];
- String dbName = args[1];
- String tblName = args[2];
- String[] colNames = {"id", "name", "phone", "street", "city", "state"};
- Config config = new Config();
- config.setNumWorkers(1);
- UserDataSpout spout = new UserDataSpout();
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames));
- HiveOptions hiveOptions;
- if (args.length == 6) {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(100)
- .withIdleTimeout(10)
- .withKerberosKeytab(args[4])
- .withKerberosPrincipal(args[5]);
- } else {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(100)
- .withIdleTimeout(10)
- .withMaxOpenConnections(1);
- }
-
- HiveBolt hiveBolt = new HiveBolt(hiveOptions);
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout(USER_SPOUT_ID, spout, 1);
- // SentenceSpout --> MyBolt
- builder.setBolt(BOLT_ID, hiveBolt, 1)
- .shuffleGrouping(USER_SPOUT_ID);
-
- String topoName = TOPOLOGY_NAME;
- if (args.length >= 4) {
- topoName = args[3];
- }
- StormSubmitter.submitTopology(topoName, config, builder.createTopology());
- }
-
- public static void waitForSeconds(int seconds) {
- try {
- Thread.sleep(seconds * 1000);
- } catch (InterruptedException e) {
- //ignore
- }
- }
-
- public static class UserDataSpout extends BaseRichSpout {
- private ConcurrentHashMap<UUID, Values> pending;
- private SpoutOutputCollector collector;
- private String[] sentences = {
- "1,user1,123456,street1,sunnyvale,ca",
- "2,user2,123456,street2,sunnyvale,ca",
- "3,user3,123456,street3,san jose,ca",
- "4,user4,123456,street4,san jose,ca",
- };
- private int index = 0;
- private int count = 0;
- private long total = 0L;
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("id", "name", "phone", "street", "city", "state"));
- }
-
- @Override
- public void open(Map<String, Object> config, TopologyContext context,
- SpoutOutputCollector collector) {
- this.collector = collector;
- this.pending = new ConcurrentHashMap<UUID, Values>();
- }
-
- @Override
- public void nextTuple() {
- String[] user = sentences[index].split(",");
- Values values = new Values(Integer.parseInt(user[0]), user[1], user[2], user[3], user[4], user[5]);
- UUID msgId = UUID.randomUUID();
- this.pending.put(msgId, values);
- this.collector.emit(values, msgId);
- index++;
- if (index >= sentences.length) {
- index = 0;
- }
- count++;
- total++;
- if (count > 1000) {
- count = 0;
- System.out.println("Pending count: " + this.pending.size() + ", total: " + this.total);
- }
- Thread.yield();
- }
-
- @Override
- public void ack(Object msgId) {
- this.pending.remove(msgId);
- }
-
- @Override
- public void fail(Object msgId) {
- System.out.println("**** RESENDING FAILED TUPLE");
- this.collector.emit(this.pending.get(msgId), msgId);
- }
- }
-}
diff --git a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopologyPartitioned.java b/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopologyPartitioned.java
deleted file mode 100644
index ca9fd5f7f..000000000
--- a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/bolt/HiveTopologyPartitioned.java
+++ /dev/null
@@ -1,148 +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.storm.hive.bolt;
-
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.Utils;
-
-
-public class HiveTopologyPartitioned {
- static final String USER_SPOUT_ID = "hive-user-spout-partitioned";
- static final String BOLT_ID = "my-hive-bolt-partitioned";
- static final String TOPOLOGY_NAME = "hive-test-topology-partitioned";
-
- public static void main(String[] args) throws Exception {
- String metaStoreUri = args[0];
- String dbName = args[1];
- String tblName = args[2];
- String[] partNames = {"city", "state"};
- String[] colNames = {"id", "name", "phone", "street"};
- Config config = new Config();
- config.setNumWorkers(1);
- UserDataSpout spout = new UserDataSpout();
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions;
- if (args.length == 6) {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(1000)
- .withIdleTimeout(10)
- .withKerberosKeytab(args[4])
- .withKerberosPrincipal(args[5]);
- } else {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(1000)
- .withIdleTimeout(10);
- }
-
- HiveBolt hiveBolt = new HiveBolt(hiveOptions);
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout(USER_SPOUT_ID, spout, 1);
- // SentenceSpout --> MyBolt
- builder.setBolt(BOLT_ID, hiveBolt, 1)
- .shuffleGrouping(USER_SPOUT_ID);
- String topoName = TOPOLOGY_NAME;
- if (args.length > 3) {
- topoName = args[3];
- }
- StormSubmitter.submitTopology(topoName, config, builder.createTopology());
- }
-
- public static void waitForSeconds(int seconds) {
- try {
- Thread.sleep(seconds * 1000);
- } catch (InterruptedException e) {
- //ignore
- }
- }
-
- public static class UserDataSpout extends BaseRichSpout {
- private ConcurrentHashMap<UUID, Values> pending;
- private SpoutOutputCollector collector;
- private String[] sentences = {
- "1,user1,123456,street1,sunnyvale,ca",
- "2,user2,123456,street2,sunnyvale,ca",
- "3,user3,123456,street3,san jose,ca",
- "4,user4,123456,street4,san jose,ca",
- };
- private int index = 0;
- private int count = 0;
- private long total = 0L;
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("id", "name", "phone", "street", "city", "state"));
- }
-
- @Override
- public void open(Map<String, Object> config, TopologyContext context,
- SpoutOutputCollector collector) {
- this.collector = collector;
- this.pending = new ConcurrentHashMap<UUID, Values>();
- }
-
- @Override
- public void nextTuple() {
- String[] user = sentences[index].split(",");
- Values values = new Values(Integer.parseInt(user[0]), user[1], user[2], user[3], user[4], user[5]);
- UUID msgId = UUID.randomUUID();
- this.pending.put(msgId, values);
- this.collector.emit(values, msgId);
- index++;
- if (index >= sentences.length) {
- index = 0;
- }
- count++;
- total++;
- if (count > 1000) {
- Utils.sleep(1000);
- count = 0;
- System.out.println("Pending count: " + this.pending.size() + ", total: " + this.total);
- }
- }
-
- @Override
- public void ack(Object msgId) {
- this.pending.remove(msgId);
- }
-
- @Override
- public void fail(Object msgId) {
- System.out.println("**** RESENDING FAILED TUPLE");
- this.collector.emit(this.pending.get(msgId), msgId);
- }
- }
-}
diff --git a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/trident/TridentHiveTopology.java b/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/trident/TridentHiveTopology.java
deleted file mode 100644
index 1ba734ddf..000000000
--- a/examples/storm-hive-examples/src/main/java/org/apache/storm/hive/trident/TridentHiveTopology.java
+++ /dev/null
@@ -1,187 +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.storm.hive.trident;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.hooks.SubmitterHookException;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.trident.Stream;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.spout.IBatchSpout;
-import org.apache.storm.trident.state.StateFactory;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TridentHiveTopology {
- private static final Logger LOG = LoggerFactory.getLogger(TridentHiveTopology.class);
-
- public static StormTopology buildTopology(String metaStoreUri, String dbName, String tblName, Object keytab, Object principal) {
- int batchSize = 100;
- FixedBatchSpout spout = new FixedBatchSpout(batchSize);
- spout.setCycle(true);
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("hiveTridentspout1", spout);
- String[] partNames = {"city", "state"};
- String[] colNames = {"id", "name", "phone", "street"};
- Fields hiveFields = new Fields("id", "name", "phone", "street", "city", "state");
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions;
- if (keytab != null && principal != null) {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(batchSize)
- .withIdleTimeout(10)
- .withCallTimeout(30000)
- .withKerberosKeytab((String) keytab)
- .withKerberosPrincipal((String) principal);
- } else {
- hiveOptions = new HiveOptions(metaStoreUri, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(batchSize)
- .withCallTimeout(30000)
- .withIdleTimeout(10);
- }
- StateFactory factory = new HiveStateFactory().withOptions(hiveOptions);
- TridentState state = stream.partitionPersist(factory, hiveFields, new HiveUpdater(), new Fields());
- return topology.build();
- }
-
- public static void waitForSeconds(int seconds) {
- try {
- Thread.sleep(seconds * 1000);
- } catch (InterruptedException e) {
- //ignore
- }
- }
-
- public static void main(String[] args) throws Exception {
- String metaStoreUri = args[0];
- String dbName = args[1];
- String tblName = args[2];
- Config conf = new Config();
- conf.setMaxSpoutPending(5);
- String topoName = "tridentHiveTopology";
- String keytab = null;
- String principal = null;
-
- if (args.length > 3) {
- topoName = args[3];
- }
- if (args.length == 6) {
- keytab = args[4];
- principal = args[5];
- } else if (args.length != 3 && args.length != 4) {
- LOG.info("Usage: TridentHiveTopology metastoreURI dbName tableName [topologyName] [keytab principal]");
- return;
- }
-
- try {
- StormSubmitter.submitTopology(args[3], conf, buildTopology(metaStoreUri, dbName, tblName, null, null));
- } catch (SubmitterHookException e) {
- LOG.warn("Topology is submitted but invoking ISubmitterHook failed", e);
- } catch (Exception e) {
- LOG.warn("Failed to submit topology ", e);
- }
- }
-
- public static class FixedBatchSpout implements IBatchSpout {
- int maxBatchSize;
- HashMap<Long, List<List<Object>>> batches = new HashMap<Long, List<List<Object>>>();
- private Values[] outputs = {
- new Values("1", "user1", "123456", "street1", "sunnyvale", "ca"),
- new Values("2", "user2", "123456", "street2", "sunnyvale", "ca"),
- new Values("3", "user3", "123456", "street3", "san jose", "ca"),
- new Values("4", "user4", "123456", "street4", "san jose", "ca"),
- };
- private int index = 0;
- boolean cycle = false;
-
- public FixedBatchSpout(int maxBatchSize) {
- this.maxBatchSize = maxBatchSize;
- }
-
- public void setCycle(boolean cycle) {
- this.cycle = cycle;
- }
-
- @Override
- public Fields getOutputFields() {
- return new Fields("id", "name", "phone", "street", "city", "state");
- }
-
- @Override
- public void open(Map<String, Object> conf, TopologyContext context) {
- index = 0;
- }
-
- @Override
- public void emitBatch(long batchId, TridentCollector collector) {
- List<List<Object>> batch = this.batches.get(batchId);
- if (batch == null) {
- batch = new ArrayList<List<Object>>();
- if (index >= outputs.length && cycle) {
- index = 0;
- }
- for (int i = 0; i < maxBatchSize; index++, i++) {
- if (index == outputs.length) {
- index = 0;
- }
- batch.add(outputs[index]);
- }
- this.batches.put(batchId, batch);
- }
- for (List<Object> list : batch) {
- collector.emit(list);
- }
- }
-
- @Override
- public void ack(long batchId) {
- this.batches.remove(batchId);
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Config conf = new Config();
- conf.setMaxTaskParallelism(1);
- return conf;
- }
-
- }
-
-}
diff --git a/external/storm-autocreds/pom.xml b/external/storm-autocreds/pom.xml
index 53077dcdf..0787a3cb9 100644
--- a/external/storm-autocreds/pom.xml
+++ b/external/storm-autocreds/pom.xml
@@ -103,54 +103,6 @@
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.apache.hive.hcatalog</groupId>
- <artifactId>hive-webhcat-java-client</artifactId>
- <version>${hive.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.pentaho</groupId>
- <artifactId>pentaho-aggdesigner-algorithm</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </exclusion>
- <exclusion>
- <!-- Not needed for runtime ops -->
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-cli</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.bouncycastle</groupId>
- <artifactId>bcpkix-jdk15on</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.bouncycastle</groupId>
- <artifactId>bcprov-jdk15on</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHive.java b/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHive.java
deleted file mode 100644
index daf3c6a2e..000000000
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHive.java
+++ /dev/null
@@ -1,49 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.storm.hive.security;
-
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_CREDENTIALS;
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_CREDENTIALS_CONFIG_KEYS;
-
-import java.util.Map;
-
-import org.apache.storm.common.AbstractHadoopAutoCreds;
-
-/**
- * Auto credentials plugin for Hive implementation. This class provides a way to automatically
- * push credentials to a topology and to retrieve them in the worker.
- */
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class AutoHive extends AbstractHadoopAutoCreds {
- @Override
- public void doPrepare(Map<String, Object> conf) {
- }
-
- @Override
- protected String getConfigKeyString() {
- return HIVE_CREDENTIALS_CONFIG_KEYS;
- }
-
- @Override
- public String getCredentialKey(String configKey) {
- return HIVE_CREDENTIALS + configKey;
- }
-
-}
-
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveCommand.java b/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveCommand.java
deleted file mode 100644
index 662053f7d..000000000
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveCommand.java
+++ /dev/null
@@ -1,68 +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.storm.hive.security;
-
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_KEYTAB_FILE_KEY;
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_PRINCIPAL_KEY;
-
-import java.util.HashMap;
-import java.util.Map;
-import javax.security.auth.Subject;
-
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.storm.Config;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Command tool of Hive credential renewer.
- */
-@Deprecated(since = "2.7.1", forRemoval = true)
-public final class AutoHiveCommand {
- private static final Logger LOG = LoggerFactory.getLogger(AutoHiveCommand.class);
-
- private AutoHiveCommand() {
- }
-
- @SuppressWarnings("unchecked")
- public static void main(String[] args) throws Exception {
- Map<String, Object> conf = new HashMap<>();
- conf.put(HIVE_PRINCIPAL_KEY, args[1]); // hive principal storm-hive@WITZEN.COM
- conf.put(HIVE_KEYTAB_FILE_KEY, args[2]); // storm hive keytab /etc/security/keytabs/storm-hive.keytab
- // hive.metastore.uris : "thrift://pm-eng1-cluster1.field.hortonworks.com:9083"
- conf.put(HiveConf.ConfVars.METASTOREURIS.varname, args[3]);
-
- AutoHive autoHive = new AutoHive();
- autoHive.prepare(conf);
- AutoHiveNimbus autoHiveNimbus = new AutoHiveNimbus();
- autoHiveNimbus.prepare(conf);
-
- Map<String, String> creds = new HashMap<>();
- autoHiveNimbus.populateCredentials(creds, conf, args[0]);
- LOG.info("Got Hive credentials" + autoHive.getCredentials(creds));
-
- Subject subject = new Subject();
- autoHive.populateSubject(subject, creds);
- LOG.info("Got a Subject " + subject);
-
- autoHiveNimbus.renew(creds, conf, args[0]);
- LOG.info("Renewed credentials" + autoHive.getCredentials(creds));
- }
-
-}
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveNimbus.java b/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveNimbus.java
deleted file mode 100644
index 947034edf..000000000
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/AutoHiveNimbus.java
+++ /dev/null
@@ -1,260 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.storm.hive.security;
-
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_CREDENTIALS;
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_CREDENTIALS_CONFIG_KEYS;
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_KEYTAB_FILE_KEY;
-import static org.apache.storm.hive.security.HiveSecurityUtil.HIVE_PRINCIPAL_KEY;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.io.ObjectOutputStream;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.math3.util.Pair;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
-import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hive.hcatalog.api.HCatClient;
-import org.apache.hive.hcatalog.common.HCatException;
-import org.apache.storm.common.AbstractHadoopNimbusPluginAutoCreds;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Auto credentials nimbus plugin for Hive implementation. This class automatically
- * gets Hive delegation tokens and push it to user's topology.
- */
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class AutoHiveNimbus extends AbstractHadoopNimbusPluginAutoCreds {
- private static final Logger LOG = LoggerFactory.getLogger(AutoHiveNimbus.class);
-
- public String hiveKeytab;
- public String hivePrincipal;
- @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
- public String metaStoreURI;
-
- @Override
- public void doPrepare(Map<String, Object> conf) {
- if (conf.containsKey(HIVE_KEYTAB_FILE_KEY) && conf.containsKey(HIVE_PRINCIPAL_KEY)) {
- hiveKeytab = (String) conf.get(HIVE_KEYTAB_FILE_KEY);
- hivePrincipal = (String) conf.get(HIVE_PRINCIPAL_KEY);
- metaStoreURI = (String) conf.get(HiveConf.ConfVars.METASTOREURIS.varname);
- }
- }
-
- @Override
- protected String getConfigKeyString() {
- return HIVE_CREDENTIALS_CONFIG_KEYS;
- }
-
- @Override
- public void shutdown() {
- //no op.
- }
-
- @Override
- protected byte[] getHadoopCredentials(Map<String, Object> conf, String configKey, final String topologyOwnerPrincipal) {
- Configuration configuration = getHadoopConfiguration(conf, configKey);
- return getHadoopCredentials(conf, configuration, topologyOwnerPrincipal);
- }
-
- @Override
- protected byte[] getHadoopCredentials(Map<String, Object> conf, final String topologyOwnerPrincipal) {
- Configuration configuration = new Configuration();
- return getHadoopCredentials(conf, configuration, topologyOwnerPrincipal);
- }
-
- @SuppressWarnings("unchecked")
- protected byte[] getHadoopCredentials(Map<String, Object> conf, final Configuration configuration, final String topologySubmitterUser) {
- try {
- if (UserGroupInformation.isSecurityEnabled()) {
- String hiveMetaStoreUri = getMetaStoreUri(configuration);
- String hiveMetaStorePrincipal = getMetaStorePrincipal(configuration);
- HiveConf hcatConf = createHiveConf(hiveMetaStoreUri, hiveMetaStorePrincipal);
- login(configuration);
-
- UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
- UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(topologySubmitterUser, currentUser);
- try {
- Token<DelegationTokenIdentifier> delegationTokenId =
- getDelegationToken(hcatConf, hiveMetaStorePrincipal, topologySubmitterUser);
- proxyUser.addToken(delegationTokenId);
- LOG.info("Obtained Hive tokens, adding to user credentials.");
-
- Credentials credential = proxyUser.getCredentials();
- ByteArrayOutputStream bao = new ByteArrayOutputStream();
- ObjectOutputStream out = new ObjectOutputStream(bao);
- credential.write(out);
- out.flush();
- out.close();
- return bao.toByteArray();
- } catch (Exception ex) {
- LOG.debug(" Exception" + ex.getMessage());
- throw ex;
- }
- } else {
- throw new RuntimeException("Security is not enabled for Hadoop");
- }
- } catch (Exception ex) {
- throw new RuntimeException("Failed to get delegation tokens.", ex);
- }
- }
-
- private Configuration getHadoopConfiguration(Map<String, Object> topoConf, String configKey) {
- Configuration configuration = new Configuration();
- fillHadoopConfiguration(topoConf, configKey, configuration);
- return configuration;
- }
-
- public HiveConf createHiveConf(String metaStoreUri, String hiveMetaStorePrincipal) throws IOException {
- HiveConf hcatConf = new HiveConf();
- hcatConf.setVar(HiveConf.ConfVars.METASTOREURIS, metaStoreUri);
- hcatConf.setIntVar(HiveConf.ConfVars.METASTORETHRIFTCONNECTIONRETRIES, 3);
- hcatConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false);
- hcatConf.setBoolVar(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL, true);
- hcatConf.set(HiveConf.ConfVars.METASTORE_KERBEROS_PRINCIPAL.varname, hiveMetaStorePrincipal);
- return hcatConf;
- }
-
- private Token<DelegationTokenIdentifier> getDelegationToken(HiveConf hcatConf,
- String metaStoreServicePrincipal,
- String topologySubmitterUser) throws IOException {
- LOG.info("Creating delegation tokens for principal={}", metaStoreServicePrincipal);
-
- HCatClient hcatClient = null;
- try {
- hcatClient = HCatClient.create(hcatConf);
- String delegationToken = hcatClient.getDelegationToken(topologySubmitterUser, metaStoreServicePrincipal);
- Token<DelegationTokenIdentifier> delegationTokenId = new Token<DelegationTokenIdentifier>();
- delegationTokenId.decodeFromUrlString(delegationToken);
-
- DelegationTokenIdentifier d = new DelegationTokenIdentifier();
- d.readFields(new DataInputStream(new ByteArrayInputStream(
- delegationTokenId.getIdentifier())));
- LOG.info("Created Delegation Token for : " + d.getUser());
-
- return delegationTokenId;
- } finally {
- if (hcatClient != null) {
- hcatClient.close();
- }
- }
- }
-
- private String getMetaStoreUri(Configuration configuration) {
- if (configuration.get(HiveConf.ConfVars.METASTOREURIS.varname) == null) {
- return metaStoreURI;
- } else {
- return configuration.get(HiveConf.ConfVars.METASTOREURIS.varname);
- }
- }
-
- private String getMetaStorePrincipal(Configuration configuration) {
- if (configuration.get(HIVE_PRINCIPAL_KEY) == null) {
- return hivePrincipal;
- } else {
- return configuration.get(HIVE_PRINCIPAL_KEY);
- }
- }
-
- private void login(Configuration configuration) throws IOException {
- if (configuration.get(HIVE_KEYTAB_FILE_KEY) == null) {
- configuration.set(HIVE_KEYTAB_FILE_KEY, hiveKeytab);
- }
- if (configuration.get(HIVE_PRINCIPAL_KEY) == null) {
- configuration.set(HIVE_PRINCIPAL_KEY, hivePrincipal);
- }
- SecurityUtil.login(configuration, HIVE_KEYTAB_FILE_KEY, HIVE_PRINCIPAL_KEY);
- LOG.info("Logged into hive with principal {}", configuration.get(HIVE_PRINCIPAL_KEY));
- }
-
- @Override
- public void doRenew(Map<String, String> credentials, Map<String, Object> topologyConf, final String topologyOwnerPrincipal) {
- List<String> configKeys = getConfigKeys(topologyConf);
- for (Pair<String, Credentials> cred : getCredentials(credentials, configKeys)) {
- try {
- Configuration configuration = getHadoopConfiguration(topologyConf, cred.getFirst());
- String hiveMetaStoreUri = getMetaStoreUri(configuration);
- String hiveMetaStorePrincipal = getMetaStorePrincipal(configuration);
-
- Collection<Token<? extends TokenIdentifier>> tokens = cred.getSecond().getAllTokens();
- login(configuration);
-
- if (tokens != null && !tokens.isEmpty()) {
- for (Token<? extends TokenIdentifier> token : tokens) {
- long expiration = renewToken(token, hiveMetaStoreUri, hiveMetaStorePrincipal);
- LOG.info("Hive delegation token renewed, new expiration time {}", expiration);
- }
- } else {
- LOG.debug("No tokens found for credentials, skipping renewal.");
- }
- } catch (Exception e) {
- LOG.warn("could not renew the credentials, one of the possible reason is tokens are beyond "
- + "renewal period so attempting to get new tokens.",
- e);
- populateCredentials(credentials, topologyConf);
- }
- }
- }
-
- private long renewToken(Token token, String metaStoreUri, String hiveMetaStorePrincipal) {
- HCatClient hcatClient = null;
- if (UserGroupInformation.isSecurityEnabled()) {
- try {
- String tokenStr = token.encodeToUrlString();
- HiveConf hcatConf = createHiveConf(metaStoreUri, hiveMetaStorePrincipal);
- LOG.debug("renewing delegation tokens for principal={}", hiveMetaStorePrincipal);
- hcatClient = HCatClient.create(hcatConf);
- Long expiryTime = hcatClient.renewDelegationToken(tokenStr);
- LOG.info("Renewed delegation token. new expiryTime={}", expiryTime);
- return expiryTime;
- } catch (Exception ex) {
- throw new RuntimeException("Failed to renew delegation tokens.", ex);
- } finally {
- if (hcatClient != null) {
- try {
- hcatClient.close();
- } catch (HCatException e) {
- LOG.error(" Exception", e);
- }
- }
- }
- } else {
- throw new RuntimeException("Security is not enabled for Hadoop");
- }
- }
-
- @Override
- public String getCredentialKey(String configKey) {
- return HIVE_CREDENTIALS + configKey;
- }
-
-}
-
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/HiveSecurityUtil.java b/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/HiveSecurityUtil.java
deleted file mode 100644
index c3424e106..000000000
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hive/security/HiveSecurityUtil.java
+++ /dev/null
@@ -1,34 +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.storm.hive.security;
-
-/**
- * This class provides util methods for storm-hdfs connector communicating
- * with secured Hive.
- */
-@Deprecated(since = "2.7.1", forRemoval = true)
-public final class HiveSecurityUtil {
- public static final String HIVE_KEYTAB_FILE_KEY = "hive.keytab.file";
- public static final String HIVE_PRINCIPAL_KEY = "hive.kerberos.principal";
- public static final String HIVE_CREDENTIALS_CONFIG_KEYS = "hiveCredentialsConfigKeys";
- public static final String HIVE_CREDENTIALS = "HIVE_CREDENTIALS";
-
- private HiveSecurityUtil() {
- }
-}
diff --git a/external/storm-hive/README.md b/external/storm-hive/README.md
deleted file mode 100644
index 5e5df1ff2..000000000
--- a/external/storm-hive/README.md
+++ /dev/null
@@ -1,189 +0,0 @@
-# Storm Hive Bolt & Trident State
-
- Hive offers streaming API that allows data to be written continuously into Hive. The incoming data
- can be continuously committed in small batches of records into existing Hive partition or table. Once the data
- is committed its immediately visible to all hive queries. More info on Hive Streaming API
- https://cwiki.apache.org/confluence/display/Hive/Streaming+Data+Ingest
-
- With the help of Hive Streaming API, HiveBolt and HiveState allows users to stream data from Storm into Hive directly.
- To use Hive streaming API users need to create a bucketed table with ORC format. Example below
-
- ```code
- create table test_table ( id INT, name STRING, phone STRING, street STRING) partitioned by (city STRING, state STRING) stored as orc tblproperties ("orc.compress"="NONE");
- ```
-
-
-## HiveBolt (org.apache.storm.hive.bolt.HiveBolt)
-
-HiveBolt streams tuples directly into Hive. Tuples are written using Hive Transactions.
-Partitions to which HiveBolt will stream to can either created or pre-created or optionally
-HiveBolt can create them if they are missing. Fields from Tuples are mapped to table columns.
-User should make sure that Tuple field names are matched to the table column names.
-
-```java
-DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames));
-HiveOptions hiveOptions = new HiveOptions(metaStoreURI,dbName,tblName,mapper);
-HiveBolt hiveBolt = new HiveBolt(hiveOptions);
-```
-
-### RecordHiveMapper
- This class maps Tuple field names to Hive table column names.
- There are two implementaitons available
-
-
- + DelimitedRecordHiveMapper (org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper)
- + JsonRecordHiveMapper (org.apache.storm.hive.bolt.mapper.JsonRecordHiveMapper)
-
- ```java
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- or
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withTimeAsPartitionField("YYYY/MM/DD");
- ```
-
-|Arg | Description | Type
-|--- |--- |---
-|withColumnFields| field names in a tuple to be mapped to table column names | Fields (required) |
-|withPartitionFields| field names in a tuple can be mapped to hive table partitions | Fields |
-|withTimeAsPartitionField| users can select system time as partition in hive table| String . Date format|
-
-### HiveOptions (org.apache.storm.hive.common.HiveOptions)
-
-HiveBolt takes in HiveOptions as a constructor arg.
-
- ```java
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI,dbName,tblName,mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(1000)
- .withIdleTimeout(10)
- ```
-
-
-HiveOptions params
-
-|Arg |Description | Type
-|--- |--- |---
-|metaStoreURI | hive meta store URI (can be found in hive-site.xml) | String (required) |
-|dbName | database name | String (required) |
-|tblName | table name | String (required) |
-|mapper| Mapper class to map Tuple field names to Table column names | DelimitedRecordHiveMapper or JsonRecordHiveMapper (required) |
-|withTxnsPerBatch | Hive grants a *batch of transactions* instead of single transactions to streaming clients like HiveBolt.This setting configures the number of desired transactions per Transaction Batch. Data from all transactions in a single batch end up in a single file. HiveBolt will write a maximum of batchSize events in each transaction in the batch. This setting in conjunction with batchSize provides control over the size of each file. Note that eventually Hive will transparently [...]
-|withMaxOpenConnections| Allow only this number of open connections. If this number is exceeded, the least recently used connection is closed.| Integer . default 100|
-|withBatchSize| Max number of events written to Hive in a single Hive transaction| Integer. default 15000|
-|withCallTimeout| (In milliseconds) Timeout for Hive & HDFS I/O operations, such as openTxn, write, commit, abort. | Integer. default 10000|
-|withHeartBeatInterval| (In seconds) Interval between consecutive heartbeats sent to Hive to keep unused transactions from expiring. Set this value to 0 to disable heartbeats.| Integer. default 240 |
-|withAutoCreatePartitions| HiveBolt will automatically create the necessary Hive partitions to stream to. |Boolean. default true |
-|withKerberosPrinicipal| Kerberos user principal for accessing secure Hive | String|
-|withKerberosKeytab| Kerberos keytab for accessing secure Hive | String |
-|withTickTupleInterval| (In seconds) If > 0 then the Hive Bolt will periodically flush transaction batches. Enabling this is recommended to avoid tuple timeouts while waiting for a batch to fill up.| Integer. default 0|
-
-
-
-## HiveState (org.apache.storm.hive.trident.HiveTrident)
-
-Hive Trident state also follows similar pattern to HiveBolt it takes in HiveOptions as an arg.
-
-```code
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withTimeAsPartitionField("YYYY/MM/DD");
-
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI,dbName,tblName,mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(1000)
- .withIdleTimeout(10)
-
- StateFactory factory = new HiveStateFactory().withOptions(hiveOptions);
- TridentState state = stream.partitionPersist(factory, hiveFields, new HiveUpdater(), new Fields());
-```
-
-
-##Working with Secure Hive
-If your topology is going to interact with secure Hive, your bolts/states needs to be authenticated by Hive Server. We
-currently have 2 options to support this:
-
-### Using keytabs on all worker hosts
-If you have distributed the keytab files for hive user on all potential worker hosts then you can use this method. You should specify
-hive configs using the methods HiveOptions.withKerberosKeytab(), HiveOptions.withKerberosPrincipal() methods.
-
-On worker hosts the bolt/trident-state code will use the keytab file with principal provided in the config to authenticate with
-Hive. This method is little dangerous as you need to ensure all workers have the keytab file at the same location and you need
-to remember this as you bring up new hosts in the cluster.
-
-
-### Using Hive MetaStore delegation tokens
-Your administrator can configure nimbus to automatically get delegation tokens on behalf of the topology submitter user.
-Since Hive depends on HDFS, we should also configure HDFS delegation tokens.
-
-More details about Hadoop Tokens here: https://github.com/apache/storm/blob/master/docs/storm-hdfs.md
-
-The nimbus should be started with following configurations:
-
-```
-nimbus.autocredential.plugins.classes : ["org.apache.storm.hive.security.AutoHive", "org.apache.storm.hdfs.security.AutoHDFS"]
-nimbus.credential.renewers.classes : ["org.apache.storm.hive.security.AutoHive", "org.apache.storm.hdfs.security.AutoHDFS"]
-nimbus.credential.renewers.freq.secs : 82800 (23 hours)
-
-hive.keytab.file: "/path/to/keytab/on/nimbus" (Keytab of The Hive metastore thrift server service principal. This is used to impersonate other users.)
-hive.kerberos.principal: "hive-metastore/_HOST@EXAMPLE.com" (The service principal for the metastore thrift server.)
-hive.metastore.uris: "thrift://server:9083"
-
-//hdfs configs
-hdfs.keytab.file: "/path/to/keytab/on/nimbus" (This is the keytab of hdfs super user that can impersonate other users.)
-hdfs.kerberos.principal: "superuser@EXAMPLE.com"
-```
-
-
-Your topology configuration should have:
-
-```
-topology.auto-credentials :["org.apache.storm.hive.security.AutoHive", "org.apache.storm.hdfs.security.AutoHDFS"]
-```
-
-If nimbus did not have the above configuration you need to add and then restart it. Ensure the hadoop configuration
-files (core-site.xml, hdfs-site.xml and hive-site.xml) and the storm-hive connector jar with all the dependencies is present in nimbus's classpath.
-
-As an alternative to adding the configuration files (core-site.xml, hdfs-site.xml and hive-site.xml) to the classpath, you could specify the configurations
-as a part of the topology configuration. E.g. in you custom storm.yaml (or -c option while submitting the topology),
-
-
-```
-hiveCredentialsConfigKeys : ["hivecluster1", "hivecluster2"] (the hive clusters you want to fetch the tokens from)
-"hivecluster1": {"config1": "value1", "config2": "value2", ... } (A map of config key-values specific to cluster1)
-"hivecluster2": {"config1": "value1", "hive.keytab.file": "/path/to/keytab/for/cluster2/on/nimubs", "hive.kerberos.principal": "cluster2user@EXAMPLE.com", "hive.metastore.uris": "thrift://server:9083"} (here along with other configs, we have custom keytab and principal for "cluster2" which will override the keytab/principal specified at topology level)
-
-hdfsCredentialsConfigKeys : ["hdfscluster1", "hdfscluster2"] (the hdfs clusters you want to fetch the tokens from)
-"hdfscluster1": {"config1": "value1", "config2": "value2", ... } (A map of config key-values specific to cluster1)
-"hdfscluster2": {"config1": "value1", "hdfs.keytab.file": "/path/to/keytab/for/cluster2/on/nimubs", "hdfs.kerberos.principal": "cluster2user@EXAMPLE.com"} (here along with other configs, we have custom keytab and principal for "cluster2" which will override the keytab/principal specified at topology level)
-```
-
-Instead of specifying key values you may also directly specify the resource files for e.g.,
-
-```
-"cluster1": {"resources": ["/path/to/core-site1.xml", "/path/to/hdfs-site1.xml", "/path/to/hive-site1.xml"]}
-"cluster2": {"resources": ["/path/to/core-site2.xml", "/path/to/hdfs-site2.xml", "/path/to/hive-site2.xml"]}
-```
-
-Storm will download the tokens separately for each of the clusters and populate it into the subject and also renew the tokens periodically. This way it would be possible to run multiple bolts connecting to separate Hive cluster within the same topology.
-
-Nimbus will use the keytab and principal specified in the config to authenticate with Hive metastore. From then on for every
-topology submission, nimbus will impersonate the topology submitter user and acquire delegation tokens on behalf of the
-topology submitter user. If topology was started with topology.auto-credentials set to AutoHive, nimbus will push the
-delegation tokens to all the workers for your topology and the hive bolt/state will authenticate with Hive Server using
-these tokens.
-
-As nimbus is impersonating topology submitter user, you need to ensure the user specified in hive.kerberos.principal
-has permissions to acquire tokens on behalf of other users.
-
-## Committer Sponsors
- * Sriharha Chintalapani ([sriharsha@apache.org](mailto:sriharsha@apache.org))
- * P. Taylor Goetz ([ptgoetz@apache.org](mailto:ptgoetz@apache.org))
- * Bobby Evans ([bobby@apache.org](mailto:bobby@apache.org))
-
-
-
-
diff --git a/external/storm-hive/pom.xml b/external/storm-hive/pom.xml
deleted file mode 100644
index 8b6696322..000000000
--- a/external/storm-hive/pom.xml
+++ /dev/null
@@ -1,200 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <artifactId>storm</artifactId>
- <groupId>org.apache.storm</groupId>
- <version>2.8.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <packaging>jar</packaging>
- <artifactId>storm-hive</artifactId>
- <name>storm-hive</name>
- <developers>
- <developer>
- <id>harshach</id>
- <name>Sriharsha Chintalapani</name>
- <email>mail@harsha.io</email>
- </developer>
- </developers>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <scope>${provided.scope}</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hive.hcatalog</groupId>
- <artifactId>hive-hcatalog-streaming</artifactId>
- <version>${hive.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.pentaho</groupId>
- <artifactId>pentaho-aggdesigner-algorithm</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.calcite</groupId>
- <artifactId>calcite-avatica</artifactId>
- </exclusion>
- <exclusion>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.springframework</groupId>
- <artifactId>spring-test</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.springframework</groupId>
- <artifactId>spring-core</artifactId>
- </exclusion>
- <exclusion>
- <!-- Not needed for runtime ops -->
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-cli</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.curator</groupId>
- <artifactId>apache-curator</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-framework</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.bouncycastle</groupId>
- <artifactId>bcpkix-jdk15on</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.bouncycastle</groupId>
- <artifactId>bcprov-jdk15on</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>net.minidev</groupId>
- <artifactId>json-smart</artifactId>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.hamcrest</groupId>
- <artifactId>hamcrest</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-autocreds</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <reuseForks>true</reuseForks>
- <forkCount>1</forkCount>
- </configuration>
- </plugin>
- <plugin>
- <artifactId>maven-clean-plugin</artifactId>
- <executions>
- <execution>
- <id>cleanup</id>
- <phase>test-compile</phase>
- <goals>
- <goal>clean</goal>
- </goals>
- <configuration>
- <excludeDefaultDirectories>true</excludeDefaultDirectories>
- <filesets>
- <fileset>
- <directory>./metastore_db/</directory>
- </fileset>
- </filesets>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- <!--Note - the version would be inherited-->
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-pmd-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
deleted file mode 100644
index 14c6195cb..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
+++ /dev/null
@@ -1,334 +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.storm.hive.bolt;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.SerializationError;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.storm.Config;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.hive.common.HiveUtils;
-import org.apache.storm.hive.common.HiveWriter;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseRichBolt;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.utils.BatchHelper;
-import org.apache.storm.utils.TupleUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveBolt extends BaseRichBolt {
- private static final Logger LOG = LoggerFactory.getLogger(HiveBolt.class);
- @VisibleForTesting
- Map<HiveEndPoint, HiveWriter> allWriters;
- private OutputCollector collector;
- private HiveOptions options;
- private ExecutorService callTimeoutPool;
- private transient Timer heartBeatTimer;
- private AtomicBoolean sendHeartBeat = new AtomicBoolean(false);
- private UserGroupInformation ugi = null;
- private BatchHelper batchHelper;
- private boolean tokenAuthEnabled;
-
- public HiveBolt(HiveOptions options) {
- this.options = options;
- }
-
- @Override
- public void prepare(Map<String, Object> conf, TopologyContext topologyContext, OutputCollector collector) {
- try {
- tokenAuthEnabled = HiveUtils.isTokenAuthEnabled(conf);
- try {
- ugi = HiveUtils.authenticate(tokenAuthEnabled, options.getKerberosKeytab(), options.getKerberosPrincipal());
- } catch (HiveUtils.AuthenticationFailed ex) {
- LOG.error("Hive kerberos authentication failed " + ex.getMessage(), ex);
- throw new IllegalArgumentException(ex);
- }
-
- this.collector = collector;
- this.batchHelper = new BatchHelper(options.getBatchSize(), collector);
- allWriters = new ConcurrentHashMap<HiveEndPoint, HiveWriter>();
- String timeoutName = "hive-bolt-%d";
- this.callTimeoutPool = Executors.newFixedThreadPool(1,
- new ThreadFactoryBuilder().setNameFormat(timeoutName).build());
-
- sendHeartBeat.set(true);
- heartBeatTimer = new Timer(topologyContext.getThisTaskId() + "-hb-timer", true);
- setupHeartBeatTimer();
-
- } catch (Exception e) {
- LOG.warn("unable to make connection to hive ", e);
- }
- }
-
- @Override
- public void execute(Tuple tuple) {
- try {
- if (batchHelper.shouldHandle(tuple)) {
- List<String> partitionVals = options.getMapper().mapPartitions(tuple);
- HiveEndPoint endPoint = HiveUtils.makeEndPoint(partitionVals, options);
- HiveWriter writer = getOrCreateWriter(endPoint);
- writer.write(options.getMapper().mapRecord(tuple));
- batchHelper.addBatch(tuple);
- }
-
- if (batchHelper.shouldFlush()) {
- flushAllWriters(true);
- LOG.info("acknowledging tuples after writers flushed ");
- batchHelper.ack();
- }
- if (TupleUtils.isTick(tuple)) {
- retireIdleWriters();
- }
- } catch (SerializationError se) {
- LOG.info("Serialization exception occurred, tuple is acknowledged but not written to Hive.", tuple);
- this.collector.reportError(se);
- collector.ack(tuple);
- } catch (Exception e) {
- batchHelper.fail(e);
- abortAndCloseWriters();
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
-
- }
-
- @Override
- public void cleanup() {
- sendHeartBeat.set(false);
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- try {
- HiveWriter w = entry.getValue();
- w.flushAndClose();
- } catch (Exception ex) {
- LOG.warn("Error while closing writer to " + entry.getKey() + ". Exception follows.",
- ex);
- if (ex instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- }
- }
-
- ExecutorService[] toShutdown = { callTimeoutPool };
- for (ExecutorService execService : toShutdown) {
- execService.shutdown();
- try {
- while (!execService.isTerminated()) {
- execService.awaitTermination(
- options.getCallTimeOut(), TimeUnit.MILLISECONDS);
- }
- } catch (InterruptedException ex) {
- LOG.warn("shutdown interrupted on " + execService, ex);
- }
- }
-
- callTimeoutPool = null;
- if (heartBeatTimer != null) {
- heartBeatTimer.cancel();
- }
- super.cleanup();
- LOG.info("Hive Bolt stopped");
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Map<String, Object> conf = super.getComponentConfiguration();
- if (conf == null) {
- conf = new Config();
- }
-
- if (options.getTickTupleInterval() > 0) {
- conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, options.getTickTupleInterval());
- }
-
- return conf;
- }
-
- private void setupHeartBeatTimer() {
- if (options.getHeartBeatInterval() > 0) {
- heartBeatTimer.schedule(new TimerTask() {
- @Override
- public void run() {
- try {
- if (sendHeartBeat.get()) {
- LOG.debug("Start sending heartbeat on all writers");
- sendHeartBeatOnAllWriters();
- setupHeartBeatTimer();
- }
- } catch (Exception e) {
- LOG.warn("Failed to heartbeat on HiveWriter ", e);
- }
- }
- }, options.getHeartBeatInterval() * 1000);
- }
- }
-
- private void sendHeartBeatOnAllWriters() throws InterruptedException {
- for (HiveWriter writer : allWriters.values()) {
- writer.heartBeat();
- }
- }
-
- void flushAllWriters(boolean rollToNext)
- throws HiveWriter.CommitFailure, HiveWriter.TxnBatchFailure, HiveWriter.TxnFailure, InterruptedException {
- for (HiveWriter writer : allWriters.values()) {
- writer.flush(rollToNext);
- }
- }
-
- void abortAndCloseWriters() {
- try {
- abortAllWriters();
- closeAllWriters();
- } catch (Exception ie) {
- LOG.warn("unable to close hive connections. ", ie);
- }
- }
-
- /**
- * Abort current Txn on all writers.
- */
- private void abortAllWriters() throws InterruptedException, StreamingException, HiveWriter.TxnBatchFailure {
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- try {
- entry.getValue().abort();
- } catch (Exception e) {
- LOG.error("Failed to abort hive transaction batch, HiveEndPoint " + entry.getValue() + " due to exception ", e);
- }
- }
- }
-
- /**
- * Closes all writers and remove them from cache.
- */
- private void closeAllWriters() {
- //1) Retire writers
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- try {
- entry.getValue().close();
- } catch (Exception e) {
- LOG.warn("unable to close writers. ", e);
- }
- }
- //2) Clear cache
- allWriters.clear();
- }
-
- @VisibleForTesting
- HiveWriter getOrCreateWriter(HiveEndPoint endPoint)
- throws HiveWriter.ConnectFailure, InterruptedException {
- try {
- HiveWriter writer = allWriters.get(endPoint);
- if (writer == null) {
- LOG.debug("Creating Writer to Hive end point : " + endPoint);
- writer = HiveUtils.makeHiveWriter(endPoint, callTimeoutPool, ugi, options, tokenAuthEnabled);
- if (allWriters.size() > (options.getMaxOpenConnections() - 1)) {
- LOG.info("cached HiveEndPoint size {} exceeded maxOpenConnections {} ", allWriters.size(),
- options.getMaxOpenConnections());
- int retired = retireIdleWriters();
- if (retired == 0) {
- retireEldestWriter();
- }
- }
- allWriters.put(endPoint, writer);
- HiveUtils.logAllHiveEndPoints(allWriters);
- }
- return writer;
- } catch (HiveWriter.ConnectFailure e) {
- LOG.error("Failed to create HiveWriter for endpoint: " + endPoint, e);
- throw e;
- }
- }
-
- /**
- * Locate writer that has not been used for longest time and retire it.
- */
- private void retireEldestWriter() {
- LOG.info("Attempting close eldest writers");
- long oldestTimeStamp = System.currentTimeMillis();
- HiveEndPoint eldest = null;
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- if (entry.getValue().getLastUsed() < oldestTimeStamp) {
- eldest = entry.getKey();
- oldestTimeStamp = entry.getValue().getLastUsed();
- }
- }
- try {
- LOG.info("Closing least used Writer to Hive end point : " + eldest);
- allWriters.remove(eldest).flushAndClose();
- } catch (IOException e) {
- LOG.warn("Failed to close writer for end point: " + eldest, e);
- } catch (InterruptedException e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + eldest, e);
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + eldest, e);
- }
- }
-
- /**
- * Locate all writers past idle timeout and retire them.
- * @return number of writers retired
- */
- private int retireIdleWriters() {
- LOG.info("Attempting close idle writers");
- int count = 0;
- long now = System.currentTimeMillis();
-
- //1) Find retirement candidates
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- if (now - entry.getValue().getLastUsed() > options.getIdleTimeout()) {
- ++count;
- retire(entry.getKey());
- }
- }
- return count;
- }
-
- private void retire(HiveEndPoint ep) {
- try {
- HiveWriter writer = allWriters.remove(ep);
- if (writer != null) {
- LOG.info("Closing idle Writer to Hive end point : {}", ep);
- writer.flushAndClose();
- }
- } catch (IOException e) {
- LOG.warn("Failed to close writer for end point: {}. Error: " + ep, e);
- } catch (InterruptedException e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + ep, e);
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + ep, e);
- }
- }
-
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/DelimitedRecordHiveMapper.java b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/DelimitedRecordHiveMapper.java
deleted file mode 100644
index 0e3af0463..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/DelimitedRecordHiveMapper.java
+++ /dev/null
@@ -1,142 +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.storm.hive.bolt.mapper;
-
-import com.google.common.annotations.VisibleForTesting;
-import java.io.IOException;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import org.apache.hive.hcatalog.streaming.DelimitedInputWriter;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.RecordWriter;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.hive.hcatalog.streaming.TransactionBatch;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class DelimitedRecordHiveMapper implements HiveMapper {
- private static final Logger LOG = LoggerFactory.getLogger(DelimitedRecordHiveMapper.class);
- private static final String DEFAULT_FIELD_DELIMITER = ",";
- private Fields columnFields;
- private Fields partitionFields;
- private String[] columnNames;
- private String timeFormat;
- private String fieldDelimiter = DEFAULT_FIELD_DELIMITER;
- private SimpleDateFormat parseDate;
-
- public DelimitedRecordHiveMapper() {
- }
-
- public DelimitedRecordHiveMapper withColumnFields(Fields columnFields) {
- this.columnFields = columnFields;
- List<String> tempColumnNamesList = this.columnFields.toList();
- columnNames = new String[tempColumnNamesList.size()];
- tempColumnNamesList.toArray(columnNames);
- return this;
- }
-
- public DelimitedRecordHiveMapper withPartitionFields(Fields partitionFields) {
- this.partitionFields = partitionFields;
- return this;
- }
-
- public DelimitedRecordHiveMapper withFieldDelimiter(String delimiter) {
- this.fieldDelimiter = delimiter;
- return this;
- }
-
- public DelimitedRecordHiveMapper withTimeAsPartitionField(String timeFormat) {
- this.timeFormat = timeFormat;
- parseDate = new SimpleDateFormat(timeFormat);
- return this;
- }
-
- @Override
- public RecordWriter createRecordWriter(HiveEndPoint endPoint)
- throws StreamingException, IOException, ClassNotFoundException {
- return new DelimitedInputWriter(columnNames, fieldDelimiter, endPoint);
- }
-
- @Override
- public void write(TransactionBatch txnBatch, Tuple tuple)
- throws StreamingException, IOException, InterruptedException {
- txnBatch.write(mapRecord(tuple));
- }
-
- @Override
- public List<String> mapPartitions(Tuple tuple) {
- List<String> partitionList = new ArrayList<String>();
- if (this.partitionFields != null) {
- for (String field : this.partitionFields) {
- partitionList.add(tuple.getStringByField(field));
- }
- }
- if (this.timeFormat != null) {
- partitionList.add(getPartitionsByTimeFormat());
- }
- return partitionList;
- }
-
- @Override
- public List<String> mapPartitions(TridentTuple tuple) {
- List<String> partitionList = new ArrayList<String>();
- if (this.partitionFields != null) {
- for (String field : this.partitionFields) {
- partitionList.add(tuple.getStringByField(field));
- }
- }
- if (this.timeFormat != null) {
- partitionList.add(getPartitionsByTimeFormat());
- }
- return partitionList;
- }
-
- @Override
- public byte[] mapRecord(Tuple tuple) {
- StringBuilder builder = new StringBuilder();
- if (this.columnFields != null) {
- for (String field : this.columnFields) {
- builder.append(tuple.getValueByField(field));
- builder.append(fieldDelimiter);
- }
- }
- return builder.toString().getBytes();
- }
-
- @Override
- public byte[] mapRecord(TridentTuple tuple) {
- StringBuilder builder = new StringBuilder();
- if (this.columnFields != null) {
- for (String field : this.columnFields) {
- builder.append(tuple.getValueByField(field));
- builder.append(fieldDelimiter);
- }
- }
- return builder.toString().getBytes();
- }
-
- private String getPartitionsByTimeFormat() {
- return parseDate.format(System.currentTimeMillis());
- }
-
- @VisibleForTesting
- public String getFieldDelimiter() {
- return fieldDelimiter;
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/HiveMapper.java b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/HiveMapper.java
deleted file mode 100644
index bb9e5f0f0..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/HiveMapper.java
+++ /dev/null
@@ -1,62 +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.storm.hive.bolt.mapper;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.List;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.RecordWriter;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.hive.hcatalog.streaming.TransactionBatch;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Tuple;
-
-/**
- * Maps a <code>org.apache.storm.tuple.Tupe</code> object
- * to a row in an Hive table.
- */
-@Deprecated(since = "2.7.1", forRemoval = true)
-public interface HiveMapper extends Serializable {
-
- /**
- * Given a endPoint, returns a RecordWriter with columnNames.
- */
-
- RecordWriter createRecordWriter(HiveEndPoint endPoint)
- throws StreamingException, IOException, ClassNotFoundException;
-
- void write(TransactionBatch txnBatch, Tuple tuple)
- throws StreamingException, IOException, InterruptedException;
-
- /**
- * Given a tuple, return a hive partition values list.
- */
- List<String> mapPartitions(Tuple tuple);
-
- /**
- * Given a TridetnTuple, return a hive partition values list.
- */
- List<String> mapPartitions(TridentTuple tuple);
-
- /**
- * Given a tuple, maps to a HiveRecord based on columnFields.
- */
- byte[] mapRecord(Tuple tuple);
-
- /**
- * Given a TridentTuple, maps to a HiveRecord based on columnFields.
- */
- byte[] mapRecord(TridentTuple tuple);
-
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/JsonRecordHiveMapper.java b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/JsonRecordHiveMapper.java
deleted file mode 100644
index a380704f3..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/JsonRecordHiveMapper.java
+++ /dev/null
@@ -1,123 +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.storm.hive.bolt.mapper;
-
-import java.io.IOException;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.List;
-import net.minidev.json.JSONObject;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.RecordWriter;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.hive.hcatalog.streaming.StrictJsonWriter;
-import org.apache.hive.hcatalog.streaming.TransactionBatch;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class JsonRecordHiveMapper implements HiveMapper {
- private static final Logger LOG = LoggerFactory.getLogger(JsonRecordHiveMapper.class);
- private Fields columnFields;
- private Fields partitionFields;
- private String timeFormat;
- private SimpleDateFormat parseDate;
-
- public JsonRecordHiveMapper() {
- }
-
- public JsonRecordHiveMapper withColumnFields(Fields columnFields) {
- this.columnFields = columnFields;
- return this;
- }
-
- public JsonRecordHiveMapper withPartitionFields(Fields partitionFields) {
- this.partitionFields = partitionFields;
- return this;
- }
-
- public JsonRecordHiveMapper withTimeAsPartitionField(String timeFormat) {
- this.timeFormat = timeFormat;
- parseDate = new SimpleDateFormat(timeFormat);
- return this;
- }
-
- @Override
- public RecordWriter createRecordWriter(HiveEndPoint endPoint)
- throws StreamingException, IOException, ClassNotFoundException {
- return new StrictJsonWriter(endPoint);
- }
-
- @Override
- public void write(TransactionBatch txnBatch, Tuple tuple)
- throws StreamingException, IOException, InterruptedException {
- txnBatch.write(mapRecord(tuple));
- }
-
- @Override
- public List<String> mapPartitions(Tuple tuple) {
- List<String> partitionList = new ArrayList<String>();
- if (this.partitionFields != null) {
- for (String field : this.partitionFields) {
- partitionList.add(tuple.getStringByField(field));
- }
- }
- if (this.timeFormat != null) {
- partitionList.add(getPartitionsByTimeFormat());
- }
- return partitionList;
- }
-
- @Override
- public List<String> mapPartitions(TridentTuple tuple) {
- List<String> partitionList = new ArrayList<String>();
- if (this.partitionFields != null) {
- for (String field : this.partitionFields) {
- partitionList.add(tuple.getStringByField(field));
- }
- }
- if (this.timeFormat != null) {
- partitionList.add(getPartitionsByTimeFormat());
- }
- return partitionList;
- }
-
- @Override
- public byte[] mapRecord(Tuple tuple) {
- JSONObject obj = new JSONObject();
- if (this.columnFields != null) {
- for (String field : this.columnFields) {
- obj.put(field, tuple.getValueByField(field));
- }
- }
- return obj.toJSONString().getBytes();
- }
-
- @Override
- public byte[] mapRecord(TridentTuple tuple) {
- JSONObject obj = new JSONObject();
- if (this.columnFields != null) {
- for (String field : this.columnFields) {
- obj.put(field, tuple.getValueByField(field));
- }
- }
- return obj.toJSONString().getBytes();
- }
-
- private String getPartitionsByTimeFormat() {
- return parseDate.format(System.currentTimeMillis());
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveOptions.java b/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveOptions.java
deleted file mode 100644
index 1d4f042d2..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveOptions.java
+++ /dev/null
@@ -1,154 +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.storm.hive.common;
-
-import java.io.Serializable;
-import org.apache.storm.hive.bolt.mapper.HiveMapper;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveOptions implements Serializable {
- /**
- * Half of the default Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS
- */
- public static final int DEFAULT_TICK_TUPLE_INTERVAL_SECS = 15;
-
- protected HiveMapper mapper;
- protected String databaseName;
- protected String tableName;
- @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
- protected String metaStoreURI;
- protected Integer txnsPerBatch = 100;
- protected Integer maxOpenConnections = 10;
- protected Integer batchSize = 15000;
- protected Integer idleTimeout = 60000;
- protected Integer callTimeout = 0;
- protected Integer heartBeatInterval = 60;
- protected Boolean autoCreatePartitions = true;
- protected String kerberosPrincipal;
- protected String kerberosKeytab;
- protected Integer tickTupleInterval = DEFAULT_TICK_TUPLE_INTERVAL_SECS;
-
- public HiveOptions(String metaStoreUri, String databaseName, String tableName, HiveMapper mapper) {
- this.metaStoreURI = metaStoreUri;
- this.databaseName = databaseName;
- this.tableName = tableName;
- this.mapper = mapper;
- }
-
- public HiveOptions withTickTupleInterval(Integer tickInterval) {
- this.tickTupleInterval = tickInterval;
- return this;
- }
-
- public HiveOptions withTxnsPerBatch(Integer txnsPerBatch) {
- this.txnsPerBatch = txnsPerBatch;
- return this;
- }
-
- public HiveOptions withMaxOpenConnections(Integer maxOpenConnections) {
- this.maxOpenConnections = maxOpenConnections;
- return this;
- }
-
- public HiveOptions withBatchSize(Integer batchSize) {
- this.batchSize = batchSize;
- return this;
- }
-
- public HiveOptions withIdleTimeout(Integer idleTimeout) {
- this.idleTimeout = idleTimeout;
- return this;
- }
-
- public HiveOptions withCallTimeout(Integer callTimeout) {
- this.callTimeout = callTimeout;
- return this;
- }
-
- public HiveOptions withHeartBeatInterval(Integer heartBeatInterval) {
- this.heartBeatInterval = heartBeatInterval;
- return this;
- }
-
- public HiveOptions withAutoCreatePartitions(Boolean autoCreatePartitions) {
- this.autoCreatePartitions = autoCreatePartitions;
- return this;
- }
-
- public HiveOptions withKerberosKeytab(String kerberosKeytab) {
- this.kerberosKeytab = kerberosKeytab;
- return this;
- }
-
- public HiveOptions withKerberosPrincipal(String kerberosPrincipal) {
- this.kerberosPrincipal = kerberosPrincipal;
- return this;
- }
-
- @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
- public String getMetaStoreURI() {
- return metaStoreURI;
- }
-
- public String getDatabaseName() {
- return databaseName;
- }
-
- public String getTableName() {
- return tableName;
- }
-
- public HiveMapper getMapper() {
- return mapper;
- }
-
- public Integer getBatchSize() {
- return batchSize;
- }
-
- public Integer getCallTimeOut() {
- return callTimeout;
- }
-
- public Integer getHeartBeatInterval() {
- return heartBeatInterval;
- }
-
- public Integer getMaxOpenConnections() {
- return maxOpenConnections;
- }
-
- public Integer getIdleTimeout() {
- return idleTimeout;
- }
-
- public Integer getTxnsPerBatch() {
- return txnsPerBatch;
- }
-
- public Boolean getAutoCreatePartitions() {
- return autoCreatePartitions;
- }
-
- public String getKerberosPrincipal() {
- return kerberosPrincipal;
- }
-
- public String getKerberosKeytab() {
- return kerberosKeytab;
- }
-
- public Integer getTickTupleInterval() {
- return tickTupleInterval;
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveUtils.java b/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveUtils.java
deleted file mode 100644
index 0265ad769..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveUtils.java
+++ /dev/null
@@ -1,115 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.storm.hive.common;
-
-import static org.apache.storm.Config.TOPOLOGY_AUTO_CREDENTIALS;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hive.hcatalog.streaming.ConnectionError;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.storm.hive.security.AutoHive;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveUtils {
- private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
-
- public static HiveEndPoint makeEndPoint(List<String> partitionVals, HiveOptions options) throws ConnectionError {
- if (partitionVals == null) {
- return new HiveEndPoint(options.getMetaStoreURI(), options.getDatabaseName(), options.getTableName(), null);
- }
- return new HiveEndPoint(options.getMetaStoreURI(), options.getDatabaseName(), options.getTableName(), partitionVals);
- }
-
- public static HiveWriter makeHiveWriter(HiveEndPoint endPoint, ExecutorService callTimeoutPool, UserGroupInformation ugi,
- HiveOptions options, boolean tokenAuthEnabled)
- throws HiveWriter.ConnectFailure, InterruptedException {
- return new HiveWriter(endPoint, options.getTxnsPerBatch(), options.getAutoCreatePartitions(),
- options.getCallTimeOut(), callTimeoutPool, options.getMapper(), ugi, tokenAuthEnabled);
- }
-
- public static synchronized UserGroupInformation authenticate(boolean isTokenAuthEnabled, String keytab, String principal) throws
- AuthenticationFailed {
-
- if (isTokenAuthEnabled) {
- return getCurrentUser(principal);
- }
-
- boolean kerberosEnabled = false;
-
- if (principal == null && keytab == null) {
- kerberosEnabled = false;
- } else if (principal != null && keytab != null) {
- kerberosEnabled = true;
- } else {
- throw new IllegalArgumentException("To enable Kerberos, need to set both KerberosPrincipal and KerberosKeytab");
- }
-
- if (kerberosEnabled) {
- File kfile = new File(keytab);
-
- if (!(kfile.isFile() && kfile.canRead())) {
- throw new IllegalArgumentException("The keyTab file: " + keytab + " is nonexistent or can't read. "
- + "Please specify a readable keytab file for Kerberos auth.");
- }
-
- try {
- principal = SecurityUtil.getServerPrincipal(principal, "");
- } catch (Exception e) {
- throw new AuthenticationFailed("Host lookup error when resolving principal " + principal, e);
- }
-
- try {
- UserGroupInformation.loginUserFromKeytab(principal, keytab);
- return UserGroupInformation.getLoginUser();
- } catch (IOException e) {
- throw new AuthenticationFailed("Login failed for principal " + principal, e);
- }
- }
-
- return null;
-
- }
-
- public static void logAllHiveEndPoints(Map<HiveEndPoint, HiveWriter> allWriters) {
- for (Map.Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- LOG.info("cached writers {} ", entry.getValue());
- }
- }
-
- public static boolean isTokenAuthEnabled(Map<String, Object> conf) {
- return conf.get(TOPOLOGY_AUTO_CREDENTIALS) != null
- && (((List) conf.get(TOPOLOGY_AUTO_CREDENTIALS)).contains(AutoHive.class.getName()));
- }
-
- private static UserGroupInformation getCurrentUser(String principal) throws AuthenticationFailed {
- try {
- return UserGroupInformation.getCurrentUser();
- } catch (IOException e) {
- throw new AuthenticationFailed("Login failed for principal " + principal, e);
- }
- }
-
- public static class AuthenticationFailed extends Exception {
- public AuthenticationFailed(String reason, Exception cause) {
- super("Kerberos Authentication Failed. " + reason, cause);
- }
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveWriter.java b/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveWriter.java
deleted file mode 100644
index dde370a69..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveWriter.java
+++ /dev/null
@@ -1,479 +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.storm.hive.common;
-
-import com.google.common.annotations.VisibleForTesting;
-import java.io.IOException;
-import java.security.PrivilegedExceptionAction;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.RecordWriter;
-import org.apache.hive.hcatalog.streaming.SerializationError;
-import org.apache.hive.hcatalog.streaming.StreamingConnection;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.hive.hcatalog.streaming.StreamingIOFailure;
-import org.apache.hive.hcatalog.streaming.TransactionBatch;
-import org.apache.storm.hive.bolt.mapper.HiveMapper;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveWriter {
-
- private static final Logger LOG = LoggerFactory
- .getLogger(HiveWriter.class);
-
- private final HiveEndPoint endPoint;
- private final StreamingConnection connection;
- private final int txnsPerBatch;
- private final RecordWriter recordWriter;
- private final ExecutorService callTimeoutPool;
- private final long callTimeout;
- private final Object txnBatchLock = new Object();
- protected boolean closed; // flag indicating HiveWriter was closed
- private TransactionBatch txnBatch;
- private long lastUsed; // time of last flush on this writer
- private boolean autoCreatePartitions;
- private UserGroupInformation ugi;
- private int totalRecords = 0;
-
- public HiveWriter(HiveEndPoint endPoint, int txnsPerBatch,
- boolean autoCreatePartitions, long callTimeout,
- ExecutorService callTimeoutPool, HiveMapper mapper,
- UserGroupInformation ugi, boolean tokenAuthEnabled)
- throws InterruptedException, ConnectFailure {
- try {
- this.autoCreatePartitions = autoCreatePartitions;
- this.callTimeout = callTimeout;
- this.callTimeoutPool = callTimeoutPool;
- this.endPoint = endPoint;
- this.ugi = ugi;
- this.connection = newConnection(ugi, tokenAuthEnabled);
- this.txnsPerBatch = txnsPerBatch;
- this.recordWriter = getRecordWriter(mapper, tokenAuthEnabled);
- this.txnBatch = nextTxnBatch(recordWriter);
- this.closed = false;
- this.lastUsed = System.currentTimeMillis();
- } catch (InterruptedException e) {
- throw e;
- } catch (RuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw new ConnectFailure(endPoint, e);
- }
- }
-
- /**
- * If the current thread has been interrupted, then throws an
- * exception.
- */
- private static void checkAndThrowInterruptedException()
- throws InterruptedException {
- if (Thread.currentThread().interrupted()) {
- throw new InterruptedException("Timed out before Hive call was made. "
- + "Your callTimeout might be set too low or Hive calls are "
- + "taking too long.");
- }
- }
-
- public RecordWriter getRecordWriter(final HiveMapper mapper, final boolean tokenAuthEnabled) throws Exception {
- if (!tokenAuthEnabled) {
- return mapper.createRecordWriter(endPoint);
- }
-
- try {
- return ugi.doAs(
- new PrivilegedExceptionAction<RecordWriter>() {
- @Override
- public RecordWriter run() throws StreamingException, IOException, ClassNotFoundException {
- return mapper.createRecordWriter(endPoint);
- }
- }
- );
- } catch (Exception e) {
- throw new ConnectFailure(endPoint, e);
- }
- }
-
- private HiveConf createHiveConf(String metaStoreUri, boolean tokenAuthEnabled) {
- if (!tokenAuthEnabled) {
- return null;
- }
-
- HiveConf hcatConf = new HiveConf();
- hcatConf.setVar(HiveConf.ConfVars.METASTOREURIS, metaStoreUri);
- hcatConf.setBoolVar(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL, true);
- return hcatConf;
- }
-
- @Override
- public String toString() {
- return "{ "
- + "endPoint = " + endPoint.toString()
- + ", TransactionBatch = " + txnBatch.toString() + " }";
- }
-
- /**
- * Write data.
- */
- public synchronized void write(final byte[] record)
- throws WriteFailure, SerializationError, InterruptedException {
- if (closed) {
- throw new IllegalStateException("This hive streaming writer was closed "
- + "and thus no longer able to write : "
- + endPoint);
- }
- // write the tuple
- try {
- LOG.debug("Writing event to {}", endPoint);
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws StreamingException, InterruptedException {
- txnBatch.write(record);
- totalRecords++;
- return null;
- }
- });
- } catch (SerializationError se) {
- throw new SerializationError(endPoint.toString() + " SerializationError", se);
- } catch (StreamingException e) {
- throw new WriteFailure(endPoint, txnBatch.getCurrentTxnId(), e);
- } catch (TimeoutException e) {
- throw new WriteFailure(endPoint, txnBatch.getCurrentTxnId(), e);
- }
- }
-
- /**
- * Commits the current Txn if totalRecordsPerTransaction > 0 .
- * If 'rollToNext' is true, will switch to next Txn in batch or to a
- * new TxnBatch if current Txn batch is exhausted
- */
- public void flush(boolean rollToNext)
- throws CommitFailure, TxnBatchFailure, TxnFailure, InterruptedException {
- // if there are no records do not call flush
- if (totalRecords <= 0) {
- return;
- }
- try {
- synchronized (txnBatchLock) {
- commitTxn();
- nextTxn(rollToNext);
- totalRecords = 0;
- lastUsed = System.currentTimeMillis();
- }
- } catch (StreamingException e) {
- throw new TxnFailure(txnBatch, e);
- }
- }
-
- /** Queues up a heartbeat request on the current and remaining txns using the
- * heartbeatThdPool and returns immediately.
- */
- public void heartBeat() throws InterruptedException {
- // 1) schedule the heartbeat on one thread in pool
- synchronized (txnBatchLock) {
- try {
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws Exception {
- try {
- LOG.info("Sending heartbeat on batch " + txnBatch);
- txnBatch.heartbeat();
- } catch (StreamingException e) {
- LOG.warn("Heartbeat error on batch " + txnBatch, e);
- }
- return null;
- }
- });
- } catch (InterruptedException e) {
- throw e;
- } catch (Exception e) {
- LOG.warn("Unable to send heartbeat on Txn Batch " + txnBatch, e);
- // Suppressing exceptions as we don't care for errors on heartbeats
- }
- }
- }
-
- /**
- * returns totalRecords written so far in a transaction.
- */
- public int getTotalRecords() {
- return totalRecords;
- }
-
- /**
- * Flush and Close current transactionBatch.
- */
- public void flushAndClose() throws TxnBatchFailure, TxnFailure, CommitFailure,
- IOException, InterruptedException {
- flush(false);
- close();
- }
-
- /**
- * Close the Transaction Batch and connection.
- */
- public void close() throws IOException, InterruptedException {
- closeTxnBatch();
- closeConnection();
- closed = true;
- }
-
- private void closeConnection() throws InterruptedException {
- LOG.info("Closing connection to end point : {}", endPoint);
- try {
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws Exception {
- connection.close(); // could block
- return null;
- }
- });
- } catch (Exception e) {
- LOG.warn("Error closing connection to EndPoint : " + endPoint, e);
- // Suppressing exceptions as we don't care for errors on connection close
- }
- }
-
- private void commitTxn() throws CommitFailure, InterruptedException {
- LOG.debug("Committing Txn id {} to {}", txnBatch.getCurrentTxnId(), endPoint);
- try {
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws Exception {
- txnBatch.commit(); // could block
- return null;
- }
- });
- } catch (StreamingException e) {
- throw new CommitFailure(endPoint, txnBatch.getCurrentTxnId(), e);
- } catch (TimeoutException e) {
- throw new CommitFailure(endPoint, txnBatch.getCurrentTxnId(), e);
- }
- }
-
- @VisibleForTesting
- StreamingConnection newConnection(final UserGroupInformation ugi, final boolean tokenAuthEnabled)
- throws InterruptedException, ConnectFailure {
- try {
- return callWithTimeout(new CallRunner<StreamingConnection>() {
- @Override
- public StreamingConnection call() throws Exception {
- return endPoint
- .newConnection(autoCreatePartitions, createHiveConf(endPoint.metaStoreUri, tokenAuthEnabled), ugi); // could block
- }
- });
- } catch (StreamingException e) {
- throw new ConnectFailure(endPoint, e);
- } catch (TimeoutException e) {
- throw new ConnectFailure(endPoint, e);
- }
- }
-
- private TransactionBatch nextTxnBatch(final RecordWriter recordWriter)
- throws InterruptedException, TxnBatchFailure {
- LOG.debug("Fetching new Txn Batch for {}", endPoint);
- TransactionBatch batch = null;
- try {
- batch = callWithTimeout(new CallRunner<TransactionBatch>() {
- @Override
- public TransactionBatch call() throws Exception {
- return connection.fetchTransactionBatch(txnsPerBatch, recordWriter); // could block
- }
- });
- batch.beginNextTransaction();
- LOG.debug("Acquired {}. Switching to first txn", batch);
- } catch (TimeoutException e) {
- throw new TxnBatchFailure(endPoint, e);
- } catch (StreamingException e) {
- throw new TxnBatchFailure(endPoint, e);
- }
- return batch;
- }
-
- private void closeTxnBatch() throws InterruptedException {
- try {
- LOG.debug("Closing Txn Batch {}", txnBatch);
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws Exception {
- if (txnBatch != null) {
- txnBatch.close(); // could block
- }
- return null;
- }
- });
- } catch (InterruptedException e) {
- throw e;
- } catch (Exception e) {
- LOG.warn("Error closing txn batch " + txnBatch, e);
- }
- }
-
- /**
- * Aborts the current Txn and switches to next Txn.
- * @throws StreamingException if could not get new Transaction Batch, or switch to next Txn
- */
- public void abort() throws StreamingException, TxnBatchFailure, InterruptedException {
- synchronized (txnBatchLock) {
- abortTxn();
- nextTxn(true); // roll to next
- }
- }
-
- /**
- * Aborts current Txn in the txnBatch.
- */
- private void abortTxn() throws InterruptedException {
- LOG.info("Aborting Txn id {} on End Point {}", txnBatch.getCurrentTxnId(), endPoint);
- try {
- callWithTimeout(new CallRunner<Void>() {
- @Override
- public Void call() throws StreamingException, InterruptedException {
- txnBatch.abort(); // could block
- return null;
- }
- });
- } catch (InterruptedException e) {
- throw e;
- } catch (TimeoutException e) {
- LOG.warn("Timeout while aborting Txn " + txnBatch.getCurrentTxnId() + " on EndPoint: " + endPoint, e);
- } catch (Exception e) {
- LOG.warn("Error aborting Txn " + txnBatch.getCurrentTxnId() + " on EndPoint: " + endPoint, e);
- // Suppressing exceptions as we don't care for errors on abort
- }
- }
-
- /**
- * if there are remainingTransactions in current txnBatch, begins nextTransactions
- * otherwise creates new txnBatch.
- */
- private void nextTxn(boolean rollToNext) throws StreamingException, InterruptedException, TxnBatchFailure {
- if (txnBatch.remainingTransactions() == 0) {
- closeTxnBatch();
- txnBatch = null;
- if (rollToNext) {
- txnBatch = nextTxnBatch(recordWriter);
- }
- } else if (rollToNext) {
- LOG.debug("Switching to next Txn for {}", endPoint);
- txnBatch.beginNextTransaction(); // does not block
- }
- }
-
- /**
- * Execute the callable on a separate thread and wait for the completion
- * for the specified amount of time in milliseconds. In case of timeout
- * cancel the callable and throw an IOException
- */
- private <T> T callWithTimeout(final CallRunner<T> callRunner)
- throws TimeoutException, StreamingException, InterruptedException {
- Future<T> future = callTimeoutPool.submit(new Callable<T>() {
- @Override
- public T call() throws Exception {
- return callRunner.call();
- }
- });
- try {
- if (callTimeout > 0) {
- return future.get(callTimeout, TimeUnit.MILLISECONDS);
- } else {
- return future.get();
- }
- } catch (TimeoutException timeoutException) {
- future.cancel(true);
- throw timeoutException;
- } catch (ExecutionException e1) {
- Throwable cause = e1.getCause();
- if (cause instanceof IOException) {
- throw new StreamingIOFailure("I/O Failure", (IOException) cause);
- } else if (cause instanceof StreamingException) {
- throw (StreamingException) cause;
- } else if (cause instanceof InterruptedException) {
- throw (InterruptedException) cause;
- } else if (cause instanceof RuntimeException) {
- throw (RuntimeException) cause;
- } else if (cause instanceof TimeoutException) {
- throw new StreamingException("Operation Timed Out.", (TimeoutException) cause);
- } else {
- throw new RuntimeException(e1);
- }
- }
- }
-
- public long getLastUsed() {
- return lastUsed;
- }
-
- private byte[] generateRecord(Tuple tuple) {
- StringBuilder buf = new StringBuilder();
- for (Object o : tuple.getValues()) {
- buf.append(o);
- buf.append(",");
- }
- return buf.toString().getBytes();
- }
-
- /**
- * Simple interface whose <tt>call</tt> method is called by
- * {#callWithTimeout} in a new thread inside a
- * {@linkplain java.security.PrivilegedExceptionAction#run()} call.
- */
- private interface CallRunner<T> {
- T call() throws Exception;
- }
-
- public static class Failure extends Exception {
- public Failure(String message, Throwable cause) {
- super(message, cause);
- }
- }
-
- public static class WriteFailure extends Failure {
- public WriteFailure(HiveEndPoint endPoint, Long currentTxnId, Throwable cause) {
- super("Failed writing to : " + endPoint + ". TxnID : " + currentTxnId, cause);
- }
- }
-
- public static class CommitFailure extends Failure {
- public CommitFailure(HiveEndPoint endPoint, Long txnId, Throwable cause) {
- super("Commit of Txn " + txnId + " failed on EndPoint: " + endPoint, cause);
- }
- }
-
- public static class ConnectFailure extends Failure {
- public ConnectFailure(HiveEndPoint ep, Throwable cause) {
- super("Failed connecting to EndPoint " + ep, cause);
- }
- }
-
- public static class TxnBatchFailure extends Failure {
- public TxnBatchFailure(HiveEndPoint ep, Throwable cause) {
- super("Failed acquiring Transaction Batch from EndPoint: " + ep, cause);
- }
- }
-
- public static class TxnFailure extends Failure {
- public TxnFailure(TransactionBatch txnBatch, Throwable cause) {
- super("Failed switching to next Txn in TxnBatch " + txnBatch, cause);
- }
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java b/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
deleted file mode 100644
index 60d1a937e..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
+++ /dev/null
@@ -1,304 +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.storm.hive.trident;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.hive.common.HiveUtils;
-import org.apache.storm.hive.common.HiveWriter;
-import org.apache.storm.task.IMetricsContext;
-import org.apache.storm.topology.FailedException;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.state.State;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveState implements State {
- private static final Logger LOG = LoggerFactory.getLogger(HiveState.class);
- private HiveOptions options;
- private Integer currentBatchSize;
- private ExecutorService callTimeoutPool;
- private transient Timer heartBeatTimer;
- private AtomicBoolean timeToSendHeartBeat = new AtomicBoolean(false);
- private Boolean sendHeartBeat = true;
- private UserGroupInformation ugi = null;
- private Boolean kerberosEnabled = false;
- private Map<HiveEndPoint, HiveWriter> allWriters;
- private boolean tokenAuthEnabled;
-
- public HiveState(HiveOptions options) {
- this.options = options;
- this.currentBatchSize = 0;
- }
-
-
- @Override
- public void beginCommit(Long txId) {
- }
-
- @Override
- public void commit(Long txId) {
- try {
- flushAllWriters();
- currentBatchSize = 0;
- } catch (HiveWriter.TxnFailure | InterruptedException | HiveWriter.CommitFailure | HiveWriter.TxnBatchFailure ex) {
- LOG.warn("Commit failed. Failing the batch.", ex);
- throw new FailedException(ex);
- }
- }
-
- public void prepare(Map<String, Object> conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
- try {
- tokenAuthEnabled = HiveUtils.isTokenAuthEnabled(conf);
- try {
- ugi = HiveUtils.authenticate(tokenAuthEnabled, options.getKerberosKeytab(), options.getKerberosPrincipal());
- } catch (HiveUtils.AuthenticationFailed ex) {
- LOG.error("Hive kerberos authentication failed " + ex.getMessage(), ex);
- throw new IllegalArgumentException(ex);
- }
-
- allWriters = new ConcurrentHashMap<HiveEndPoint, HiveWriter>();
- String timeoutName = "hive-bolt-%d";
- this.callTimeoutPool = Executors.newFixedThreadPool(1,
- new ThreadFactoryBuilder().setNameFormat(timeoutName).build());
- heartBeatTimer = new Timer("hive-hb-timer", true);
- setupHeartBeatTimer();
- } catch (Exception e) {
- LOG.warn("unable to make connection to hive ", e);
- }
- }
-
- public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
- try {
- writeTuples(tuples);
- } catch (Exception e) {
- abortAndCloseWriters();
- LOG.warn("hive streaming failed.", e);
- throw new FailedException(e);
- }
- }
-
- private void writeTuples(List<TridentTuple> tuples)
- throws Exception {
- for (TridentTuple tuple : tuples) {
- List<String> partitionVals = options.getMapper().mapPartitions(tuple);
- HiveEndPoint endPoint = HiveUtils.makeEndPoint(partitionVals, options);
- HiveWriter writer = getOrCreateWriter(endPoint);
- writer.write(options.getMapper().mapRecord(tuple));
- currentBatchSize++;
- if (currentBatchSize >= options.getBatchSize()) {
- flushAllWriters();
- currentBatchSize = 0;
- }
- }
- }
-
- private void abortAndCloseWriters() {
- try {
- sendHeartBeat = false;
- abortAllWriters();
- closeAllWriters();
- } catch (Exception ie) {
- LOG.warn("unable to close hive connections. ", ie);
- }
- }
-
- /**
- * Abort current Txn on all writers.
- */
- private void abortAllWriters() throws InterruptedException, StreamingException, HiveWriter.TxnBatchFailure {
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- entry.getValue().abort();
- }
- }
-
-
- /**
- * Closes all writers and remove them from cache.
- * @return number of writers retired
- */
- private void closeAllWriters() throws InterruptedException, IOException {
- //1) Retire writers
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- entry.getValue().close();
- }
- //2) Clear cache
- allWriters.clear();
- }
-
- private void setupHeartBeatTimer() {
- if (options.getHeartBeatInterval() > 0) {
- heartBeatTimer.schedule(new TimerTask() {
- @Override
- public void run() {
- try {
- if (sendHeartBeat) {
- LOG.debug("Start sending heartbeat on all writers");
- sendHeartBeatOnAllWriters();
- setupHeartBeatTimer();
- }
- } catch (Exception e) {
- LOG.warn("Failed to heartbeat on HiveWriter ", e);
- }
- }
- }, options.getHeartBeatInterval() * 1000);
- }
- }
-
- private void flushAllWriters()
- throws HiveWriter.CommitFailure, HiveWriter.TxnBatchFailure, HiveWriter.TxnFailure, InterruptedException {
- for (HiveWriter writer : allWriters.values()) {
- writer.flush(true);
- }
- }
-
- private void sendHeartBeatOnAllWriters() throws InterruptedException {
- for (HiveWriter writer : allWriters.values()) {
- writer.heartBeat();
- }
- }
-
- private HiveWriter getOrCreateWriter(HiveEndPoint endPoint)
- throws HiveWriter.ConnectFailure, InterruptedException {
- try {
- HiveWriter writer = allWriters.get(endPoint);
- if (writer == null) {
- LOG.info("Creating Writer to Hive end point : " + endPoint);
- writer = HiveUtils.makeHiveWriter(endPoint, callTimeoutPool, ugi, options, tokenAuthEnabled);
- if (allWriters.size() > (options.getMaxOpenConnections() - 1)) {
- int retired = retireIdleWriters();
- if (retired == 0) {
- retireEldestWriter();
- }
- }
- allWriters.put(endPoint, writer);
- }
- return writer;
- } catch (HiveWriter.ConnectFailure e) {
- LOG.error("Failed to create HiveWriter for endpoint: " + endPoint, e);
- throw e;
- }
-
- }
-
-
- /**
- * Locate writer that has not been used for longest time and retire it.
- */
- private void retireEldestWriter() {
- long oldestTimeStamp = System.currentTimeMillis();
- HiveEndPoint eldest = null;
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- if (entry.getValue().getLastUsed() < oldestTimeStamp) {
- eldest = entry.getKey();
- oldestTimeStamp = entry.getValue().getLastUsed();
- }
- }
- try {
- LOG.info("Closing least used Writer to Hive end point : " + eldest);
- allWriters.remove(eldest).flushAndClose();
- } catch (IOException e) {
- LOG.warn("Failed to close writer for end point: " + eldest, e);
- } catch (InterruptedException e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + eldest, e);
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + eldest, e);
- }
- }
-
- /**
- * Locate all writers past idle timeout and retire them.
- * @return number of writers retired
- */
- private int retireIdleWriters() {
- int count = 0;
- long now = System.currentTimeMillis();
- ArrayList<HiveEndPoint> retirees = new ArrayList<HiveEndPoint>();
-
- //1) Find retirement candidates
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- if (now - entry.getValue().getLastUsed() > options.getIdleTimeout()) {
- ++count;
- retirees.add(entry.getKey());
- }
- }
- //2) Retire them
- for (HiveEndPoint ep : retirees) {
- try {
- LOG.info("Closing idle Writer to Hive end point : {}", ep);
- allWriters.remove(ep).flushAndClose();
- } catch (IOException e) {
- LOG.warn("Failed to close writer for end point: {}. Error: " + ep, e);
- } catch (InterruptedException e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + ep, e);
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- LOG.warn("Interrupted when attempting to close writer for end point: " + ep, e);
- }
- }
- return count;
- }
-
- public void cleanup() {
- for (Entry<HiveEndPoint, HiveWriter> entry : allWriters.entrySet()) {
- try {
- sendHeartBeat = false;
- HiveWriter w = entry.getValue();
- LOG.info("Flushing writer to {}", w);
- w.flush(false);
- LOG.info("Closing writer to {}", w);
- w.close();
- } catch (Exception ex) {
- LOG.warn("Error while closing writer to " + entry.getKey() + ". Exception follows.",
- ex);
- if (ex instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- }
- }
-
- ExecutorService[] toShutdown = { callTimeoutPool };
- for (ExecutorService execService : toShutdown) {
- execService.shutdown();
- try {
- while (!execService.isTerminated()) {
- execService.awaitTermination(
- options.getCallTimeOut(), TimeUnit.MILLISECONDS);
- }
- } catch (InterruptedException ex) {
- LOG.warn("shutdown interrupted on " + execService, ex);
- }
- }
- heartBeatTimer.cancel();
- callTimeoutPool = null;
- }
-
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveStateFactory.java b/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveStateFactory.java
deleted file mode 100644
index d587bfb56..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveStateFactory.java
+++ /dev/null
@@ -1,49 +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.storm.hive.trident;
-
-import java.util.Map;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.task.IMetricsContext;
-import org.apache.storm.trident.state.State;
-import org.apache.storm.trident.state.StateFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveStateFactory implements StateFactory {
- private static final Logger LOG = LoggerFactory.getLogger(HiveStateFactory.class);
- private HiveOptions options;
-
- public HiveStateFactory() {}
-
- /**
- * The options for connecting to Hive.
- */
- public HiveStateFactory withOptions(HiveOptions options) {
- if (options.getTickTupleInterval() != HiveOptions.DEFAULT_TICK_TUPLE_INTERVAL_SECS) {
- LOG.error("Tick tuple interval will be ignored for trident."
- + " The Hive writers are flushed after each batch.");
- }
- this.options = options;
- return this;
- }
-
- @Override
- public State makeState(Map<String, Object> conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
- LOG.info("makeState(partitonIndex={}, numpartitions={}", partitionIndex, numPartitions);
- HiveState state = new HiveState(this.options);
- state.prepare(conf, metrics, partitionIndex, numPartitions);
- return state;
- }
-}
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveUpdater.java b/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveUpdater.java
deleted file mode 100644
index 47beda81c..000000000
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveUpdater.java
+++ /dev/null
@@ -1,26 +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.storm.hive.trident;
-
-import java.util.List;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.state.BaseStateUpdater;
-import org.apache.storm.trident.tuple.TridentTuple;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveUpdater extends BaseStateUpdater<HiveState> {
- @Override
- public void updateState(HiveState state, List<TridentTuple> tuples, TridentCollector collector) {
- state.updateState(tuples, collector);
- }
-}
diff --git a/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/HiveSetupUtil.java b/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/HiveSetupUtil.java
deleted file mode 100644
index 39a87b914..000000000
--- a/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/HiveSetupUtil.java
+++ /dev/null
@@ -1,214 +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.storm.hive.bolt;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RawLocalFileSystem;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
-import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.metastore.TableType;
-import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
-import org.apache.hadoop.hive.metastore.api.Database;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.SerDeInfo;
-import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
-import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
-import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
-import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.serde.serdeConstants;
-import org.apache.hadoop.hive.shims.ShimLoader;
-import org.apache.thrift.TException;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class HiveSetupUtil {
- private final static String txnMgr = "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager";
-
- public static HiveConf getHiveConf() {
- HiveConf conf = new HiveConf();
- // String metastoreDBLocation = "jdbc:derby:databaseName=/tmp/metastore_db;create=true";
- // conf.set("javax.jdo.option.ConnectionDriverName","org.apache.derby.jdbc.EmbeddedDriver");
- // conf.set("javax.jdo.option.ConnectionURL",metastoreDBLocation);
- conf.set("fs.raw.impl", RawFileSystem.class.getName());
- conf.setVar(HiveConf.ConfVars.HIVE_TXN_MANAGER, txnMgr);
- conf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, true);
- return conf;
- }
-
- public static void createDbAndTable(HiveConf conf, String databaseName,
- String tableName, List<String> partVals,
- String[] colNames, String[] colTypes,
- String[] partNames, String dbLocation)
- throws Exception {
- IMetaStoreClient client = new HiveMetaStoreClient(conf);
- try {
- Database db = new Database();
- db.setName(databaseName);
- db.setLocationUri(dbLocation);
- client.createDatabase(db);
-
- Table tbl = new Table();
- tbl.setDbName(databaseName);
- tbl.setTableName(tableName);
- tbl.setTableType(TableType.MANAGED_TABLE.toString());
- StorageDescriptor sd = new StorageDescriptor();
- sd.setCols(getTableColumns(colNames, colTypes));
- sd.setNumBuckets(1);
- sd.setLocation(dbLocation + Path.SEPARATOR + tableName);
- if (partNames != null && partNames.length != 0) {
- tbl.setPartitionKeys(getPartitionKeys(partNames));
- }
-
- tbl.setSd(sd);
-
- sd.setBucketCols(new ArrayList<String>(2));
- sd.setSerdeInfo(new SerDeInfo());
- sd.getSerdeInfo().setName(tbl.getTableName());
- sd.getSerdeInfo().setParameters(new HashMap<String, String>());
- sd.getSerdeInfo().getParameters().put(serdeConstants.SERIALIZATION_FORMAT, "1");
-
- sd.getSerdeInfo().setSerializationLib(OrcSerde.class.getName());
- sd.setInputFormat(OrcInputFormat.class.getName());
- sd.setOutputFormat(OrcOutputFormat.class.getName());
-
- Map<String, String> tableParams = new HashMap<String, String>();
- tbl.setParameters(tableParams);
- client.createTable(tbl);
- try {
- if (partVals != null && partVals.size() > 0) {
- addPartition(client, tbl, partVals);
- }
- } catch (AlreadyExistsException e) {
- }
- } finally {
- client.close();
- }
- }
-
- // delete db and all tables in it
- public static void dropDB(HiveConf conf, String databaseName) throws HiveException, MetaException {
- IMetaStoreClient client = new HiveMetaStoreClient(conf);
- try {
- for (String table : client.listTableNamesByFilter(databaseName, "", (short) -1)) {
- client.dropTable(databaseName, table, true, true);
- }
- client.dropDatabase(databaseName);
- } catch (TException e) {
- client.close();
- }
- }
-
- private static void addPartition(IMetaStoreClient client, Table tbl
- , List<String> partValues)
- throws IOException, TException {
- Partition part = new Partition();
- part.setDbName(tbl.getDbName());
- part.setTableName(tbl.getTableName());
- StorageDescriptor sd = new StorageDescriptor(tbl.getSd());
- sd.setLocation(sd.getLocation() + Path.SEPARATOR + makePartPath(tbl.getPartitionKeys(), partValues));
- part.setSd(sd);
- part.setValues(partValues);
- client.add_partition(part);
- }
-
- private static String makePartPath(List<FieldSchema> partKeys, List<String> partVals) {
- if (partKeys.size() != partVals.size()) {
- throw new IllegalArgumentException("Partition values:" + partVals +
- ", does not match the partition Keys in table :" + partKeys);
- }
- StringBuffer buff = new StringBuffer(partKeys.size() * 20);
- int i = 0;
- for (FieldSchema schema : partKeys) {
- buff.append(schema.getName());
- buff.append("=");
- buff.append(partVals.get(i));
- if (i != partKeys.size() - 1) {
- buff.append(Path.SEPARATOR);
- }
- ++i;
- }
- return buff.toString();
- }
-
- private static List<FieldSchema> getTableColumns(String[] colNames, String[] colTypes) {
- List<FieldSchema> fields = new ArrayList<FieldSchema>();
- for (int i = 0; i < colNames.length; ++i) {
- fields.add(new FieldSchema(colNames[i], colTypes[i], ""));
- }
- return fields;
- }
-
- private static List<FieldSchema> getPartitionKeys(String[] partNames) {
- List<FieldSchema> fields = new ArrayList<FieldSchema>();
- for (int i = 0; i < partNames.length; ++i) {
- fields.add(new FieldSchema(partNames[i], serdeConstants.STRING_TYPE_NAME, ""));
- }
- return fields;
- }
-
- public static class RawFileSystem extends RawLocalFileSystem {
- private static final URI NAME;
-
- static {
- try {
- NAME = new URI("raw:///");
- } catch (URISyntaxException se) {
- throw new IllegalArgumentException("bad uri", se);
- }
- }
-
- @Override
- public URI getUri() {
- return NAME;
- }
-
- @Override
- public FileStatus getFileStatus(Path path) throws IOException {
- File file = pathToFile(path);
- if (!file.exists()) {
- throw new FileNotFoundException("Can't find " + path);
- }
- // get close enough
- short mod = 0;
- if (file.canRead()) {
- mod |= 0444;
- }
- if (file.canWrite()) {
- mod |= 0200;
- }
- if (file.canExecute()) {
- mod |= 0111;
- }
- ShimLoader.getHadoopShims();
- return new FileStatus(file.length(), file.isDirectory(), 1, 1024,
- file.lastModified(), file.lastModified(),
- FsPermission.createImmutable(mod), "owen", "users", path);
- }
- }
-
-}
diff --git a/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/TestHiveBolt.java b/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/TestHiveBolt.java
deleted file mode 100644
index 9997ba141..000000000
--- a/external/storm-hive/src/test/java/org/apache/storm/hive/bolt/TestHiveBolt.java
+++ /dev/null
@@ -1,527 +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.storm.hive.bolt;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.Lists;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
-import org.apache.hadoop.hive.serde.serdeConstants;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.storm.Config;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.bolt.mapper.JsonRecordHiveMapper;
-import org.apache.storm.hive.common.HiveOptions;
-import org.apache.storm.hive.common.HiveWriter;
-import org.apache.storm.task.GeneralTopologyContext;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.TupleImpl;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.MockTupleHelpers;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.MockitoAnnotations;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.junit.jupiter.api.Assertions.*;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class TestHiveBolt {
- final static String dbName = "testdb";
- final static String tblName = "test_table";
- final static String dbName1 = "testdb1";
- final static String tblName1 = "test_table1";
- final static String PART1_NAME = "city";
- final static String PART2_NAME = "state";
- final static String[] partNames = { PART1_NAME, PART2_NAME };
- private static final String COL1 = "id";
- private static final String COL2 = "msg";
- private static final Logger LOG = LoggerFactory.getLogger(HiveBolt.class);
- final String partitionVals = "sunnyvale,ca";
- final String[] colNames = { COL1, COL2 };
- final String[] colNames1 = { COL2, COL1 };
- final String metaStoreURI;
- private final HiveConf conf;
- private String[] colTypes = { serdeConstants.INT_TYPE_NAME, serdeConstants.STRING_TYPE_NAME };
- private Config config = new Config();
- private TestingHiveBolt bolt;
- ;
- private ObjectMapper objectMapper = new ObjectMapper();
- @Mock
- private OutputCollector collector;
-
- public TestHiveBolt() throws Exception {
- //metaStoreURI = "jdbc:derby:;databaseName="+System.getProperty("java.io.tmpdir") +"metastore_db;create=true";
- metaStoreURI = null;
- conf = HiveSetupUtil.getHiveConf();
- TxnDbUtil.setConfValues(conf);
- if (metaStoreURI != null) {
- conf.setVar(HiveConf.ConfVars.METASTOREURIS, metaStoreURI);
- }
- }
-
- @BeforeEach
- public void setup() throws Exception {
- MockitoAnnotations.initMocks(this);
- }
-
- @Test
- public void testWithByteArrayIdandMessage()
- throws Exception {
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(2);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, collector);
-
- Integer id = 100;
- String msg = "test-123";
- String city = "sunnyvale";
- String state = "ca";
-
- Set<Tuple> tupleSet = new HashSet<Tuple>();
- for (int i = 0; i < 4; i++) {
- Tuple tuple = generateTestTuple(id, msg, city, state);
- bolt.execute(tuple);
- tupleSet.add(tuple);
- }
-
- List<String> partVals = Lists.newArrayList(city, state);
-
- for (Tuple t : tupleSet) {
- verify(collector).ack(t);
- }
-
- assertEquals(4, bolt.getRecordWritten(partVals).size());
-
- bolt.cleanup();
- }
-
- @Test
- public void testWithoutPartitions()
- throws Exception {
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName1, tblName1, mapper)
- .withTxnsPerBatch(2).withBatchSize(2).withAutoCreatePartitions(false);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, collector);
-
- Integer id = 100;
- String msg = "test-123";
- String city = "sunnyvale";
- String state = "ca";
-
- Set<Tuple> tupleSet = new HashSet<Tuple>();
- for (int i = 0; i < 4; i++) {
- Tuple tuple = generateTestTuple(id, msg, city, state);
- bolt.execute(tuple);
- tupleSet.add(tuple);
- }
-
- List<String> partVals = Collections.emptyList();
-
- for (Tuple t : tupleSet) {
- verify(collector).ack(t);
- }
-
- List<byte[]> recordWritten = bolt.getRecordWritten(partVals);
- assertNotNull(recordWritten);
- assertEquals(4, recordWritten.size());
-
- bolt.cleanup();
- }
-
- @Test
- public void testWithTimeformat()
- throws Exception {
- String timeFormat = "yyyy/MM/dd";
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withTimeAsPartitionField(timeFormat);
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName1, tblName1, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(1)
- .withMaxOpenConnections(1);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, collector);
-
- Integer id = 100;
- String msg = "test-123";
- Date d = new Date();
- SimpleDateFormat parseDate = new SimpleDateFormat(timeFormat);
- String today = parseDate.format(d.getTime());
-
- List<Tuple> tuples = new ArrayList<>();
- for (int i = 0; i < 2; i++) {
- Tuple tuple = generateTestTuple(id, msg, null, null);
- tuples.add(tuple);
- bolt.execute(tuple);
- }
-
- for (Tuple t : tuples) {
- verify(collector).ack(t);
- }
-
- List<String> partVals = Lists.newArrayList(today);
-
- List<byte[]> recordsWritten = bolt.getRecordWritten(partVals);
- assertNotNull(recordsWritten);
- assertEquals(2, recordsWritten.size());
-
- byte[] mapped = generateDelimiteredRecord(Lists.newArrayList(id, msg), mapper.getFieldDelimiter());
-
- for (byte[] record : recordsWritten) {
- assertArrayEquals(mapped, record);
- }
-
- bolt.cleanup();
- }
-
- @Test
- public void testData()
- throws Exception {
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(1);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, new OutputCollector(collector));
-
- Integer id = 1;
- String msg = "SJC";
- String city = "Sunnyvale";
- String state = "CA";
-
- Tuple tuple1 = generateTestTuple(id, msg, city, state);
-
- bolt.execute(tuple1);
- verify(collector).ack(tuple1);
-
- List<String> partVals = Lists.newArrayList(city, state);
-
- List<byte[]> recordsWritten = bolt.getRecordWritten(partVals);
- assertNotNull(recordsWritten);
- assertEquals(1, recordsWritten.size());
-
- byte[] mapped = generateDelimiteredRecord(Lists.newArrayList(id, msg), mapper.getFieldDelimiter());
- assertArrayEquals(mapped, recordsWritten.get(0));
-
- bolt.cleanup();
- }
-
- @Test
- public void testJsonWriter()
- throws Exception {
- // json record doesn't need columns to be in the same order
- // as table in hive.
- JsonRecordHiveMapper mapper = new JsonRecordHiveMapper()
- .withColumnFields(new Fields(colNames1))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(1);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, collector);
-
- Integer id = 1;
- String msg = "SJC";
- String city = "Sunnyvale";
- String state = "CA";
-
- Tuple tuple1 = generateTestTuple(id, msg, city, state);
-
- bolt.execute(tuple1);
- verify(collector).ack(tuple1);
-
- List<String> partVals = Lists.newArrayList(city, state);
-
- List<byte[]> recordsWritten = bolt.getRecordWritten(partVals);
- assertNotNull(recordsWritten);
- assertEquals(1, recordsWritten.size());
-
- byte[] written = recordsWritten.get(0);
-
- Map<String, ?> writtenMap = objectMapper.readValue(new String(written), new TypeReference<Map<String, ?>>() {
- });
-
- Map<String, Object> expected = new HashMap<>();
- expected.put(COL1, id);
- expected.put(COL2, msg);
-
- assertEquals(expected, writtenMap);
-
- bolt.cleanup();
- }
-
- @Test
- public void testNoAcksUntilFlushed() {
- JsonRecordHiveMapper mapper = new JsonRecordHiveMapper()
- .withColumnFields(new Fields(colNames1))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(2);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, new OutputCollector(collector));
-
- Tuple tuple1 = generateTestTuple(1, "SJC", "Sunnyvale", "CA");
- Tuple tuple2 = generateTestTuple(2, "SFO", "San Jose", "CA");
-
- bolt.execute(tuple1);
- verifyNoInteractions(collector);
-
- bolt.execute(tuple2);
- verify(collector).ack(tuple1);
- verify(collector).ack(tuple2);
- bolt.cleanup();
- }
-
- @Test
- public void testNoAcksIfFlushFails() throws Exception {
- JsonRecordHiveMapper mapper = new JsonRecordHiveMapper()
- .withColumnFields(new Fields(colNames1))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(2);
-
- HiveBolt failingBolt = new FlushFailureHiveBolt(hiveOptions);
-
- failingBolt.prepare(config, null, new OutputCollector(collector));
-
- Tuple tuple1 = generateTestTuple(1, "SJC", "Sunnyvale", "CA");
- Tuple tuple2 = generateTestTuple(2, "SFO", "San Jose", "CA");
-
- failingBolt.execute(tuple1);
- failingBolt.execute(tuple2);
-
- verify(collector, never()).ack(tuple1);
- verify(collector, never()).ack(tuple2);
-
- failingBolt.cleanup();
- }
-
- @Test
- public void testTickTuple() {
- JsonRecordHiveMapper mapper = new JsonRecordHiveMapper()
- .withColumnFields(new Fields(colNames1))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(2);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, new OutputCollector(collector));
-
- Tuple tuple1 = generateTestTuple(1, "SJC", "Sunnyvale", "CA");
- Tuple tuple2 = generateTestTuple(2, "SFO", "San Jose", "CA");
-
- bolt.execute(tuple1);
-
- //The tick should cause tuple1 to be ack'd
- Tuple mockTick = MockTupleHelpers.mockTickTuple();
- bolt.execute(mockTick);
- verify(collector).ack(tuple1);
-
- //The second tuple should NOT be ack'd because the batch should be cleared and this will be
- //the first transaction in the new batch
- bolt.execute(tuple2);
- verify(collector, never()).ack(tuple2);
-
- bolt.cleanup();
- }
-
- @Test
- public void testNoTickEmptyBatches() throws Exception {
- JsonRecordHiveMapper mapper = new JsonRecordHiveMapper()
- .withColumnFields(new Fields(colNames1))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(2)
- .withBatchSize(2);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, new OutputCollector(collector));
-
- //The tick should NOT cause any acks since the batch was empty except for acking itself
- Tuple mockTick = MockTupleHelpers.mockTickTuple();
- bolt.execute(mockTick);
- verifyNoInteractions(collector);
-
- bolt.cleanup();
- }
-
- @Test
- public void testMultiPartitionTuples()
- throws Exception {
- DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
- .withTxnsPerBatch(10)
- .withBatchSize(10);
-
- bolt = new TestingHiveBolt(hiveOptions);
- bolt.prepare(config, null, new OutputCollector(collector));
-
- Integer id = 1;
- String msg = "test";
- String city = "San Jose";
- String state = "CA";
-
- List<Tuple> tuples = new ArrayList<>();
- for (int i = 0; i < 100; i++) {
- Tuple tuple = generateTestTuple(id, msg, city, state);
- tuples.add(tuple);
- bolt.execute(tuple);
- }
-
- for (Tuple t : tuples) {
- verify(collector).ack(t);
- }
-
- List<String> partVals = Lists.newArrayList(city, state);
-
- List<byte[]> recordsWritten = bolt.getRecordWritten(partVals);
- assertNotNull(recordsWritten);
- assertEquals(100, recordsWritten.size());
-
-
- byte[] mapped = generateDelimiteredRecord(Lists.newArrayList(id, msg), mapper.getFieldDelimiter());
-
- for (byte[] record : recordsWritten) {
- assertArrayEquals(mapped, record);
- }
-
- bolt.cleanup();
- }
-
- private Tuple generateTestTuple(Object id, Object msg, Object city, Object state) {
- TopologyBuilder builder = new TopologyBuilder();
- GeneralTopologyContext topologyContext = new GeneralTopologyContext(builder.createTopology(),
- new Config(), new HashMap(), new HashMap(), new HashMap(), "") {
- @Override
- public Fields getComponentOutputFields(String componentId, String streamId) {
- return new Fields("id", "msg", "city", "state");
- }
- };
- return new TupleImpl(topologyContext, new Values(id, msg, city, state), "", 1, "");
- }
-
- private byte[] generateDelimiteredRecord(List<?> values, String fieldDelimiter) {
- StringBuilder builder = new StringBuilder();
- for (Object value : values) {
- builder.append(value);
- builder.append(fieldDelimiter);
- }
- return builder.toString().getBytes();
- }
-
- private static class TestingHiveBolt extends HiveBolt {
-
- protected Map<List<String>, List<byte[]>> partitionValuesToWrittenRecords = new HashMap<>();
-
- public TestingHiveBolt(HiveOptions options) {
- super(options);
- }
-
- @Override
- HiveWriter getOrCreateWriter(final HiveEndPoint endPoint)
- throws HiveWriter.ConnectFailure, InterruptedException {
- HiveWriter writer = allWriters.get(endPoint);
- if (writer == null) {
- // always provide mocked HiveWriter
- writer = Mockito.mock(HiveWriter.class);
- try {
- Mockito.doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- Object[] arguments = invocation.getArguments();
- List<String> partitionVals = endPoint.partitionVals;
- List<byte[]> writtenRecords = partitionValuesToWrittenRecords.get(partitionVals);
- if (writtenRecords == null) {
- writtenRecords = new ArrayList<>();
- partitionValuesToWrittenRecords.put(partitionVals, writtenRecords);
- }
- writtenRecords.add((byte[]) arguments[0]);
- return null;
- }
- }).when(writer).write(any(byte[].class));
- } catch (Exception exc) {
- throw new RuntimeException(exc);
- }
- }
- return writer;
- }
-
- public Map<List<String>, List<byte[]>> getPartitionValuesToWrittenRecords() {
- return partitionValuesToWrittenRecords;
- }
-
- public List<byte[]> getRecordWritten(List<String> partitionValues) {
- return partitionValuesToWrittenRecords.get(partitionValues);
- }
- }
-
- private static class FlushFailureHiveBolt extends TestingHiveBolt {
-
- public FlushFailureHiveBolt(HiveOptions options) {
- super(options);
- }
-
- @Override
- void flushAllWriters(boolean rollToNext) throws HiveWriter.CommitFailure, HiveWriter.TxnBatchFailure, HiveWriter.TxnFailure,
- InterruptedException {
- if (rollToNext) {
- throw new InterruptedException();
- } else {
- super.flushAllWriters(false);
- }
- }
- }
-
-}
diff --git a/external/storm-hive/src/test/java/org/apache/storm/hive/common/TestHiveWriter.java b/external/storm-hive/src/test/java/org/apache/storm/hive/common/TestHiveWriter.java
deleted file mode 100644
index f5be324b3..000000000
--- a/external/storm-hive/src/test/java/org/apache/storm/hive/common/TestHiveWriter.java
+++ /dev/null
@@ -1,207 +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.storm.hive.common;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hive.hcatalog.streaming.HiveEndPoint;
-import org.apache.hive.hcatalog.streaming.RecordWriter;
-import org.apache.hive.hcatalog.streaming.SerializationError;
-import org.apache.hive.hcatalog.streaming.StreamingConnection;
-import org.apache.hive.hcatalog.streaming.StreamingException;
-import org.apache.hive.hcatalog.streaming.TransactionBatch;
-import org.apache.storm.Config;
-import org.apache.storm.hive.bolt.HiveSetupUtil;
-import org.apache.storm.hive.bolt.mapper.DelimitedRecordHiveMapper;
-import org.apache.storm.hive.bolt.mapper.HiveMapper;
-import org.apache.storm.task.GeneralTopologyContext;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.TupleImpl;
-import org.apache.storm.tuple.Values;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@Deprecated(since = "2.7.1", forRemoval = true)
-public class TestHiveWriter {
- public static final String PART1_NAME = "city";
- public static final String PART2_NAME = "state";
- public static final String[] partNames = { PART1_NAME, PART2_NAME };
- final static String dbName = "testdb";
- final static String tblName = "test_table2";
- final String[] partitionVals = { "sunnyvale", "ca" };
- final String[] colNames = { "id", "msg" };
- private final int port;
- private final String metaStoreURI;
- private final HiveConf conf;
- int timeout = 10000; // msec
- UserGroupInformation ugi = null;
- private ExecutorService callTimeoutPool;
-
- public TestHiveWriter() throws Exception {
- port = 9083;
- metaStoreURI = null;
- int callTimeoutPoolSize = 1;
- callTimeoutPool = Executors.newFixedThreadPool(callTimeoutPoolSize,
- new ThreadFactoryBuilder().setNameFormat("hiveWriterTest").build());
-
- // 1) Start metastore
- conf = HiveSetupUtil.getHiveConf();
- TxnDbUtil.setConfValues(conf);
- if (metaStoreURI != null) {
- conf.setVar(HiveConf.ConfVars.METASTOREURIS, metaStoreURI);
- }
- }
-
- @Test
- public void testInstantiate() throws Exception {
- DelimitedRecordHiveMapper mapper = new MockedDelemiteredRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveEndPoint endPoint = new HiveEndPoint(metaStoreURI, dbName, tblName, Arrays.asList(partitionVals));
- TestingHiveWriter writer = new TestingHiveWriter(endPoint, 10, true, timeout
- , callTimeoutPool, mapper, ugi, false);
- writer.close();
- }
-
- @Test
- public void testWriteBasic() throws Exception {
- DelimitedRecordHiveMapper mapper = new MockedDelemiteredRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
- HiveEndPoint endPoint = new HiveEndPoint(metaStoreURI, dbName, tblName, Arrays.asList(partitionVals));
- TestingHiveWriter writer = new TestingHiveWriter(endPoint, 10, true, timeout
- , callTimeoutPool, mapper, ugi, false);
- writeTuples(writer, mapper, 3);
- writer.flush(false);
- writer.close();
- Mockito.verify(writer.getMockedTxBatch(), Mockito.times(3)).write(Mockito.any(byte[].class));
- }
-
- @Test
- public void testWriteMultiFlush() throws Exception {
- DelimitedRecordHiveMapper mapper = new MockedDelemiteredRecordHiveMapper()
- .withColumnFields(new Fields(colNames))
- .withPartitionFields(new Fields(partNames));
-
- HiveEndPoint endPoint = new HiveEndPoint(metaStoreURI, dbName, tblName, Arrays.asList(partitionVals));
- TestingHiveWriter writer = new TestingHiveWriter(endPoint, 10, true, timeout
- , callTimeoutPool, mapper, ugi, false);
- Tuple tuple = generateTestTuple("1", "abc");
- writer.write(mapper.mapRecord(tuple));
- tuple = generateTestTuple("2", "def");
- writer.write(mapper.mapRecord(tuple));
- assertEquals(writer.getTotalRecords(), 2);
- Mockito.verify(writer.getMockedTxBatch(), Mockito.times(2)).write(Mockito.any(byte[].class));
- Mockito.verify(writer.getMockedTxBatch(), Mockito.never()).commit();
- writer.flush(true);
- assertEquals(writer.getTotalRecords(), 0);
- Mockito.verify(writer.getMockedTxBatch(), Mockito.atLeastOnce()).commit();
-
- tuple = generateTestTuple("3", "ghi");
- writer.write(mapper.mapRecord(tuple));
- writer.flush(true);
-
- tuple = generateTestTuple("4", "klm");
- writer.write(mapper.mapRecord(tuple));
- writer.flush(true);
- writer.close();
- Mockito.verify(writer.getMockedTxBatch(), Mockito.times(4)).write(Mockito.any(byte[].class));
- }
-
- private Tuple generateTestTuple(Object id, Object msg) {
- TopologyBuilder builder = new TopologyBuilder();
- GeneralTopologyContext topologyContext = new GeneralTopologyContext(builder.createTopology(),
- new Config(), new HashMap(), new HashMap(), new HashMap(), "") {
- @Override
- public Fields getComponentOutputFields(String componentId, String streamId) {
- return new Fields("id", "msg");
- }
- };
- return new TupleImpl(topologyContext, new Values(id, msg), "", 1, "");
- }
-
- private void writeTuples(HiveWriter writer, HiveMapper mapper, int count)
- throws HiveWriter.WriteFailure, InterruptedException, SerializationError {
- Integer id = 100;
- String msg = "test-123";
- for (int i = 1; i <= count; i++) {
- Tuple tuple = generateTestTuple(id, msg);
- writer.write(mapper.mapRecord(tuple));
- }
- }
-
- private static class TestingHiveWriter extends HiveWriter {
-
- private StreamingConnection mockedStreamingConn;
- private TransactionBatch mockedTxBatch;
-
- public TestingHiveWriter(HiveEndPoint endPoint, int txnsPerBatch, boolean autoCreatePartitions, long callTimeout,
- ExecutorService callTimeoutPool, HiveMapper mapper, UserGroupInformation ugi,
- boolean tokenAuthEnabled) throws InterruptedException, ConnectFailure {
- super(endPoint, txnsPerBatch, autoCreatePartitions, callTimeout, callTimeoutPool, mapper, ugi, tokenAuthEnabled);
- }
-
- @Override
- synchronized StreamingConnection newConnection(UserGroupInformation ugi, boolean tokenAuthEnabled) throws InterruptedException,
- ConnectFailure {
- if (mockedStreamingConn == null) {
- mockedStreamingConn = Mockito.mock(StreamingConnection.class);
- mockedTxBatch = Mockito.mock(TransactionBatch.class);
-
- try {
- Mockito.when(mockedStreamingConn.fetchTransactionBatch(Mockito.anyInt(), Mockito.any(RecordWriter.class)))
- .thenReturn(mockedTxBatch);
- } catch (StreamingException e) {
- throw new RuntimeException(e);
- }
- }
-
- return mockedStreamingConn;
- }
-
- public TransactionBatch getMockedTxBatch() {
- return mockedTxBatch;
- }
- }
-
- private static class MockedDelemiteredRecordHiveMapper extends DelimitedRecordHiveMapper {
- private final RecordWriter mockedRecordWriter;
-
- public MockedDelemiteredRecordHiveMapper() {
- this.mockedRecordWriter = Mockito.mock(RecordWriter.class);
- }
-
- @Override
- public RecordWriter createRecordWriter(HiveEndPoint endPoint) throws StreamingException, IOException, ClassNotFoundException {
- return mockedRecordWriter;
- }
-
- public RecordWriter getMockedRecordWriter() {
- return mockedRecordWriter;
- }
- }
-
-}
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 433009c0b..4d12f436a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -114,7 +114,6 @@
<zookeeper.version>3.9.3</zookeeper.version>
<snappy.version>1.1.10.7</snappy.version>
<jline.version>2.14.6</jline.version>
- <hive.version>3.1.3</hive.version>
<hikari.version>6.2.0</hikari.version>
<hadoop.version>3.4.1</hadoop.version>
<hdfs.version>${hadoop.version}</hdfs.version>
@@ -493,7 +492,6 @@
<module>external/storm-hdfs</module>
<module>external/storm-hdfs-blobstore</module>
<module>external/storm-hdfs-oci</module>
- <module>external/storm-hive</module>
<module>external/storm-jdbc</module>
<module>external/storm-redis</module>
<module>external/storm-metrics</module>
@@ -522,7 +520,6 @@
<module>examples/storm-kafka-client-examples</module>
<module>examples/storm-jdbc-examples</module>
<module>examples/storm-hdfs-examples</module>
- <module>examples/storm-hive-examples</module>
<module>examples/storm-jms-examples</module>
<module>examples/storm-perf</module>
</modules>
diff --git a/storm-dist/binary/final-package/src/main/assembly/binary.xml b/storm-dist/binary/final-package/src/main/assembly/binary.xml
index 602dde805..d396a1f6c 100644
--- a/storm-dist/binary/final-package/src/main/assembly/binary.xml
+++ b/storm-dist/binary/final-package/src/main/assembly/binary.xml
@@ -131,13 +131,6 @@
<include>README.*</include>
</includes>
</fileSet>
- <fileSet>
- <directory>${project.basedir}/../../../external/storm-hive</directory>
- <outputDirectory>external/storm-hive</outputDirectory>
- <includes>
- <include>README.*</include>
- </includes>
- </fileSet>
<fileSet>
<directory>${project.basedir}/../../../external/storm-eventhubs</directory>
<outputDirectory>external/storm-eventhubs</outputDirectory>