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 2023/10/19 07:25:34 UTC
[storm] 03/13: STORM-3988 - Remove "storm-hbase"
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch STORM-3988
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 7adbe83b37937e16824a188963c7667e133e858b
Author: Richard Zowalla <ri...@hs-heilbronn.de>
AuthorDate: Thu Oct 19 08:57:46 2023 +0200
STORM-3988 - Remove "storm-hbase"
---
examples/storm-hbase-examples/README.md | 62 ---
examples/storm-hbase-examples/pom.xml | 96 -----
.../storm/hbase/topology/LookupWordCount.java | 68 ----
.../storm/hbase/topology/PersistentWordCount.java | 79 ----
.../storm/hbase/topology/TotalWordCounter.java | 68 ----
.../storm/hbase/topology/WordCountClient.java | 54 ---
.../storm/hbase/topology/WordCountValueMapper.java | 62 ---
.../apache/storm/hbase/topology/WordCounter.java | 56 ---
.../org/apache/storm/hbase/topology/WordSpout.java | 87 -----
.../apache/storm/hbase/trident/PrintFunction.java | 34 --
.../storm/hbase/trident/WordCountTrident.java | 94 -----
external/storm-hbase/README.md | 315 ---------------
external/storm-hbase/pom.xml | 127 ------
.../apache/storm/hbase/bolt/AbstractHBaseBolt.java | 82 ----
.../org/apache/storm/hbase/bolt/HBaseBolt.java | 114 ------
.../apache/storm/hbase/bolt/HBaseLookupBolt.java | 117 ------
.../storm/hbase/bolt/mapper/HBaseMapper.java | 35 --
.../hbase/bolt/mapper/HBaseProjectionCriteria.java | 87 -----
.../storm/hbase/bolt/mapper/HBaseValueMapper.java | 36 --
.../storm/hbase/bolt/mapper/SimpleHBaseMapper.java | 85 ----
.../org/apache/storm/hbase/common/ColumnList.java | 198 ----------
.../org/apache/storm/hbase/common/HBaseClient.java | 181 ---------
.../org/apache/storm/hbase/common/IColumn.java | 26 --
.../org/apache/storm/hbase/common/ICounter.java | 24 --
.../java/org/apache/storm/hbase/common/Utils.java | 123 ------
.../storm/hbase/state/HBaseKeyValueState.java | 434 ---------------------
.../hbase/state/HBaseKeyValueStateIterator.java | 146 -------
.../hbase/state/HBaseKeyValueStateProvider.java | 165 --------
.../mapper/SimpleTridentHBaseMapMapper.java | 50 ---
.../trident/mapper/SimpleTridentHBaseMapper.java | 79 ----
.../trident/mapper/TridentHBaseMapMapper.java | 34 --
.../hbase/trident/mapper/TridentHBaseMapper.java | 40 --
.../storm/hbase/trident/state/HBaseMapState.java | 264 -------------
.../storm/hbase/trident/state/HBaseQuery.java | 35 --
.../storm/hbase/trident/state/HBaseState.java | 171 --------
.../hbase/trident/state/HBaseStateFactory.java | 34 --
.../storm/hbase/trident/state/HBaseUpdater.java | 27 --
.../hbase/trident/windowing/HBaseWindowsStore.java | 267 -------------
.../windowing/HBaseWindowsStoreFactory.java | 48 ---
.../storm/hbase/state/HBaseClientTestUtil.java | 387 ------------------
.../state/HBaseKeyValueStateIteratorTest.java | 203 ----------
.../state/HBaseKeyValueStateProviderTest.java | 86 ----
.../storm/hbase/state/HBaseKeyValueStateTest.java | 115 ------
pom.xml | 2 -
44 files changed, 4897 deletions(-)
diff --git a/examples/storm-hbase-examples/README.md b/examples/storm-hbase-examples/README.md
deleted file mode 100644
index 83c6f3da6..000000000
--- a/examples/storm-hbase-examples/README.md
+++ /dev/null
@@ -1,62 +0,0 @@
-# Storm HBase Integration Example
-
-This is a very simple set of topologies that show how to use the storm-hbase package for accessing HBase from storm.
-
-## HBase Setup
-
-First you need an instance of HBase that is setup and running. If you have one already you can start setting up the table using hbase shell, if not download a copy from http://archive.apache.org/dist/hbase/1.4.4/ and untar the result into a directory you want to run it in. Then follow the instructions from https://hbase.apache.org/0.94/book/quickstart.html to setup a standalone HBase instance. Be aware that when you run `start-hbase.sh` an instance of zookeeper will also be started. I [...]
-
-Before we can run the topology we need to setup a table in HBase to store the results.
-
-First launch the hbase shell
-
-```
-hbase shell
-```
-
-Next create a table called WordCount with a single column family called cf.
-
-```
-create 'WordCount', 'cf'
-```
-
-## PersistentWordCount
-
-PersistentWordCount is a very simple topology that performs a word count and stores the results in HBase.
-
-### Run The Topology
-
-```
-storm jar storm-hbase-examples-${STORM_VERSION}.jar org.apache.storm.hbase.topology.PersistentWordCount ${HBASE_HOME}
-```
-
-In this `${STORM_VERSION}` should be the version of storm you are running, and `${HBASE_HOME}` is where your installed version of HBase is. `${HBASE_HOME}` is mostly to get the config started. Please refer to the documentation for storm-hbase for more information on how to configure your topology.
-
-### Verify The Results
-
-If you want to see the results of the topology live you can run the command
-
-```
-storm jar storm-hbase-examples-${STORM_VERSION}.jar org.apache.storm.hbase.topology.WordCountClient ${HBASE_HOME}
-```
-
-## WordCountTrident
-
-WordCountTrident does essentially the same as PersistentWordCount but using Trident instead.
-
-### Run The Trident Topology
-
-```
-storm jar storm-hbase-examples-${STORM_VERSION}.jar org.apache.storm.hbase.trident.WordCountTrident ${HBASE_HOME}
-```
-
-### Verify The Trident Results
-
-Verifying the results for the trident topology is a little more difficult. The data is stored in the same format as PersistentWordCount, but the keys are different. To verify that it is working you can look at the logs and you will occasionally see log messages like
-
-```
-o.a.s.h.t.PrintFunction Thread-16-b-1-executor[7, 7] [INFO] [storm, 10]
-```
-
-Or you can use the `hbase shell` to `scan 'WordCount'` and see that the values are being updated consistently.
-
diff --git a/examples/storm-hbase-examples/pom.xml b/examples/storm-hbase-examples/pom.xml
deleted file mode 100644
index 8e32febbe..000000000
--- a/examples/storm-hbase-examples/pom.xml
+++ /dev/null
@@ -1,96 +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.6.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <artifactId>storm-hbase-examples</artifactId>
-
- <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-hbase</artifactId>
- <version>${project.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-hbase-examples/src/main/java/org/apache/storm/hbase/topology/LookupWordCount.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/LookupWordCount.java
deleted file mode 100644
index 7f3afb1c8..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/LookupWordCount.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.hbase.topology;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.hbase.bolt.HBaseLookupBolt;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.SimpleHBaseMapper;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-
-
-public class LookupWordCount {
- private static final String WORD_SPOUT = "WORD_SPOUT";
- private static final String LOOKUP_BOLT = "LOOKUP_BOLT";
- private static final String TOTAL_COUNT_BOLT = "TOTAL_COUNT_BOLT";
-
- public static void main(String[] args) throws Exception {
- Config config = new Config();
-
- Map<String, Object> hbConf = new HashMap<String, Object>();
- if (args.length > 0) {
- hbConf.put("hbase.rootdir", args[0]);
- }
- config.put("hbase.conf", hbConf);
-
- WordSpout spout = new WordSpout();
-
- SimpleHBaseMapper mapper = new SimpleHBaseMapper().withRowKeyField("word");
- HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria();
- projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"));
-
- WordCountValueMapper rowToTupleMapper = new WordCountValueMapper();
-
- HBaseLookupBolt lookupBolt = new HBaseLookupBolt("WordCount", mapper, rowToTupleMapper)
- .withConfigKey("hbase.conf")
- .withProjectionCriteria(projectionCriteria);
-
- //wordspout -> lookupbolt -> totalCountBolt
- TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout(WORD_SPOUT, spout, 1);
- builder.setBolt(LOOKUP_BOLT, lookupBolt, 1).shuffleGrouping(WORD_SPOUT);
- TotalWordCounter totalBolt = new TotalWordCounter();
- builder.setBolt(TOTAL_COUNT_BOLT, totalBolt, 1).fieldsGrouping(LOOKUP_BOLT, new Fields("columnName"));
- String topoName = "test";
- if (args.length == 1) {
- topoName = args[1];
- } else if (args.length > 1) {
- System.out.println("Usage: LookupWordCount <hbase.rootdir>");
- return;
- }
-
- StormSubmitter.submitTopology(topoName, config, builder.createTopology());
- }
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/PersistentWordCount.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/PersistentWordCount.java
deleted file mode 100644
index 0c929a3c1..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/PersistentWordCount.java
+++ /dev/null
@@ -1,79 +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.hbase.topology;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.hbase.bolt.HBaseBolt;
-import org.apache.storm.hbase.bolt.mapper.SimpleHBaseMapper;
-import org.apache.storm.hbase.security.HBaseSecurityUtil;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-
-
-public class PersistentWordCount {
- private static final String WORD_SPOUT = "WORD_SPOUT";
- private static final String COUNT_BOLT = "COUNT_BOLT";
- private static final String HBASE_BOLT = "HBASE_BOLT";
-
-
- public static void main(String[] args) throws Exception {
- Config config = new Config();
-
- Map<String, Object> hbConf = new HashMap<>();
- if (args.length > 0) {
- hbConf.put("hbase.rootdir", args[0]);
- }
- config.put("hbase.conf", hbConf);
-
- WordSpout spout = new WordSpout();
- WordCounter bolt = new WordCounter();
-
- SimpleHBaseMapper mapper = new SimpleHBaseMapper()
- .withRowKeyField("word")
- .withColumnFields(new Fields("word"))
- .withCounterFields(new Fields("count"))
- .withColumnFamily("cf");
-
- HBaseBolt hbase = new HBaseBolt("WordCount", mapper)
- .withConfigKey("hbase.conf");
-
-
- // wordSpout ==> countBolt ==> HBaseBolt
- TopologyBuilder builder = new TopologyBuilder();
-
- builder.setSpout(WORD_SPOUT, spout, 1);
- builder.setBolt(COUNT_BOLT, bolt, 1).shuffleGrouping(WORD_SPOUT);
- builder.setBolt(HBASE_BOLT, hbase, 1).fieldsGrouping(COUNT_BOLT, new Fields("word"));
-
- String topoName = "test";
- if (args.length == 2) {
- topoName = args[0];
- } else if (args.length == 4) {
- System.out.println("hdfs url: " + args[0]
- + ", keytab file: " + args[2]
- + ", principal name: " + args[3]
- + ", toplogy name: " + args[1]);
- hbConf.put(HBaseSecurityUtil.STORM_KEYTAB_FILE_KEY, args[2]);
- hbConf.put(HBaseSecurityUtil.STORM_USER_NAME_KEY, args[3]);
- config.setNumWorkers(3);
- topoName = args[1];
- } else if (args.length != 1) {
- System.out.println("Usage: PersistentWordCount <hbase.rootdir> [topology name] [keytab file] [principal name]");
- return;
- }
- StormSubmitter.submitTopology(topoName, config, builder.createTopology());
- }
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/TotalWordCounter.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/TotalWordCounter.java
deleted file mode 100644
index 41993191a..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/TotalWordCounter.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.hbase.topology;
-
-import static org.apache.storm.utils.Utils.tuple;
-
-import java.math.BigInteger;
-import java.util.Map;
-import java.util.Random;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.BasicOutputCollector;
-import org.apache.storm.topology.IBasicBolt;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TotalWordCounter implements IBasicBolt {
-
- private static final Logger LOG = LoggerFactory.getLogger(TotalWordCounter.class);
- private static final Random RANDOM = new Random();
- private BigInteger total = BigInteger.ZERO;
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context) {
- }
-
- /*
- * Just output the word value with a count of 1.
- * The HBaseBolt will handle incrementing the counter.
- */
- @Override
- public void execute(Tuple input, BasicOutputCollector collector) {
- total = total.add(new BigInteger(input.getValues().get(1).toString()));
- collector.emit(tuple(total.toString()));
- //prints the total with low probability.
- if (RANDOM.nextInt(1000) > 995) {
- LOG.info("Running total = " + total);
- }
- }
-
- @Override
- public void cleanup() {
- LOG.info("Final total = " + total);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("total"));
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountClient.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountClient.java
deleted file mode 100644
index d8f6eb2c1..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountClient.java
+++ /dev/null
@@ -1,54 +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.hbase.topology;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.util.Bytes;
-
-/**
- * Connects to the 'WordCount' table and prints counts for each word.
- *
- * <p>Assumes you have run (or are running) <code>PersistentWordCount</code>
- */
-public class WordCountClient {
-
- public static void main(String[] args) throws Exception {
- Configuration config = HBaseConfiguration.create();
- if (args.length > 0) {
- config.set("hbase.rootdir", args[0]);
- }
-
- Table table = ConnectionFactory.createConnection(config).getTable(TableName.valueOf("WordCount"));
-
-
- for (String word : WordSpout.words) {
- Get get = new Get(Bytes.toBytes(word));
- Result result = table.get(get);
-
- byte[] countBytes = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("count"));
- byte[] wordBytes = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("word"));
-
- String wordStr = Bytes.toString(wordBytes);
- System.out.println(wordStr);
- long count = Bytes.toLong(countBytes);
- System.out.println("Word: '" + wordStr + "', Count: " + count);
- }
-
- }
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountValueMapper.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountValueMapper.java
deleted file mode 100644
index 9cb8115c8..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCountValueMapper.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.hbase.topology;
-
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.ITuple;
-import org.apache.storm.tuple.Values;
-
-/**
- * Takes a Hbase result and returns a value list that has a value instance for each column and corresponding value. So
- * if the result from Hbase was
- * <pre>
- * WORD, COUNT
- * apple, 10
- * bannana, 20
- * </pre>
- *
- * <p>this will return
- * <pre>
- * [WORD, apple]
- * [COUNT, 10]
- * [WORD, banana]
- * [COUNT, 20]
- * </pre>
- */
-public class WordCountValueMapper implements HBaseValueMapper {
-
- @Override
- public List<Values> toValues(ITuple tuple, Result result) throws Exception {
- List<Values> values = new ArrayList<Values>();
- Cell[] cells = result.rawCells();
- for (Cell cell : cells) {
- Values value = new Values(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toLong(CellUtil.cloneValue(cell)));
- values.add(value);
- }
- return values;
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("columnName", "columnValue"));
- }
-
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCounter.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCounter.java
deleted file mode 100644
index b2cadb8a2..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordCounter.java
+++ /dev/null
@@ -1,56 +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.hbase.topology;
-
-import static org.apache.storm.utils.Utils.tuple;
-
-import java.util.Map;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.BasicOutputCollector;
-import org.apache.storm.topology.IBasicBolt;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-
-public class WordCounter implements IBasicBolt {
-
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context) {
- }
-
- /*
- * Just output the word value with a count of 1.
- * The HBaseBolt will handle incrementing the counter.
- */
- @Override
- public void execute(Tuple input, BasicOutputCollector collector) {
- collector.emit(tuple(input.getValues().get(0), 1));
- }
-
- @Override
- public void cleanup() {
-
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("word", "count"));
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordSpout.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordSpout.java
deleted file mode 100644
index a80e80af0..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/topology/WordSpout.java
+++ /dev/null
@@ -1,87 +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.hbase.topology;
-
-import java.util.Map;
-import java.util.Random;
-import java.util.UUID;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.IRichSpout;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-
-public class WordSpout implements IRichSpout {
- public static final String[] words = new String[]{ "apple", "orange", "pineapple", "banana", "watermelon" };
- boolean isDistributed;
- SpoutOutputCollector collector;
-
- public WordSpout() {
- this(true);
- }
-
- public WordSpout(boolean isDistributed) {
- this.isDistributed = isDistributed;
- }
-
- public boolean isDistributed() {
- return this.isDistributed;
- }
-
- @Override
- public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
- this.collector = collector;
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public void nextTuple() {
- final Random rand = new Random();
- final String word = words[rand.nextInt(words.length)];
- this.collector.emit(new Values(word), UUID.randomUUID());
- Thread.yield();
- }
-
- @Override
- public void ack(Object msgId) {
-
- }
-
- @Override
- public void fail(Object msgId) {
-
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("word"));
- }
-
- @Override
- public void activate() {
- }
-
- @Override
- public void deactivate() {
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/PrintFunction.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/PrintFunction.java
deleted file mode 100644
index 7285e547a..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/PrintFunction.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.hbase.trident;
-
-import java.util.Random;
-import org.apache.storm.trident.operation.BaseFunction;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PrintFunction extends BaseFunction {
-
- private static final Logger LOG = LoggerFactory.getLogger(PrintFunction.class);
-
- private static final Random RANDOM = new Random();
-
- @Override
- public void execute(TridentTuple tuple, TridentCollector tridentCollector) {
- if (RANDOM.nextInt(1000) > 995) {
- LOG.info(tuple.toString());
- }
- }
-}
diff --git a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/WordCountTrident.java b/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/WordCountTrident.java
deleted file mode 100644
index ee7cee0f5..000000000
--- a/examples/storm-hbase-examples/src/main/java/org/apache/storm/hbase/trident/WordCountTrident.java
+++ /dev/null
@@ -1,94 +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.hbase.trident;
-
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
-import org.apache.storm.hbase.topology.WordCountValueMapper;
-import org.apache.storm.hbase.trident.mapper.SimpleTridentHBaseMapper;
-import org.apache.storm.hbase.trident.mapper.TridentHBaseMapper;
-import org.apache.storm.hbase.trident.state.HBaseQuery;
-import org.apache.storm.hbase.trident.state.HBaseState;
-import org.apache.storm.hbase.trident.state.HBaseStateFactory;
-import org.apache.storm.hbase.trident.state.HBaseUpdater;
-import org.apache.storm.trident.Stream;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.state.StateFactory;
-import org.apache.storm.trident.testing.FixedBatchSpout;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-
-public class WordCountTrident {
- public static StormTopology buildTopology(String hbaseRoot) {
- Fields fields = new Fields("word", "count");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", 1),
- new Values("trident", 1),
- new Values("needs", 1),
- new Values("javadoc", 1)
- );
- spout.setCycle(true);
-
- TridentHBaseMapper tridentHBaseMapper = new SimpleTridentHBaseMapper()
- .withColumnFamily("cf")
- .withColumnFields(new Fields("word"))
- .withCounterFields(new Fields("count"))
- .withRowKeyField("word");
-
- HBaseValueMapper rowToStormValueMapper = new WordCountValueMapper();
-
- HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria();
- projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"));
-
- HBaseState.Options options = new HBaseState.Options()
- .withConfigKey(hbaseRoot)
- .withDurability(Durability.SYNC_WAL)
- .withMapper(tridentHBaseMapper)
- .withProjectionCriteria(projectionCriteria)
- .withRowToStormValueMapper(rowToStormValueMapper)
- .withTableName("WordCount");
-
- StateFactory factory = new HBaseStateFactory(options);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- stream.partitionPersist(factory, fields, new HBaseUpdater(), new Fields());
-
- TridentState state = topology.newStaticState(factory);
- stream = stream.stateQuery(state, new Fields("word"), new HBaseQuery(), new Fields("columnName", "columnValue"));
- stream.each(new Fields("word", "columnValue"), new PrintFunction(), new Fields());
- return topology.build();
- }
-
- public static void main(String[] args) throws Exception {
- Config conf = new Config();
- conf.setMaxSpoutPending(5);
- String topoName = "wordCounter";
-
- if (args.length == 2) {
- topoName = args[1];
- } else if (args.length > 2) {
- System.out.println("Usage: WordCountTrident <hbase.rootdir> [topology name]");
- return;
- }
- conf.setNumWorkers(3);
- StormSubmitter.submitTopology(topoName, conf, buildTopology(args[0]));
- }
-
-}
diff --git a/external/storm-hbase/README.md b/external/storm-hbase/README.md
deleted file mode 100644
index 0326bcdc4..000000000
--- a/external/storm-hbase/README.md
+++ /dev/null
@@ -1,315 +0,0 @@
-# Storm HBase
-
-Storm/Trident integration for [Apache HBase](https://hbase.apache.org)
-
-## Usage
-The main API for interacting with HBase is the `org.apache.storm.hbase.bolt.mapper.HBaseMapper`
-interface:
-
-```java
-public interface HBaseMapper extends Serializable {
- byte[] rowKey(Tuple tuple);
-
- ColumnList columns(Tuple tuple);
-}
-```
-
-The `rowKey()` method is straightforward: given a Storm tuple, return a byte array representing the
-row key.
-
-The `columns()` method defines what will be written to an HBase row. The `ColumnList` class allows you
-to add both standard HBase columns as well as HBase counter columns.
-
-To add a standard column, use one of the `addColumn()` methods:
-
-```java
-ColumnList cols = new ColumnList();
-cols.addColumn(this.columnFamily, field.getBytes(), toBytes(tuple.getValueByField(field)));
-```
-
-To add a counter column, use one of the `addCounter()` methods:
-
-```java
-ColumnList cols = new ColumnList();
-cols.addCounter(this.columnFamily, field.getBytes(), toLong(tuple.getValueByField(field)));
-```
-
-When the remote HBase is security enabled, a kerberos keytab and the corresponding principal name need to be
-provided for the storm-hbase connector. Specifically, the Config object passed into the topology should contain
-{(“storm.keytab.file”, “$keytab”), ("storm.kerberos.principal", “$principal”)}. Example:
-
-```java
-Config config = new Config();
-...
-config.put("storm.keytab.file", "$keytab");
-config.put("storm.kerberos.principal", "$principal");
-StormSubmitter.submitTopology("$topologyName", config, builder.createTopology());
-```
-
-## Working with Secure HBASE using delegation tokens.
-If your topology is going to interact with secure HBase, your bolts/states needs to be authenticated by HBase.
-The approach described above requires that all potential worker hosts have "storm.keytab.file" on them. If you have
-multiple topologies on a cluster , each with different hbase user, you will have to create multiple keytabs and distribute
-it to all workers. Instead of doing that you could use the following approach:
-
-Your administrator can configure nimbus to automatically get delegation tokens on behalf of the topology submitter user. The nimbus should be started with following configurations:
-
-```
-nimbus.autocredential.plugins.classes : ["org.apache.storm.hbase.security.AutoHBase"]
-nimbus.credential.renewers.classes : ["org.apache.storm.hbase.security.AutoHBase"]
-hbase.keytab.file: "/path/to/keytab/on/nimbus" (This is the keytab of hbase super user that can impersonate other users.)
-hbase.kerberos.principal: "superuser@EXAMPLE.com"
-nimbus.credential.renewers.freq.secs : 518400 (6 days, hbase tokens by default expire every 7 days and can not be renewed, if you have custom settings for hbase.auth.token.max.lifetime in hbase-site.xml than you should ensure this value is atleast 1 hour less then that.)
-```
-
-Your topology configuration should have:
-
-```
-topology.auto-credentials :["org.apache.storm.hbase.security.AutoHBase"]
-```
-
-If nimbus did not have the above configuration you need to add it and then restart it. Ensure the hbase configuration
-files(core-site.xml, hdfs-site.xml and hbase-site.xml) and the storm-hbase 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 hbase-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),
-
-```
-hbaseCredentialsConfigKeys : ["cluster1", "cluster2"] (the hbase clusters you want to fetch the tokens from)
-"cluster1": {"config1": "value1", "config2": "value2", ... } (A map of config key-values specific to cluster1)
-"cluster2": {"config1": "value1", "hbase.keytab.file": "/path/to/keytab/for/cluster2/on/nimubs", "hbase.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/hbase-site1.xml"]}
-"cluster2": {"resources": ["/path/to/core-site2.xml", "/path/to/hbase-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 HBase cluster within the same topology.
-
-Nimbus will use the keytab and principal specified in the config to authenticate with HBase. 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 AutoHBase, nimbus will push the
-delegation tokens to all the workers for your topology and the hbase bolt/state will authenticate with these tokens.
-
-As nimbus is impersonating topology submitter user, you need to ensure the user specified in hbase.kerberos.principal
-has permissions to acquire tokens on behalf of other users. To achieve this you need to follow configuration directions
-listed on this link
-
-http://hbase.apache.org/book/security.html#security.rest.gateway
-
-You can read about setting up secure HBase here:http://hbase.apache.org/book/security.html.
-
-### SimpleHBaseMapper
-`storm-hbase` includes a general purpose `HBaseMapper` implementation called `SimpleHBaseMapper` that can map Storm
-tuples to both regular HBase columns as well as counter columns.
-
-To use `SimpleHBaseMapper`, you simply tell it which fields to map to which types of columns.
-
-The following code create a `SimpleHBaseMapper` instance that:
-
-1. Uses the `word` tuple value as a row key.
-2. Adds a standard HBase column for the tuple field `word`.
-3. Adds an HBase counter column for the tuple field `count`.
-4. Writes values to the `cf` column family.
-
-```java
-SimpleHBaseMapper mapper = new SimpleHBaseMapper()
- .withRowKeyField("word")
- .withColumnFields(new Fields("word"))
- .withCounterFields(new Fields("count"))
- .withColumnFamily("cf");
-```
-### HBaseBolt
-To use the `HBaseBolt`, construct it with the name of the table to write to, an a `HBaseMapper` implementation:
-
- ```java
-HBaseBolt hbase = new HBaseBolt("WordCount", mapper);
- ```
-
- HBaseBolt params
-
-| Arg | Description | Type | Default |
-| ---------------- |---------------------------------------------------- |------------------- | ------- |
-| writeToWAL | To turn Durability SYNC_WAL or SKIP_WAL | Boolean (Optional) | True |
-| configKey | Any Hbase related configs | Map (Optional) | |
-| batchSize | Max no.of Tuples batched together to write to HBase | Int (Optional) | 15000 |
-| flushIntervalSecs| (In seconds) If > 0 HBase Bolt will periodically flush transaction batches. Enabling this is recommended to avoid tuple timeouts while waiting for a batch to fill up. | Int (Optional) | 0 |
-
-
-The `HBaseBolt` will delegate to the `mapper` instance to figure out how to persist tuple data to HBase.
-
-### HBaseValueMapper
-This class allows you to transform the HBase lookup result into storm Values that will be emitted by the `HBaseLookupBolt`.
-
-```java
-public interface HBaseValueMapper extends Serializable {
- public List<Values> toTuples(Result result) throws Exception;
- void declareOutputFields(OutputFieldsDeclarer declarer);
-}
-```
-
-The `toTuples` method takes in a HBase `Result` instance and expects a List of `Values` instant.
-Each of the value returned by this function will be emitted by the `HBaseLookupBolt`.
-
-The `declareOutputFields` should be used to declare the outputFields of the `HBaseLookupBolt`.
-
-There is an example implementation in `examples/storm-hbase-examples/src/main/java` directory.
-
-### HBaseProjectionCriteria
-This class allows you to specify the projection criteria for your HBase Get function. This is optional parameter
-for the lookupBolt and if you do not specify this instance all the columns will be returned by `HBaseLookupBolt`.
-
-```java
-public class HBaseProjectionCriteria implements Serializable {
- public HBaseProjectionCriteria addColumnFamily(String columnFamily);
- public HBaseProjectionCriteria addColumn(ColumnMetaData column);
-```
-`addColumnFamily` takes in columnFamily. Setting this parameter means all columns for this family will be included
- in the projection.
-
-`addColumn` takes in a columnMetaData instance. Setting this parameter means only this column from the column familty
- will be part of your projection.
-The following code creates a projectionCriteria which specifies a projection criteria that:
-
-1. includes count column from column family cf.
-2. includes all columns from column family cf2.
-
-```java
-HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria()
- .addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"))
- .addColumnFamily("cf2");
-```
-
-### HBaseLookupBolt
-To use the `HBaseLookupBolt`, Construct it with the name of the table to write to, an implementation of `HBaseMapper`
-and an implementation of `HBaseRowToStormValueMapper`. You can optionally specify a `HBaseProjectionCriteria`.
-
-The `HBaseLookupBolt` will use the mapper to get rowKey to lookup for. It will use the `HBaseProjectionCriteria` to
-figure out which columns to include in the result and it will leverage the `HBaseRowToStormValueMapper` to get the
-values to be emitted by the bolt.
-
-In addition, the `HBaseLookupBolt` supports bolt-side HBase result caching using an in-memory LRU cache using Caffeine. To enable caching:
-
-`hbase.cache.enable` - to enable caching (default false)
-
-`hbase.cache.ttl.seconds` - set time to live for LRU cache in seconds (default 300)
-
-`hbase.cache.size` - set size of the cache (default 1000)
-
-You can look at an example topology LookupWordCount.java under `examples/storm-hbase-examples/src/main/java`.
-## Example: Persistent Word Count
-A runnable example can be found in the `examples/storm-hbase-examples/src/main/java` directory.
-
-### Setup
-The following steps assume you are running HBase locally, or there is an `hbase-site.xml` on the
-classpath pointing to your HBase cluster.
-
-Use the `hbase shell` command to create the schema:
-
-```
-> create 'WordCount', 'cf'
-```
-
-### Execution
-Run the `org.apache.storm.hbase.topology.PersistenWordCount` class (it will run the topology for 10 seconds, then exit).
-
-After (or while) the word count topology is running, run the `org.apache.storm.hbase.topology.WordCountClient` class
-to view the counter values stored in HBase. You should see something like to following:
-
-```
-Word: 'apple', Count: 6867
-Word: 'orange', Count: 6645
-Word: 'pineapple', Count: 6954
-Word: 'banana', Count: 6787
-Word: 'watermelon', Count: 6806
-```
-
-For reference, the sample topology is listed below:
-
-```java
-public class PersistentWordCount {
- private static final String WORD_SPOUT = "WORD_SPOUT";
- private static final String COUNT_BOLT = "COUNT_BOLT";
- private static final String HBASE_BOLT = "HBASE_BOLT";
-
-
- public static void main(String[] args) throws Exception {
- Config config = new Config();
-
- Map<String, Object> hbConf = new HashMap<String, Object>();
- if(args.length > 0){
- hbConf.put("hbase.rootdir", args[0]);
- }
- config.put("hbase.conf", hbConf);
-
- WordSpout spout = new WordSpout();
- WordCounter bolt = new WordCounter();
-
- SimpleHBaseMapper mapper = new SimpleHBaseMapper()
- .withRowKeyField("word")
- .withColumnFields(new Fields("word"))
- .withCounterFields(new Fields("count"))
- .withColumnFamily("cf");
-
- HBaseBolt hbase = new HBaseBolt("WordCount", mapper)
- .withConfigKey("hbase.conf");
-
-
- // wordSpout ==> countBolt ==> HBaseBolt
- TopologyBuilder builder = new TopologyBuilder();
-
- builder.setSpout(WORD_SPOUT, spout, 1);
- builder.setBolt(COUNT_BOLT, bolt, 1).shuffleGrouping(WORD_SPOUT);
- builder.setBolt(HBASE_BOLT, hbase, 1).fieldsGrouping(COUNT_BOLT, new Fields("word"));
-
-
- if (args.length == 1) {
- try (LocalCluster cluster = new LocalCluster();
- LocalTopology topo = cluster.submitTopology("test", config, builder.createTopology());) {
- Thread.sleep(30000);
- }
- System.exit(0);
- } else if (args.length == 2) {
- StormSubmitter.submitTopology(args[1], config, builder.createTopology());
- } else if (args.length == 4) {
- System.out.println("hdfs url: " + args[0] + ", keytab file: " + args[2] +
- ", principal name: " + args[3] + ", toplogy name: " + args[1]);
- hbConf.put(HBaseSecurityUtil.STORM_KEYTAB_FILE_KEY, args[2]);
- hbConf.put(HBaseSecurityUtil.STORM_USER_NAME_KEY, args[3]);
- config.setNumWorkers(3);
- StormSubmitter.submitTopology(args[1], config, builder.createTopology());
- } else {
- System.out.println("Usage: PersistentWordCount <hbase.rootdir> [topology name] [keytab file] [principal name]");
- }
- }
-}
-```
-
-
-## License
-
-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.
-
-## Committer Sponsors
-
- * P. Taylor Goetz ([ptgoetz@apache.org](mailto:ptgoetz@apache.org))
- * Sriharsha Chintalapani ([sriharsha@apache.org](mailto:sriharsha@apache.org))
-
diff --git a/external/storm-hbase/pom.xml b/external/storm-hbase/pom.xml
deleted file mode 100644
index 238f78542..000000000
--- a/external/storm-hbase/pom.xml
+++ /dev/null
@@ -1,127 +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.6.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <artifactId>storm-hbase</artifactId>
-
- <developers>
- <developer>
- <id>ptgoetz</id>
- <name>P. Taylor Goetz</name>
- <email>ptgoetz@gmail.com</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.hbase</groupId>
- <artifactId>hbase-annotations</artifactId>
- <version>${hbase.version}</version>
- <exclusions>
- <exclusion>
- <groupId>jdk.tools</groupId>
- <artifactId>jdk.tools</artifactId>
- </exclusion>
- <!-- https://jira.apache.org/jira/browse/HBASE-12909 -->
- <exclusion>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-client</artifactId>
- <version>${hbase.version}</version>
- <exclusions>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </exclusion>
- <exclusion>
- <!-- This is leaking from hadoop-annotations. -->
- <groupId>jdk.tools</groupId>
- <artifactId>jdk.tools</artifactId>
- </exclusion>
- <!-- https://jira.apache.org/jira/browse/HBASE-12909 -->
- <exclusion>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>com.github.ben-manes.caffeine</groupId>
- <artifactId>caffeine</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-autocreds</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- <version>${jackson.databind.version}</version>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <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-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
deleted file mode 100644
index 4d0855e2c..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
+++ /dev/null
@@ -1,82 +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.hbase.bolt;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.commons.lang.Validate;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.storm.Config;
-import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.base.BaseRichBolt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-// TODO support more configuration options, for now we're defaulting to the hbase-*.xml files found on the classpath
-public abstract class AbstractHBaseBolt extends BaseRichBolt {
- private static final Logger LOG = LoggerFactory.getLogger(AbstractHBaseBolt.class);
-
- protected transient OutputCollector collector;
- @SuppressWarnings("checkstyle:MemberName")
- protected transient HBaseClient hBaseClient;
- protected String tableName;
- protected HBaseMapper mapper;
- protected String configKey;
-
- public AbstractHBaseBolt(String tableName, HBaseMapper mapper) {
- Validate.notEmpty(tableName, "Table name can not be blank or null");
- Validate.notNull(mapper, "mapper can not be null");
- this.tableName = tableName;
- this.mapper = mapper;
- }
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext topologyContext, OutputCollector collector) {
- this.collector = collector;
- final Configuration hbConfig = HBaseConfiguration.create();
-
- Map<String, Object> conf = (Map<String, Object>) topoConf.get(this.configKey);
- if (conf == null) {
- LOG.warn("HBase configuration not found using key '" + this.configKey + "'");
- conf = Collections.emptyMap();
- }
-
- if (conf.get("hbase.rootdir") == null) {
- LOG.warn("No 'hbase.rootdir' value found in configuration! Using HBase defaults.");
- }
- for (String key : conf.keySet()) {
- hbConfig.set(key, String.valueOf(conf.get(key)));
- }
-
- //heck for backward compatibility, we need to pass TOPOLOGY_AUTO_CREDENTIALS to hbase conf
- //the conf instance is instance of persistentMap so making a copy.
- Map<String, Object> hbaseConfMap = new HashMap<>(conf);
- hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, topoConf.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
- this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, tableName);
- }
-
- @Override
- public void cleanup() {
- try {
- hBaseClient.close();
- } catch (IOException e) {
- LOG.error("HBase Client Close Failed ", e);
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
deleted file mode 100644
index 16759ca42..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ /dev/null
@@ -1,114 +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.hbase.bolt;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-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;
-
-/**
- * Basic bolt for writing to HBase.
- *
- * <p>Note: Each HBaseBolt defined in a topology is tied to a specific table.
- */
-@SuppressWarnings("checkstyle:AbbreviationAsWordInName")
-public class HBaseBolt extends AbstractHBaseBolt {
- private static final Logger LOG = LoggerFactory.getLogger(HBaseBolt.class);
- private static final int DEFAULT_FLUSH_INTERVAL_SECS = 1;
-
- @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
- boolean writeToWAL = true;
- List<Mutation> batchMutations;
- int flushIntervalSecs = DEFAULT_FLUSH_INTERVAL_SECS;
- int batchSize;
- BatchHelper batchHelper;
-
- public HBaseBolt(String tableName, HBaseMapper mapper) {
- super(tableName, mapper);
- this.batchMutations = new LinkedList<>();
- }
-
- @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
- public HBaseBolt writeToWAL(boolean writeToWAL) {
- this.writeToWAL = writeToWAL;
- return this;
- }
-
- public HBaseBolt withConfigKey(String configKey) {
- this.configKey = configKey;
- return this;
- }
-
- public HBaseBolt withBatchSize(int batchSize) {
- this.batchSize = batchSize;
- return this;
- }
-
- public HBaseBolt withFlushIntervalSecs(int flushIntervalSecs) {
- this.flushIntervalSecs = flushIntervalSecs;
- return this;
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return TupleUtils.putTickFrequencyIntoComponentConfig(super.getComponentConfiguration(), flushIntervalSecs);
- }
-
-
- @Override
- public void execute(Tuple tuple) {
- try {
- if (batchHelper.shouldHandle(tuple)) {
- byte[] rowKey = this.mapper.rowKey(tuple);
- ColumnList cols = this.mapper.columns(tuple);
- List<Mutation> mutations =
- hBaseClient.constructMutationReq(rowKey, cols, writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
- batchMutations.addAll(mutations);
- batchHelper.addBatch(tuple);
- }
-
- if (batchHelper.shouldFlush()) {
- this.hBaseClient.batchMutate(batchMutations);
- LOG.debug("acknowledging tuples after batchMutate");
- batchHelper.ack();
- batchMutations.clear();
- }
- } catch (Exception e) {
- batchHelper.fail(e);
- batchMutations.clear();
- }
- }
-
- @Override
- public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector collector) {
- super.prepare(map, topologyContext, collector);
- this.batchHelper = new BatchHelper(batchSize, collector);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
-
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
deleted file mode 100644
index 85262ca1a..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
+++ /dev/null
@@ -1,117 +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.hbase.bolt;
-
-import com.github.benmanes.caffeine.cache.CacheLoader;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.LoadingCache;
-import com.google.common.collect.Lists;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.lang.Validate;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.TupleUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Basic bolt for querying from HBase.
- *
- * <p>Note: Each HBaseBolt defined in a topology is tied to a specific table.
- */
-public class HBaseLookupBolt extends AbstractHBaseBolt {
- private static final Logger LOG = LoggerFactory.getLogger(HBaseLookupBolt.class);
-
- private HBaseValueMapper rowToTupleMapper;
- private HBaseProjectionCriteria projectionCriteria;
- private transient LoadingCache<byte[], Result> cache;
- private transient boolean cacheEnabled;
-
- public HBaseLookupBolt(String tableName, HBaseMapper mapper, HBaseValueMapper rowToTupleMapper) {
- super(tableName, mapper);
- Validate.notNull(rowToTupleMapper, "rowToTupleMapper can not be null");
- this.rowToTupleMapper = rowToTupleMapper;
- }
-
- public HBaseLookupBolt withConfigKey(String configKey) {
- this.configKey = configKey;
- return this;
- }
-
- public HBaseLookupBolt withProjectionCriteria(HBaseProjectionCriteria projectionCriteria) {
- this.projectionCriteria = projectionCriteria;
- return this;
- }
-
- @Override
- public void prepare(Map<String, Object> config, TopologyContext topologyContext, OutputCollector collector) {
- super.prepare(config, topologyContext, collector);
- cacheEnabled = Boolean.parseBoolean(config.getOrDefault("hbase.cache.enable", "false").toString());
- int cacheTtl = Integer.parseInt(config.getOrDefault("hbase.cache.ttl.seconds", "300").toString());
- int maxCacheSize = Integer.parseInt(config.getOrDefault("hbase.cache.size", "1000").toString());
- if (cacheEnabled) {
- cache = Caffeine.newBuilder().maximumSize(maxCacheSize).expireAfterWrite(cacheTtl, TimeUnit.SECONDS)
- .build(new CacheLoader<byte[], Result>() {
-
- @Override
- public Result load(byte[] rowKey) throws Exception {
- Get get = hBaseClient.constructGetRequests(rowKey, projectionCriteria);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Cache miss for key:" + new String(rowKey));
- }
- return hBaseClient.batchGet(Lists.newArrayList(get))[0];
- }
-
- });
- }
- }
-
- @Override
- public void execute(Tuple tuple) {
- if (TupleUtils.isTick(tuple)) {
- collector.ack(tuple);
- return;
- }
- byte[] rowKey = this.mapper.rowKey(tuple);
- Result result = null;
- try {
- if (cacheEnabled) {
- result = cache.get(rowKey);
- } else {
- Get get = hBaseClient.constructGetRequests(rowKey, projectionCriteria);
- result = hBaseClient.batchGet(Lists.newArrayList(get))[0];
- }
- for (Values values : rowToTupleMapper.toValues(tuple, result)) {
- this.collector.emit(tuple, values);
- }
- this.collector.ack(tuple);
- } catch (Exception e) {
- this.collector.reportError(e);
- this.collector.fail(tuple);
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
- rowToTupleMapper.declareOutputFields(outputFieldsDeclarer);
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseMapper.java
deleted file mode 100644
index 278509390..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseMapper.java
+++ /dev/null
@@ -1,35 +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.hbase.bolt.mapper;
-
-import java.io.Serializable;
-
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.tuple.Tuple;
-
-/**
- * Maps a <code>org.apache.storm.tuple.Tuple</code> object to a row in an HBase table.
- */
-public interface HBaseMapper extends Serializable {
-
- /**
- * Given a tuple, return the HBase rowkey.
- */
- byte[] rowKey(Tuple tuple);
-
- /**
- * Given a tuple, return a list of HBase columns to insert.
- */
- ColumnList columns(Tuple tuple);
-
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseProjectionCriteria.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseProjectionCriteria.java
deleted file mode 100644
index c9ab6d80f..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseProjectionCriteria.java
+++ /dev/null
@@ -1,87 +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.hbase.bolt.mapper;
-
-import com.google.common.collect.Lists;
-import java.io.Serializable;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- * Allows the user to specify the projection criteria. If only columnFamily is specified all columns from that family will be returned. If a
- * column is specified only that column from that family will be returned.
- */
-public class HBaseProjectionCriteria implements Serializable {
- private List<byte[]> columnFamilies;
- private List<ColumnMetaData> columns;
-
- public HBaseProjectionCriteria() {
- columnFamilies = Lists.newArrayList();
- columns = Lists.newArrayList();
- }
-
- /**
- * all columns from this family will be included as result of HBase lookup.
- */
- public HBaseProjectionCriteria addColumnFamily(String columnFamily) {
- this.columnFamilies.add(columnFamily.getBytes());
- return this;
- }
-
- /**
- * all columns from this family will be included as result of HBase lookup.
- */
- public HBaseProjectionCriteria addColumnFamily(byte[] columnFamily) {
- this.columnFamilies.add(Arrays.copyOf(columnFamily, columnFamily.length));
- return this;
- }
-
- /**
- * Only this column from the the columnFamily will be included as result of HBase lookup.
- */
- public HBaseProjectionCriteria addColumn(ColumnMetaData column) {
- this.columns.add(column);
- return this;
- }
-
- public List<ColumnMetaData> getColumns() {
- return columns;
- }
-
- public List<byte[]> getColumnFamilies() {
- return columnFamilies;
- }
-
- public static class ColumnMetaData implements Serializable {
- private byte[] columnFamily;
- private byte[] qualifier;
-
- public ColumnMetaData(String columnFamily, String qualifier) {
- this.columnFamily = columnFamily.getBytes();
- this.qualifier = qualifier.getBytes();
- }
-
- public ColumnMetaData(byte[] columnFamily, byte[] qualifier) {
- this.columnFamily = Arrays.copyOf(columnFamily, columnFamily.length);
- this.qualifier = Arrays.copyOf(qualifier, qualifier.length);
- }
-
- public byte[] getColumnFamily() {
- return columnFamily;
- }
-
- public byte[] getQualifier() {
- return qualifier;
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java
deleted file mode 100644
index a234de718..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java
+++ /dev/null
@@ -1,36 +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.hbase.bolt.mapper;
-
-import java.io.Serializable;
-import java.util.List;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.ITuple;
-import org.apache.storm.tuple.Values;
-
-public interface HBaseValueMapper extends Serializable {
-
- /**
- * Map to values.
- * @param input tuple
- * @param result HBase lookup result instance
- * @return list of values that should be emitted by the lookup bolt
- */
- List<Values> toValues(ITuple input, Result result) throws Exception;
-
- /**
- * declares the output fields for the lookup bolt.
- */
- void declareOutputFields(OutputFieldsDeclarer declarer);
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/SimpleHBaseMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/SimpleHBaseMapper.java
deleted file mode 100644
index 152a1b6b5..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/mapper/SimpleHBaseMapper.java
+++ /dev/null
@@ -1,85 +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.hbase.bolt.mapper;
-
-import static org.apache.storm.hbase.common.Utils.toBytes;
-import static org.apache.storm.hbase.common.Utils.toLong;
-
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SimpleHBaseMapper implements HBaseMapper {
- private static final Logger LOG = LoggerFactory.getLogger(SimpleHBaseMapper.class);
-
- private String rowKeyField;
- // private String timestampField;
- private byte[] columnFamily;
- private Fields columnFields;
- private Fields counterFields;
-
- public SimpleHBaseMapper() {
- }
-
-
- public SimpleHBaseMapper withRowKeyField(String rowKeyField) {
- this.rowKeyField = rowKeyField;
- return this;
- }
-
- public SimpleHBaseMapper withColumnFields(Fields columnFields) {
- this.columnFields = columnFields;
- return this;
- }
-
- public SimpleHBaseMapper withCounterFields(Fields counterFields) {
- this.counterFields = counterFields;
- return this;
- }
-
- public SimpleHBaseMapper withColumnFamily(String columnFamily) {
- this.columnFamily = columnFamily.getBytes();
- return this;
- }
-
- // public SimpleTridentHBaseMapper withTimestampField(String timestampField){
- // this.timestampField = timestampField;
- // return this;
- // }
-
- @Override
- public byte[] rowKey(Tuple tuple) {
- Object objVal = tuple.getValueByField(this.rowKeyField);
- return toBytes(objVal);
- }
-
- @Override
- public ColumnList columns(Tuple tuple) {
- ColumnList cols = new ColumnList();
- if (this.columnFields != null) {
- // TODO timestamps
- for (String field : this.columnFields) {
- cols.addColumn(this.columnFamily, field.getBytes(), toBytes(tuple.getValueByField(field)));
- }
- }
- if (this.counterFields != null) {
- for (String field : this.counterFields) {
- cols.addCounter(this.columnFamily, field.getBytes(), toLong(tuple.getValueByField(field)));
- }
- }
- return cols;
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ColumnList.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ColumnList.java
deleted file mode 100644
index 897146989..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ColumnList.java
+++ /dev/null
@@ -1,198 +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.hbase.common;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Represents a list of HBase columns.
- *
- * <p>There are two types of columns, <i>standard</i> and <i>counter</i>.
- *
- * <p>Standard columns have <i>column family</i> (required), <i>qualifier</i> (optional),
- * <i>timestamp</i> (optional), and a <i>value</i> (optional) values.
- *
- * <p>Counter columns have <i>column family</i> (required), <i>qualifier</i> (optional), and an <i>increment</i> (optional, but recommended)
- * values.
- *
- * <p>Inserts/Updates can be added via the <code>addColumn()</code> and <code>addCounter()</code> methods.
- */
-public class ColumnList {
-
- private ArrayList<Column> columns;
- private ArrayList<Column> columnsToDelete;
- private ArrayList<Counter> counters;
-
- private ArrayList<Column> columns() {
- if (this.columns == null) {
- this.columns = new ArrayList<Column>();
- }
- return this.columns;
- }
-
- private ArrayList<Column> columnsToDelete() {
- if (this.columnsToDelete == null) {
- this.columnsToDelete = new ArrayList<Column>();
- }
- return this.columnsToDelete;
- }
-
- private ArrayList<Counter> counters() {
- if (this.counters == null) {
- this.counters = new ArrayList<Counter>();
- }
- return this.counters;
- }
-
- /**
- * Add a standard HBase column.
- */
- public ColumnList addColumn(byte[] family, byte[] qualifier, long ts, byte[] value) {
- columns().add(new Column(family, qualifier, ts, value));
- return this;
- }
-
- /**
- * Add a standard HBase column.
- */
- public ColumnList addColumn(byte[] family, byte[] qualifier, byte[] value) {
- columns().add(new Column(family, qualifier, -1, value));
- return this;
- }
-
- /**
- * Add a standard HBase column given an instance of a class that implements the <code>IColumn</code> interface.
- */
- public ColumnList addColumn(IColumn column) {
- return this.addColumn(column.family(), column.qualifier(), column.timestamp(), column.value());
- }
-
- /**
- * Add an HBase counter column.
- */
- public ColumnList addCounter(byte[] family, byte[] qualifier, long incr) {
- counters().add(new Counter(family, qualifier, incr));
- return this;
- }
-
- /**
- * Add an HBase counter column given an instance of a class that implements the
- * <code>ICounter</code> interface.
- */
- public ColumnList addCounter(ICounter counter) {
- return this.addCounter(counter.family(), counter.qualifier(), counter.increment());
- }
-
- /**
- * Remove a standard HBase column.
- */
- public ColumnList deleteColumn(byte[] family, byte[] qualifier) {
- columnsToDelete().add(new Column(family, qualifier, -1, null));
- return this;
- }
-
- /**
- * Query to determine if we have column definitions.
- */
- public boolean hasColumns() {
- return this.columns != null;
- }
-
- /**
- * Query to determine if we have column delete definitions.
- */
- public boolean hasColumnsToDelete() {
- return this.columnsToDelete != null;
- }
-
- /**
- * Query to determine if we have counter definitions.
- */
- public boolean hasCounters() {
- return this.counters != null;
- }
-
- /**
- * Get the list of column definitions.
- */
- public List<Column> getColumns() {
- return this.columns;
- }
-
- /**
- * Get the list of 'column to delete' definitions.
- */
- public ArrayList<Column> getColumnsToDelete() {
- return this.columnsToDelete;
- }
-
- /**
- * Get the list of counter definitions.
- */
- public List<Counter> getCounters() {
- return this.counters;
- }
-
- public abstract static class AbstractColumn {
- byte[] family;
- byte[] qualifier;
-
- AbstractColumn(byte[] family, byte[] qualifier) {
- this.family = family;
- this.qualifier = qualifier;
- }
-
- public byte[] getFamily() {
- return family;
- }
-
- public byte[] getQualifier() {
- return qualifier;
- }
-
- }
-
- public static class Column extends AbstractColumn {
- byte[] value;
- long ts = -1;
-
- Column(byte[] family, byte[] qualifier, long ts, byte[] value) {
- super(family, qualifier);
- this.value = value;
- this.ts = ts;
- }
-
- public byte[] getValue() {
- return value;
- }
-
- public long getTs() {
- return ts;
- }
- }
-
- public static class Counter extends AbstractColumn {
- long incr = 0;
-
- Counter(byte[] family, byte[] qualifier, long incr) {
- super(family, qualifier);
- this.incr = incr;
- }
-
- public long getIncrement() {
- return incr;
- }
- }
-
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
deleted file mode 100644
index 2d0c3249e..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
+++ /dev/null
@@ -1,181 +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.hbase.common;
-
-import com.google.common.collect.Lists;
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Increment;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.security.UserProvider;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.security.HBaseSecurityUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * HBase client for constructing requests and do actual interaction with HBase.
- */
-public class HBaseClient implements Closeable {
- private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class);
-
- private Table table;
-
- public HBaseClient(Map<String, Object> map, final Configuration configuration, final String tableName) {
- try {
- UserProvider provider = HBaseSecurityUtil.login(map, configuration);
- this.table = Utils.getTable(provider, configuration, tableName);
- } catch (Exception e) {
- throw new RuntimeException("HBase bolt preparation failed: " + e.getMessage(), e);
- }
- }
-
- public List<Mutation> constructMutationReq(byte[] rowKey, ColumnList cols, Durability durability) {
- List<Mutation> mutations = Lists.newArrayList();
-
- if (cols.hasColumns()) {
- Put put = new Put(rowKey);
- put.setDurability(durability);
- for (ColumnList.Column col : cols.getColumns()) {
- if (col.getTs() > 0) {
- put.addColumn(
- col.getFamily(),
- col.getQualifier(),
- col.getTs(),
- col.getValue()
- );
- } else {
- put.addColumn(
- col.getFamily(),
- col.getQualifier(),
- col.getValue()
- );
- }
- }
- mutations.add(put);
- }
-
- if (cols.hasCounters()) {
- Increment inc = new Increment(rowKey);
- inc.setDurability(durability);
- for (ColumnList.Counter cnt : cols.getCounters()) {
- inc.addColumn(
- cnt.getFamily(),
- cnt.getQualifier(),
- cnt.getIncrement()
- );
- }
- mutations.add(inc);
- }
-
- if (cols.hasColumnsToDelete()) {
- Delete delete = new Delete(rowKey);
- delete.setDurability(durability);
- for (ColumnList.Column col : cols.getColumnsToDelete()) {
- if (col.getTs() > 0) {
- delete.addColumn(col.getFamily(), col.getQualifier(), col.getTs());
- } else {
- delete.addColumn(col.getFamily(), col.getQualifier());
- }
- }
- mutations.add(delete);
- }
-
- if (mutations.isEmpty()) {
- mutations.add(new Put(rowKey));
- }
- return mutations;
- }
-
- public void batchMutate(List<Mutation> mutations) throws Exception {
- Object[] result = new Object[mutations.size()];
- try {
- table.batch(mutations, result);
- } catch (InterruptedException e) {
- LOG.warn("Error performing a mutation to HBase.", e);
- throw e;
- } catch (IOException e) {
- LOG.warn("Error performing a mutation to HBase.", e);
- throw e;
- }
- }
-
-
- public Get constructGetRequests(byte[] rowKey, HBaseProjectionCriteria projectionCriteria) {
- Get get = new Get(rowKey);
-
- if (projectionCriteria != null) {
- for (byte[] columnFamily : projectionCriteria.getColumnFamilies()) {
- get.addFamily(columnFamily);
- }
-
- for (HBaseProjectionCriteria.ColumnMetaData columnMetaData : projectionCriteria.getColumns()) {
- get.addColumn(columnMetaData.getColumnFamily(), columnMetaData.getQualifier());
- }
- }
-
- return get;
- }
-
- public Result[] batchGet(List<Get> gets) throws Exception {
- try {
- return table.get(gets);
- } catch (Exception e) {
- LOG.warn("Could not perform HBASE lookup.", e);
- throw e;
- }
- }
-
- public ResultScanner scan(byte[] startRow, byte[] stopRow) throws Exception {
- try {
- if (startRow == null) {
- startRow = HConstants.EMPTY_START_ROW;
- }
- if (stopRow == null) {
- stopRow = HConstants.EMPTY_END_ROW;
- }
-
- Scan scan = new Scan(startRow, stopRow);
- return table.getScanner(scan);
- } catch (Exception e) {
- LOG.warn("Could not open HBASE scanner.", e);
- throw e;
- }
- }
-
- public boolean exists(Get get) throws Exception {
- try {
- return table.exists(get);
- } catch (Exception e) {
- LOG.warn("Could not perform HBASE existence check.", e);
- throw e;
- }
- }
-
- @Override
- public void close() throws IOException {
- table.close();
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/IColumn.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/IColumn.java
deleted file mode 100644
index 8f64c2e9d..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/IColumn.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.hbase.common;
-
-/**
- * Interface definition for classes that support being written to HBase as a regular column.
- */
-public interface IColumn {
- byte[] family();
-
- byte[] qualifier();
-
- byte[] value();
-
- long timestamp();
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ICounter.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ICounter.java
deleted file mode 100644
index ce9d84fc1..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/ICounter.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.hbase.common;
-
-/**
- * Interface definition for classes that support being written to HBase as a counter column.
- */
-public interface ICounter {
- byte[] family();
-
- byte[] qualifier();
-
- long increment();
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java
deleted file mode 100644
index bba279887..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.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.hbase.common;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.security.PrivilegedExceptionAction;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.security.UserProvider;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class Utils {
- public static final String TOKEN_KIND_HBASE_AUTH_TOKEN = "HBASE_AUTH_TOKEN";
- private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
-
- private Utils() {}
-
- public static Table getTable(UserProvider provider, Configuration config, String tableName)
- throws IOException, InterruptedException {
- UserGroupInformation ugi;
- if (provider != null) {
- ugi = provider.getCurrent().getUGI();
- LOG.debug("Current USER for provider: {}", ugi.getUserName());
- } else {
- // autocreds puts delegation token into current user UGI
- ugi = UserGroupInformation.getCurrentUser();
-
- LOG.debug("UGI for current USER : {}", ugi.getUserName());
- if (ugi.hasKerberosCredentials()) {
- LOG.debug("UGI has Kerberos credentials");
- } else {
- boolean foundHBaseAuthToken = false;
- for (Token<? extends TokenIdentifier> token : ugi.getTokens()) {
- LOG.debug("Token in UGI (delegation token): {} / {}", token.toString(),
- token.decodeIdentifier().getUser());
-
- // token.getKind() = Text, Text is annotated by @Stringable
- // which ensures toString() implementation
- if (token.getKind().toString().equals(TOKEN_KIND_HBASE_AUTH_TOKEN)) {
- // use UGI from token
- if (!foundHBaseAuthToken) {
- LOG.debug("Found HBASE_AUTH_TOKEN - using the token to replace current user.");
-
- ugi = token.decodeIdentifier().getUser();
- ugi.addToken(token);
-
- foundHBaseAuthToken = true;
- } else {
- LOG.warn("Found multiple HBASE_AUTH_TOKEN - will use already found token. "
- + "Please enable DEBUG log level to track delegation tokens.");
- }
- }
- }
-
- if (!foundHBaseAuthToken) {
- LOG.warn("Can't find HBase auth token in delegation tokens.");
- }
- }
- }
-
- return ugi.doAs(new PrivilegedExceptionAction<Table>() {
- @Override
- public Table run() throws IOException {
- return ConnectionFactory.createConnection(config).getTable(TableName.valueOf(tableName));
- }
- });
- }
-
- public static long toLong(Object obj) {
- long l = 0;
- if (obj != null) {
- if (obj instanceof Number) {
- l = ((Number) obj).longValue();
- } else {
- LOG.warn("Could not coerce {} to Long", obj.getClass().getName());
- }
- }
- return l;
- }
-
- public static byte[] toBytes(Object obj) {
- if (obj == null) {
- return null;
- } else if (obj instanceof String) {
- return ((String) obj).getBytes();
- } else if (obj instanceof Integer) {
- return Bytes.toBytes((Integer) obj);
- } else if (obj instanceof Long) {
- return Bytes.toBytes((Long) obj);
- } else if (obj instanceof Short) {
- return Bytes.toBytes((Short) obj);
- } else if (obj instanceof Float) {
- return Bytes.toBytes((Float) obj);
- } else if (obj instanceof Double) {
- return Bytes.toBytes((Double) obj);
- } else if (obj instanceof Boolean) {
- return Bytes.toBytes((Boolean) obj);
- } else if (obj instanceof BigDecimal) {
- return Bytes.toBytes((BigDecimal) obj);
- } else {
- LOG.error("Can't convert class to byte array: " + obj.getClass().getName());
- return new byte[0];
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueState.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueState.java
deleted file mode 100644
index 0da456c3b..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueState.java
+++ /dev/null
@@ -1,434 +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.hbase.state;
-
-import com.google.common.collect.Maps;
-import com.google.common.primitives.UnsignedBytes;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.TreeMap;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.state.DefaultStateEncoder;
-import org.apache.storm.state.DefaultStateSerializer;
-import org.apache.storm.state.KeyValueState;
-import org.apache.storm.state.Serializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A Hbase based implementation that persists the state in HBase.
- */
-public class HBaseKeyValueState<K, V> implements KeyValueState<K, V> {
- public static final int ITERATOR_CHUNK_SIZE = 1000;
- public static final NavigableMap<byte[], byte[]> EMPTY_PENDING_COMMIT_MAP = Maps.unmodifiableNavigableMap(
- new TreeMap<byte[], byte[]>(UnsignedBytes.lexicographicalComparator()));
- private static final Logger LOG = LoggerFactory.getLogger(HBaseKeyValueState.class);
- public static byte[] STATE_QUALIFIER = "s".getBytes();
- private static byte[] COMMIT_TXID_KEY = "commit".getBytes();
- private static byte[] PREPARE_TXID_KEY = "prepare".getBytes();
-
- private final byte[] keyNamespace;
- private final byte[] prepareNamespace;
- private final byte[] txidNamespace;
- private final String namespace;
- private final byte[] columnFamily;
- private final DefaultStateEncoder<K, V> encoder;
- private final HBaseClient hbaseClient;
-
- private ConcurrentNavigableMap<byte[], byte[]> pendingPrepare;
- private NavigableMap<byte[], byte[]> pendingCommit;
-
- // the key and value of txIds are guaranteed to be converted to UTF-8 encoded String
- private NavigableMap<byte[], byte[]> txIds;
-
- /**
- * Constructor.
- *
- * @param hbaseClient HBaseClient instance
- * @param columnFamily column family to store State
- * @param namespace namespace
- */
- public HBaseKeyValueState(HBaseClient hbaseClient, String columnFamily, String namespace) {
- this(hbaseClient, columnFamily, namespace, new DefaultStateSerializer<K>(),
- new DefaultStateSerializer<V>());
- }
-
- /**
- * Constructor.
- *
- * @param hbaseClient HBaseClient instance
- * @param columnFamily column family to store State
- * @param namespace namespace
- * @param keySerializer key serializer
- * @param valueSerializer value serializer
- */
- public HBaseKeyValueState(HBaseClient hbaseClient, String columnFamily, String namespace,
- Serializer<K> keySerializer, Serializer<V> valueSerializer) {
-
- this.hbaseClient = hbaseClient;
- this.columnFamily = columnFamily.getBytes();
- this.namespace = namespace;
- this.keyNamespace = (namespace + "$key:").getBytes();
- this.prepareNamespace = (namespace + "$prepare").getBytes();
- this.txidNamespace = (namespace + "$txid").getBytes();
- this.encoder = new DefaultStateEncoder<K, V>(keySerializer, valueSerializer);
- this.pendingPrepare = createPendingPrepareMap();
- initTxids();
- initPendingCommit();
- }
-
- private void initTxids() {
- HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
- criteria.addColumnFamily(columnFamily);
- Get get = hbaseClient.constructGetRequests(txidNamespace, criteria);
- try {
- Result[] results = hbaseClient.batchGet(Collections.singletonList(get));
- Result result = results[0];
- if (!result.isEmpty()) {
- NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(columnFamily);
- txIds = new TreeMap<>(familyMap);
- } else {
- txIds = new TreeMap<>(UnsignedBytes.lexicographicalComparator());
- }
-
- LOG.debug("initTxids, txIds {}", txIds);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- private void initPendingCommit() {
- HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
- criteria.addColumnFamily(columnFamily);
- Get get = hbaseClient.constructGetRequests(prepareNamespace, criteria);
- try {
- Result[] results = hbaseClient.batchGet(Collections.singletonList(get));
- Result result = results[0];
- if (!result.isEmpty()) {
- LOG.debug("Loading previously prepared commit from {}", prepareNamespace);
- NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(columnFamily);
- pendingCommit = Maps.unmodifiableNavigableMap(familyMap);
- } else {
- LOG.debug("No previously prepared commits.");
- pendingCommit = EMPTY_PENDING_COMMIT_MAP;
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void put(K key, V value) {
- LOG.debug("put key '{}', value '{}'", key, value);
- byte[] columnKey = encoder.encodeKey(key);
- byte[] columnValue = encoder.encodeValue(value);
- pendingPrepare.put(columnKey, columnValue);
- }
-
- @Override
- public V get(K key) {
- LOG.debug("get key '{}'", key);
- byte[] columnKey = encoder.encodeKey(key);
- byte[] columnValue = null;
-
- if (pendingPrepare.containsKey(columnKey)) {
- columnValue = pendingPrepare.get(columnKey);
- } else if (pendingCommit.containsKey(columnKey)) {
- columnValue = pendingCommit.get(columnKey);
- } else {
- HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
- HBaseProjectionCriteria.ColumnMetaData column = new HBaseProjectionCriteria.ColumnMetaData(columnFamily,
- STATE_QUALIFIER);
- criteria.addColumn(column);
- Get get = hbaseClient.constructGetRequests(getRowKeyForStateKey(columnKey), criteria);
- try {
- Result[] results = hbaseClient.batchGet(Collections.singletonList(get));
- Result result = results[0];
- columnValue = result.getValue(column.getColumnFamily(), column.getQualifier());
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- V value = null;
- if (columnValue != null) {
- value = encoder.decodeValue(columnValue);
- }
- LOG.debug("Value for key '{}' is '{}'", key, value);
- return value;
- }
-
- @Override
- public V get(K key, V defaultValue) {
- V val = get(key);
- return val != null ? val : defaultValue;
- }
-
- @Override
- public V delete(K key) {
- LOG.debug("delete key '{}'", key);
- byte[] columnKey = encoder.encodeKey(key);
- V curr = get(key);
- pendingPrepare.put(columnKey, encoder.getTombstoneValue());
- return curr;
- }
-
- @Override
- public Iterator<Map.Entry<K, V>> iterator() {
- return new HBaseKeyValueStateIterator<>(namespace, columnFamily, hbaseClient, pendingPrepare.entrySet().iterator(),
- pendingCommit.entrySet().iterator(), ITERATOR_CHUNK_SIZE, encoder.getKeySerializer(),
- encoder.getValueSerializer());
- }
-
- @Override
- public void prepareCommit(long txid) {
- LOG.debug("prepareCommit txid {}", txid);
- validatePrepareTxid(txid);
-
- try {
- ConcurrentNavigableMap<byte[], byte[]> currentPending = pendingPrepare;
- pendingPrepare = createPendingPrepareMap();
-
- Result result = getColumnFamily(prepareNamespace, columnFamily);
- if (!result.isEmpty()) {
- LOG.debug("Prepared txn already exists, will merge", txid);
- for (Map.Entry<byte[], byte[]> e : pendingCommit.entrySet()) {
- if (!currentPending.containsKey(e.getKey())) {
- currentPending.put(e.getKey(), e.getValue());
- }
- }
- } else {
- LOG.debug("Nothing to save for prepareCommit, txid {}.", txid);
- }
-
- if (!currentPending.isEmpty()) {
- mutateRow(prepareNamespace, columnFamily, currentPending);
- } else {
- LOG.debug("Nothing to save for prepareCommit, txid {}.", txid);
- }
-
- txIds.put(PREPARE_TXID_KEY, String.valueOf(txid).getBytes());
- mutateRow(txidNamespace, columnFamily, txIds);
- pendingCommit = Maps.unmodifiableNavigableMap(currentPending);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void commit(long txid) {
- LOG.debug("commit txid {}", txid);
- validateCommitTxid(txid);
- try {
- if (!pendingCommit.isEmpty()) {
- applyPendingStateToHBase(pendingCommit);
- } else {
- LOG.debug("Nothing to save for commit, txid {}.", txid);
- }
- txIds.put(COMMIT_TXID_KEY, String.valueOf(txid).getBytes());
- mutateRow(txidNamespace, columnFamily, txIds);
- deleteRow(prepareNamespace);
- pendingCommit = EMPTY_PENDING_COMMIT_MAP;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void commit() {
- if (!pendingPrepare.isEmpty()) {
- try {
- applyPendingStateToHBase(pendingPrepare);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- } else {
- LOG.debug("Nothing to save for commit");
- }
- pendingPrepare = createPendingPrepareMap();
- }
-
- @Override
- public void rollback() {
- LOG.debug("rollback");
- try {
- if (existsRow(prepareNamespace)) {
- deleteRow(prepareNamespace);
- } else {
- LOG.debug("Nothing to rollback, prepared data is empty");
- }
- Long lastCommittedId = lastCommittedTxid();
- if (lastCommittedId != null) {
- txIds.put(PREPARE_TXID_KEY, String.valueOf(lastCommittedId).getBytes());
- } else {
- txIds.remove(PREPARE_TXID_KEY);
- }
- if (!txIds.isEmpty()) {
- LOG.debug("put txidNamespace {}, txIds {}", txidNamespace, txIds);
- mutateRow(txidNamespace, columnFamily, txIds);
- }
- pendingCommit = EMPTY_PENDING_COMMIT_MAP;
- pendingPrepare = createPendingPrepareMap();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- /*
- * Same txid can be prepared again, but the next txid cannot be prepared
- * when previous one is not committed yet.
- */
- private void validatePrepareTxid(long txid) {
- Long committedTxid = lastCommittedTxid();
- if (committedTxid != null) {
- if (txid <= committedTxid) {
- throw new RuntimeException("Invalid txid '" + txid + "' for prepare. Txid '" + committedTxid
- + "' is already committed");
- }
- }
- }
-
- /*
- * Same txid can be committed again but the
- * txid to be committed must be the last prepared one.
- */
- private void validateCommitTxid(long txid) {
- Long committedTxid = lastCommittedTxid();
- if (committedTxid != null) {
- if (txid < committedTxid) {
- throw new RuntimeException("Invalid txid '" + txid + "' txid '" + committedTxid + "' is already committed");
- }
- }
- Long preparedTxid = lastPreparedTxid();
- if (preparedTxid != null) {
- if (txid != preparedTxid) {
- throw new RuntimeException("Invalid txid '" + txid + "' not same as prepared txid '" + preparedTxid + "'");
- }
- }
- }
-
-
- private Long lastCommittedTxid() {
- return lastId(COMMIT_TXID_KEY);
- }
-
- private Long lastPreparedTxid() {
- return lastId(PREPARE_TXID_KEY);
- }
-
- private Long lastId(byte[] key) {
- Long lastId = null;
- byte[] txId = txIds.get(key);
- if (txId != null) {
- lastId = Long.valueOf(new String(txId));
- }
- return lastId;
- }
-
- private byte[] getRowKeyForStateKey(byte[] columnKey) {
- byte[] rowKey = new byte[keyNamespace.length + columnKey.length];
- System.arraycopy(keyNamespace, 0, rowKey, 0, keyNamespace.length);
- System.arraycopy(columnKey, 0, rowKey, keyNamespace.length, columnKey.length);
- return rowKey;
- }
-
- private void applyPendingStateToHBase(NavigableMap<byte[], byte[]> pendingMap) throws Exception {
- List<Mutation> mutations = new ArrayList<>();
- for (Map.Entry<byte[], byte[]> entry : pendingMap.entrySet()) {
- byte[] rowKey = entry.getKey();
- byte[] value = entry.getValue();
-
- if (Arrays.equals(value, encoder.getTombstoneValue())) {
- mutations.add(new Delete(getRowKeyForStateKey(rowKey)));
- } else {
- List<Mutation> mutationsForRow = prepareMutateRow(getRowKeyForStateKey(rowKey), columnFamily,
- Collections.singletonMap(STATE_QUALIFIER, value));
- mutations.addAll(mutationsForRow);
- }
- }
-
- hbaseClient.batchMutate(mutations);
- }
-
- private Result getColumnFamily(byte[] rowKey, byte[] columnFamily) throws Exception {
- HBaseProjectionCriteria criteria = new HBaseProjectionCriteria();
- criteria.addColumnFamily(columnFamily);
- Get get = hbaseClient.constructGetRequests(rowKey, criteria);
- Result[] results = hbaseClient.batchGet(Collections.singletonList(get));
- return results[0];
- }
-
- private List<Mutation> prepareMutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map) {
- return prepareMutateRow(rowKey, columnFamily, map, Durability.USE_DEFAULT);
- }
-
- private List<Mutation> prepareMutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map,
- Durability durability) {
- ColumnList columnList = buildColumnList(columnFamily, map);
- return hbaseClient.constructMutationReq(rowKey, columnList, durability);
- }
-
- private void mutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map)
- throws Exception {
- mutateRow(rowKey, columnFamily, map, Durability.USE_DEFAULT);
- }
-
- private void mutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map,
- Durability durability) throws Exception {
- hbaseClient.batchMutate(prepareMutateRow(rowKey, columnFamily, map, durability));
- }
-
- private boolean existsRow(byte[] rowKey) throws Exception {
- Get get = new Get(rowKey);
- return hbaseClient.exists(get);
- }
-
- private void deleteRow(byte[] rowKey) throws Exception {
- Delete delete = new Delete(rowKey);
- hbaseClient.batchMutate(Collections.<Mutation>singletonList(delete));
- }
-
- private ColumnList buildColumnList(byte[] columnFamily, Map<byte[], byte[]> map) {
- ColumnList columnList = new ColumnList();
- for (Map.Entry<byte[], byte[]> entry : map.entrySet()) {
- columnList.addColumn(columnFamily, entry.getKey(), entry.getValue());
- }
- return columnList;
- }
-
- /**
- * Intended to extract this to separate method since only pendingPrepare uses ConcurrentNavigableMap.
- */
- private ConcurrentNavigableMap<byte[], byte[]> createPendingPrepareMap() {
- return new ConcurrentSkipListMap<>(UnsignedBytes.lexicographicalComparator());
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateIterator.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateIterator.java
deleted file mode 100644
index 13e9b99fb..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateIterator.java
+++ /dev/null
@@ -1,146 +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.hbase.state;
-
-import static org.apache.storm.hbase.state.HBaseKeyValueState.STATE_QUALIFIER;
-
-import com.google.common.primitives.UnsignedBytes;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.TreeMap;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.state.BaseBinaryStateIterator;
-import org.apache.storm.state.DefaultStateEncoder;
-import org.apache.storm.state.Serializer;
-import org.apache.storm.state.StateEncoder;
-
-/**
- * An iterator over {@link HBaseKeyValueState}.
- */
-public class HBaseKeyValueStateIterator<K, V> extends BaseBinaryStateIterator<K, V> {
-
- private final byte[] keyNamespace;
- private final byte[] endScanKey;
- private final byte[] columnFamily;
- private final HBaseClient hbaseClient;
- private final int chunkSize;
- private final StateEncoder<K, V, byte[], byte[]> encoder;
- private byte[] cursorKey;
- private Iterator<Map.Entry<byte[], byte[]>> cachedResultIterator;
-
- /**
- * Constructor.
- *
- * @param namespace The namespace of State
- * @param columnFamily The column family of state
- * @param hbaseClient The instance of HBaseClient
- * @param pendingPrepareIterator The iterator of pendingPrepare
- * @param pendingCommitIterator The iterator of pendingCommit
- * @param chunkSize The size of chunk to get entries from HBase
- * @param keySerializer The serializer of key
- * @param valueSerializer The serializer of value
- */
- public HBaseKeyValueStateIterator(String namespace, byte[] columnFamily, HBaseClient hbaseClient,
- Iterator<Map.Entry<byte[], byte[]>> pendingPrepareIterator,
- Iterator<Map.Entry<byte[], byte[]>> pendingCommitIterator,
- int chunkSize, Serializer<K> keySerializer,
- Serializer<V> valueSerializer) {
- super(pendingPrepareIterator, pendingCommitIterator);
- this.columnFamily = columnFamily;
- this.keyNamespace = (namespace + "$key:").getBytes();
- this.cursorKey = (namespace + "$key:").getBytes();
-
- // this is the end key for whole scan
- this.endScanKey = advanceRow(this.cursorKey);
- this.hbaseClient = hbaseClient;
- this.chunkSize = chunkSize;
- this.encoder = new DefaultStateEncoder<K, V>(keySerializer, valueSerializer);
- }
-
- @Override
- protected Iterator<Map.Entry<byte[], byte[]>> loadChunkFromStateStorage() {
- loadChunkFromHBase();
- return cachedResultIterator;
- }
-
- @Override
- protected boolean isEndOfDataFromStorage() {
- if (cachedResultIterator != null && cachedResultIterator.hasNext()) {
- return false;
- }
-
- try {
- ResultScanner resultScanner = hbaseClient.scan(cursorKey, endScanKey);
- return !(resultScanner.iterator().hasNext());
- } catch (Exception e) {
- throw new RuntimeException("Fail to scan from HBase state storage.");
- }
- }
-
- @Override
- protected K decodeKey(byte[] key) {
- return encoder.decodeKey(key);
- }
-
- @Override
- protected V decodeValue(byte[] value) {
- return encoder.decodeValue(value);
- }
-
- @Override
- protected boolean isTombstoneValue(byte[] value) {
- return Arrays.equals(value, encoder.getTombstoneValue());
- }
-
- private void loadChunkFromHBase() {
- Map<byte[], byte[]> chunk = new TreeMap<>(UnsignedBytes.lexicographicalComparator());
- try {
- ResultScanner resultScanner = hbaseClient.scan(cursorKey, endScanKey);
-
- Result[] results = resultScanner.next(chunkSize);
-
- for (Result result : results) {
- byte[] columnKey = extractStateKeyFromRowKey(result.getRow());
- byte[] columnValue = result.getValue(columnFamily, STATE_QUALIFIER);
-
- chunk.put(columnKey, columnValue);
- }
-
- if (results.length > 0) {
- byte[] lastRow = results[results.length - 1].getRow();
- cursorKey = advanceRow(lastRow);
- }
-
- cachedResultIterator = chunk.entrySet().iterator();
- } catch (Exception e) {
- throw new RuntimeException("Fail to scan from HBase state storage.", e);
- }
- }
-
- private byte[] advanceRow(byte[] row) {
- byte[] advancedRow = new byte[row.length];
- System.arraycopy(row, 0, advancedRow, 0, row.length);
- advancedRow[row.length - 1]++;
- return advancedRow;
- }
-
- private byte[] extractStateKeyFromRowKey(byte[] row) {
- byte[] stateKey = new byte[row.length - keyNamespace.length];
- System.arraycopy(row, keyNamespace.length, stateKey, 0, row.length - keyNamespace.length);
- return stateKey;
- }
-
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateProvider.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateProvider.java
deleted file mode 100644
index 3819bf044..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/state/HBaseKeyValueStateProvider.java
+++ /dev/null
@@ -1,165 +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.hbase.state;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.PropertyAccessor;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.storm.Config;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.state.DefaultStateSerializer;
-import org.apache.storm.state.Serializer;
-import org.apache.storm.state.State;
-import org.apache.storm.state.StateProvider;
-import org.apache.storm.task.TopologyContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Provides {@link HBaseKeyValueState}.
- */
-public class HBaseKeyValueStateProvider implements StateProvider {
- private static final Logger LOG = LoggerFactory.getLogger(HBaseKeyValueStateProvider.class);
-
- @Override
- public State newState(String namespace, Map<String, Object> stormConf, TopologyContext context) {
- try {
- return getHBaseKeyValueState(namespace, stormConf, context, getStateConfig(stormConf));
- } catch (Exception ex) {
- LOG.error("Error loading config from storm conf {}", stormConf);
- throw new RuntimeException(ex);
- }
- }
-
- StateConfig getStateConfig(Map stormConf) throws Exception {
- StateConfig stateConfig;
- String providerConfig;
- ObjectMapper mapper = new ObjectMapper();
- mapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
- if (stormConf.containsKey(Config.TOPOLOGY_STATE_PROVIDER_CONFIG)) {
- providerConfig = (String) stormConf.get(Config.TOPOLOGY_STATE_PROVIDER_CONFIG);
- stateConfig = mapper.readValue(providerConfig, StateConfig.class);
- } else {
- stateConfig = new StateConfig();
- }
-
- // assertion
- assertMandatoryParameterNotEmpty(stateConfig.hbaseConfigKey, "hbaseConfigKey");
- assertMandatoryParameterNotEmpty(stateConfig.tableName, "tableName");
- assertMandatoryParameterNotEmpty(stateConfig.columnFamily, "columnFamily");
-
- return stateConfig;
- }
-
- private HBaseKeyValueState getHBaseKeyValueState(String namespace, Map<String, Object> stormConf, TopologyContext context,
- StateConfig config) throws Exception {
- Map<String, Object> conf = getHBaseConfigMap(stormConf, config.hbaseConfigKey);
- final Configuration hbConfig = getHBaseConfigurationInstance(conf);
-
- //heck for backward compatibility, we need to pass TOPOLOGY_AUTO_CREDENTIALS to hbase conf
- //the conf instance is instance of persistentMap so making a copy.
- Map<String, Object> hbaseConfMap = new HashMap<>(conf);
- hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, stormConf.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
- HBaseClient hbaseClient = new HBaseClient(hbaseConfMap, hbConfig, config.tableName);
-
- return new HBaseKeyValueState(hbaseClient, config.columnFamily, namespace,
- getKeySerializer(stormConf, context, config), getValueSerializer(stormConf, context, config));
- }
-
- private Configuration getHBaseConfigurationInstance(Map<String, Object> conf) {
- final Configuration hbConfig = HBaseConfiguration.create();
- for (String key : conf.keySet()) {
- hbConfig.set(key, String.valueOf(conf.get(key)));
- }
- return hbConfig;
- }
-
- private Map<String, Object> getHBaseConfigMap(Map<String, Object> stormConfMap, String hbaseConfigKey) {
- Map<String, Object> conf = (Map<String, Object>) stormConfMap.get(hbaseConfigKey);
- if (conf == null) {
- throw new IllegalArgumentException("HBase configuration not found using key '" + hbaseConfigKey + "'");
- }
-
- if (conf.get("hbase.rootdir") == null) {
- LOG.warn("No 'hbase.rootdir' value found in configuration! Using HBase defaults.");
- }
- return conf;
- }
-
- private void assertMandatoryParameterNotEmpty(String paramValue, String fieldName) {
- if (StringUtils.isEmpty(paramValue)) {
- throw new IllegalArgumentException(fieldName + " should be provided.");
- }
- }
-
- private Serializer getKeySerializer(Map<String, Object> topoConf, TopologyContext context, StateConfig config) throws Exception {
- Serializer serializer;
- if (config.keySerializerClass != null) {
- Class<?> klass = Class.forName(config.keySerializerClass);
- serializer = (Serializer) klass.newInstance();
- } else if (config.keyClass != null) {
- serializer = new DefaultStateSerializer(topoConf, context, Collections.singletonList(Class.forName(config.keyClass)));
- } else {
- serializer = new DefaultStateSerializer(topoConf, context);
- }
- return serializer;
- }
-
- private Serializer getValueSerializer(Map<String, Object> topoConf, TopologyContext context, StateConfig config) throws Exception {
- Serializer serializer = null;
- if (config.valueSerializerClass != null) {
- Class<?> klass = (Class<?>) Class.forName(config.valueSerializerClass);
- serializer = (Serializer) klass.newInstance();
- } else if (config.valueClass != null) {
- serializer = new DefaultStateSerializer(topoConf, context, Collections.singletonList(Class.forName(config.valueClass)));
- } else {
- serializer = new DefaultStateSerializer(topoConf, context);
- }
- return serializer;
- }
-
- static class StateConfig {
- String keyClass;
- String valueClass;
- String keySerializerClass;
- String valueSerializerClass;
- String hbaseConfigKey;
- String tableName;
- String columnFamily;
-
- @Override
- public String toString() {
- return "StateConfig{"
- + "keyClass='" + keyClass + '\''
- + ", valueClass='" + valueClass + '\''
- + ", keySerializerClass='" + keySerializerClass + '\''
- + ", valueSerializerClass='" + valueSerializerClass + '\''
- + ", hbaseConfigKey='" + hbaseConfigKey + '\''
- + ", tableName='" + tableName + '\''
- + ", columnFamily='" + columnFamily + '\''
- + '}';
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapMapper.java
deleted file mode 100644
index 86f38d79b..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapMapper.java
+++ /dev/null
@@ -1,50 +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.hbase.trident.mapper;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.List;
-
-public class SimpleTridentHBaseMapMapper implements TridentHBaseMapMapper {
- private String qualifier;
-
- public SimpleTridentHBaseMapMapper(String qualifier) {
- this.qualifier = qualifier;
- }
-
- @Override
- public byte[] rowKey(List<Object> keys) {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- try {
- for (Object key : keys) {
- bos.write(String.valueOf(key).getBytes());
- }
- bos.close();
- } catch (IOException e) {
- throw new RuntimeException("IOException creating HBase row key.", e);
- }
- return bos.toByteArray();
- }
-
- @Override
- public String qualifier(List<Object> keys) {
- return this.qualifier;
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapper.java
deleted file mode 100644
index cf043af90..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/SimpleTridentHBaseMapper.java
+++ /dev/null
@@ -1,79 +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.hbase.trident.mapper;
-
-import static org.apache.storm.hbase.common.Utils.toBytes;
-import static org.apache.storm.hbase.common.Utils.toLong;
-
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SimpleTridentHBaseMapper implements TridentHBaseMapper {
- private static final Logger LOG = LoggerFactory.getLogger(SimpleTridentHBaseMapper.class);
-
- private String rowKeyField;
- private byte[] columnFamily;
- private Fields columnFields;
- private Fields counterFields;
-
- public SimpleTridentHBaseMapper() {
- }
-
-
- public SimpleTridentHBaseMapper withRowKeyField(String rowKeyField) {
- this.rowKeyField = rowKeyField;
- return this;
- }
-
- public SimpleTridentHBaseMapper withColumnFields(Fields columnFields) {
- this.columnFields = columnFields;
- return this;
- }
-
- public SimpleTridentHBaseMapper withCounterFields(Fields counterFields) {
- this.counterFields = counterFields;
- return this;
- }
-
- public SimpleTridentHBaseMapper withColumnFamily(String columnFamily) {
- this.columnFamily = columnFamily.getBytes();
- return this;
- }
-
-
- @Override
- public byte[] rowKey(TridentTuple tuple) {
- Object objVal = tuple.getValueByField(this.rowKeyField);
- return toBytes(objVal);
- }
-
- @Override
- public ColumnList columns(TridentTuple tuple) {
- ColumnList cols = new ColumnList();
- if (this.columnFields != null) {
- // TODO timestamps
- for (String field : this.columnFields) {
- cols.addColumn(this.columnFamily, field.getBytes(), toBytes(tuple.getValueByField(field)));
- }
- }
- if (this.counterFields != null) {
- for (String field : this.counterFields) {
- cols.addCounter(this.columnFamily, field.getBytes(), toLong(tuple.getValueByField(field)));
- }
- }
- return cols;
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapMapper.java
deleted file mode 100644
index fde9dea2c..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapMapper.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.hbase.trident.mapper;
-
-import java.io.Serializable;
-import java.util.List;
-
-public interface TridentHBaseMapMapper extends Serializable {
- /**
- * Given a tuple's grouped key list, return the HBase rowkey.
- */
- byte[] rowKey(List<Object> keys);
-
- /**
- * Given a tuple's grouped key list, return the HBase qualifier.
- */
- String qualifier(List<Object> keys);
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapper.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapper.java
deleted file mode 100644
index 9639f6e6e..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/mapper/TridentHBaseMapper.java
+++ /dev/null
@@ -1,40 +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.hbase.trident.mapper;
-
-import java.io.Serializable;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.trident.tuple.TridentTuple;
-
-/**
- * Maps a <code>org.apache.storm.trident.tuple.TridentTuple</code> object to a row in an HBase table.
- */
-public interface TridentHBaseMapper extends Serializable {
-
-
- /**
- * Given a tuple, return the HBase rowkey.
- */
- byte[] rowKey(TridentTuple tuple);
-
- /**
- * Given a tuple, return a list of HBase columns to insert.
- */
- ColumnList columns(TridentTuple tuple);
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
deleted file mode 100644
index 191d7cb70..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
+++ /dev/null
@@ -1,264 +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.hbase.trident.state;
-
-import com.google.common.collect.Maps;
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.security.UserProvider;
-import org.apache.storm.hbase.common.Utils;
-import org.apache.storm.hbase.security.HBaseSecurityUtil;
-import org.apache.storm.hbase.trident.mapper.TridentHBaseMapMapper;
-import org.apache.storm.task.IMetricsContext;
-import org.apache.storm.topology.FailedException;
-import org.apache.storm.trident.state.JSONNonTransactionalSerializer;
-import org.apache.storm.trident.state.JSONOpaqueSerializer;
-import org.apache.storm.trident.state.JSONTransactionalSerializer;
-import org.apache.storm.trident.state.OpaqueValue;
-import org.apache.storm.trident.state.Serializer;
-import org.apache.storm.trident.state.State;
-import org.apache.storm.trident.state.StateFactory;
-import org.apache.storm.trident.state.StateType;
-import org.apache.storm.trident.state.TransactionalValue;
-import org.apache.storm.trident.state.map.CachedMap;
-import org.apache.storm.trident.state.map.IBackingMap;
-import org.apache.storm.trident.state.map.MapState;
-import org.apache.storm.trident.state.map.NonTransactionalMap;
-import org.apache.storm.trident.state.map.OpaqueMap;
-import org.apache.storm.trident.state.map.SnapshottableMap;
-import org.apache.storm.trident.state.map.TransactionalMap;
-import org.apache.storm.tuple.Values;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class HBaseMapState<T> implements IBackingMap<T> {
- @SuppressWarnings("rawtypes")
- private static final Map<StateType, Serializer> DEFAULT_SERIALZERS = Maps.newHashMap();
- private static Logger LOG = LoggerFactory.getLogger(HBaseMapState.class);
-
- static {
- DEFAULT_SERIALZERS.put(StateType.NON_TRANSACTIONAL, new JSONNonTransactionalSerializer());
- DEFAULT_SERIALZERS.put(StateType.TRANSACTIONAL, new JSONTransactionalSerializer());
- DEFAULT_SERIALZERS.put(StateType.OPAQUE, new JSONOpaqueSerializer());
- }
-
- private int partitionNum;
- private Options<T> options;
- private Serializer<T> serializer;
- private Table table;
-
- /**
- * Constructor.
- *
- * @param options HBase State options.
- * @param map topology config map.
- * @param partitionNum the number of partition.
- */
- public HBaseMapState(final Options<T> options, Map<String, Object> map, int partitionNum) {
- this.options = options;
- this.serializer = options.serializer;
- this.partitionNum = partitionNum;
-
- final Configuration hbConfig = HBaseConfiguration.create();
- Map<String, Object> conf = (Map<String, Object>) map.get(options.configKey);
- if (conf == null) {
- LOG.info("HBase configuration not found using key '" + options.configKey + "'");
- LOG.info("Using HBase config from first hbase-site.xml found on classpath.");
- } else {
- if (conf.get("hbase.rootdir") == null) {
- LOG.warn("No 'hbase.rootdir' value found in configuration! Using HBase defaults.");
- }
- for (String key : conf.keySet()) {
- hbConfig.set(key, String.valueOf(conf.get(key)));
- }
- }
-
- try {
- UserProvider provider = HBaseSecurityUtil.login(map, hbConfig);
- this.table = Utils.getTable(provider, hbConfig, options.tableName);
- } catch (Exception e) {
- throw new RuntimeException("HBase bolt preparation failed: " + e.getMessage(), e);
- }
-
- }
-
- @SuppressWarnings("rawtypes")
- public static StateFactory opaque() {
- Options<OpaqueValue> options = new Options<OpaqueValue>();
- return opaque(options);
- }
-
- @SuppressWarnings("rawtypes")
- public static StateFactory opaque(Options<OpaqueValue> opts) {
-
- return new Factory(StateType.OPAQUE, opts);
- }
-
- @SuppressWarnings("rawtypes")
- public static StateFactory transactional() {
- Options<TransactionalValue> options = new Options<TransactionalValue>();
- return transactional(options);
- }
-
- @SuppressWarnings("rawtypes")
- public static StateFactory transactional(Options<TransactionalValue> opts) {
- return new Factory(StateType.TRANSACTIONAL, opts);
- }
-
- public static StateFactory nonTransactional() {
- Options<Object> options = new Options<Object>();
- return nonTransactional(options);
- }
-
- public static StateFactory nonTransactional(Options<Object> opts) {
- return new Factory(StateType.NON_TRANSACTIONAL, opts);
- }
-
- @Override
- public List<T> multiGet(List<List<Object>> keys) {
- List<Get> gets = new ArrayList<Get>();
- for (List<Object> key : keys) {
- byte[] hbaseKey = this.options.mapMapper.rowKey(key);
- String qualifier = this.options.mapMapper.qualifier(key);
-
- LOG.info("Partition: {}, GET: {}", this.partitionNum, new String(hbaseKey));
- Get get = new Get(hbaseKey);
- get.addColumn(this.options.columnFamily.getBytes(), qualifier.getBytes());
- gets.add(get);
- }
-
- List<T> retval = new ArrayList<T>();
- try {
- Result[] results = this.table.get(gets);
- for (int i = 0; i < keys.size(); i++) {
- String qualifier = this.options.mapMapper.qualifier(keys.get(i));
- Result result = results[i];
- byte[] value = result.getValue(this.options.columnFamily.getBytes(), qualifier.getBytes());
- if (value != null) {
- retval.add(this.serializer.deserialize(value));
- } else {
- retval.add(null);
- }
- }
- } catch (IOException e) {
- throw new FailedException("IOException while reading from HBase.", e);
- }
- return retval;
- }
-
- @Override
- public void multiPut(List<List<Object>> keys, List<T> values) {
- List<Put> puts = new ArrayList<Put>(keys.size());
- for (int i = 0; i < keys.size(); i++) {
- byte[] hbaseKey = this.options.mapMapper.rowKey(keys.get(i));
- String qualifier = this.options.mapMapper.qualifier(keys.get(i));
- LOG.info("Partiton: {}, Key: {}, Value: {}",
- new Object[]{ this.partitionNum, new String(hbaseKey), new String(this.serializer.serialize(values.get(i))) });
- Put put = new Put(hbaseKey);
- T val = values.get(i);
- put.addColumn(this.options.columnFamily.getBytes(),
- qualifier.getBytes(),
- this.serializer.serialize(val));
-
- puts.add(put);
- }
- try {
- this.table.put(puts);
- } catch (InterruptedIOException e) {
- throw new FailedException("Interrupted while writing to HBase", e);
- } catch (RetriesExhaustedWithDetailsException e) {
- throw new FailedException("Retries exhaused while writing to HBase", e);
- } catch (IOException e) {
- throw new FailedException("IOException while writing to HBase", e);
- }
- }
-
- public static class Options<T> implements Serializable {
-
- public Serializer<T> serializer = null;
- public int cacheSize = 5000;
- public String globalKey = "$HBASE_STATE_GLOBAL$";
- public String configKey = "hbase.config";
- public String tableName;
- public String columnFamily;
- public TridentHBaseMapMapper mapMapper;
- }
-
- protected static class Factory implements StateFactory {
- private StateType stateType;
- private Options options;
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- public Factory(StateType stateType, Options options) {
- this.stateType = stateType;
- this.options = options;
-
- if (this.options.serializer == null) {
- this.options.serializer = DEFAULT_SERIALZERS.get(stateType);
- }
-
- if (this.options.serializer == null) {
- throw new RuntimeException("Serializer should be specified for type: " + stateType);
- }
-
- if (this.options.mapMapper == null) {
- throw new RuntimeException("MapMapper should be specified for type: " + stateType);
- }
- }
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- @Override
- public State makeState(Map<String, Object> conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
- LOG.info("Preparing HBase State for partition {} of {}.", partitionIndex + 1, numPartitions);
- IBackingMap state = new HBaseMapState(options, conf, partitionIndex);
-
- if (options.cacheSize > 0) {
- state = new CachedMap(state, options.cacheSize);
- }
-
- MapState mapState;
- switch (stateType) {
- case NON_TRANSACTIONAL:
- mapState = NonTransactionalMap.build(state);
- break;
- case OPAQUE:
- mapState = OpaqueMap.build(state);
- break;
- case TRANSACTIONAL:
- mapState = TransactionalMap.build(state);
- break;
- default:
- throw new IllegalArgumentException("Unknown state type: " + stateType);
- }
- return new SnapshottableMap(mapState, new Values(options.globalKey));
- }
-
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseQuery.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseQuery.java
deleted file mode 100644
index 4de5873e6..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseQuery.java
+++ /dev/null
@@ -1,35 +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.hbase.trident.state;
-
-import java.util.List;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.state.BaseQueryFunction;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Values;
-
-public class HBaseQuery extends BaseQueryFunction<HBaseState, List<Values>> {
-
- @Override
- @SuppressWarnings("checkstyle:ParameterName")
- public List<List<Values>> batchRetrieve(HBaseState hBaseState, List<TridentTuple> tridentTuples) {
- return hBaseState.batchRetrieve(tridentTuples);
- }
-
- @Override
- public void execute(TridentTuple tuples, List<Values> values, TridentCollector tridentCollector) {
- for (Values value : values) {
- tridentCollector.emit(value);
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseState.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseState.java
deleted file mode 100644
index 338112130..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseState.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.hbase.trident.state;
-
-import com.google.common.collect.Lists;
-import java.io.Serializable;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.storm.Config;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.hbase.trident.mapper.TridentHBaseMapper;
-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.apache.storm.tuple.Values;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class HBaseState implements State {
-
- private static final Logger LOG = LoggerFactory.getLogger(HBaseState.class);
-
- private Options options;
- @SuppressWarnings("checkstyle:MemberName")
- private HBaseClient hBaseClient;
- private Map<String, Object> map;
- private int numPartitions;
- private int partitionIndex;
-
- protected HBaseState(Map<String, Object> map, int partitionIndex, int numPartitions, Options options) {
- this.options = options;
- this.map = map;
- this.partitionIndex = partitionIndex;
- this.numPartitions = numPartitions;
- }
-
- protected void prepare() {
- final Configuration hbConfig = HBaseConfiguration.create();
- Map<String, Object> conf = (Map<String, Object>) map.get(options.configKey);
- if (conf == null) {
- LOG.info("HBase configuration not found using key '" + options.configKey + "'");
- LOG.info("Using HBase config from first hbase-site.xml found on classpath.");
- conf = Collections.emptyMap();
- } else {
- if (conf.get("hbase.rootdir") == null) {
- LOG.warn("No 'hbase.rootdir' value found in configuration! Using HBase defaults.");
- }
- for (String key : conf.keySet()) {
- hbConfig.set(key, String.valueOf(conf.get(key)));
- }
- }
-
- //heck for backward compatibility, we need to pass TOPOLOGY_AUTO_CREDENTIALS to hbase conf
- //the conf instance is instance of persistentMap so making a copy.
- Map<String, Object> hbaseConfMap = new HashMap<>(conf);
- hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, map.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
-
- this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, options.tableName);
- }
-
- @Override
- public void beginCommit(Long someLong) {
- LOG.debug("beginCommit is noop.");
- }
-
- @Override
- public void commit(Long someLong) {
- LOG.debug("commit is noop.");
- }
-
- public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
- List<Mutation> mutations = Lists.newArrayList();
-
- for (TridentTuple tuple : tuples) {
- byte[] rowKey = options.mapper.rowKey(tuple);
- ColumnList cols = options.mapper.columns(tuple);
- mutations.addAll(hBaseClient.constructMutationReq(rowKey, cols, options.durability));
- }
-
- try {
- hBaseClient.batchMutate(mutations);
- } catch (Exception e) {
- collector.reportError(e);
- throw new FailedException(e);
- }
- }
-
- public List<List<Values>> batchRetrieve(List<TridentTuple> tridentTuples) {
- List<List<Values>> batchRetrieveResult = Lists.newArrayList();
- List<Get> gets = Lists.newArrayList();
- for (TridentTuple tuple : tridentTuples) {
- byte[] rowKey = options.mapper.rowKey(tuple);
- gets.add(hBaseClient.constructGetRequests(rowKey, options.projectionCriteria));
- }
-
- try {
- Result[] results = hBaseClient.batchGet(gets);
- for (int i = 0; i < results.length; i++) {
- Result result = results[i];
- TridentTuple tuple = tridentTuples.get(i);
- List<Values> values = options.rowToStormValueMapper.toValues(tuple, result);
- batchRetrieveResult.add(values);
- }
- } catch (Exception e) {
- LOG.warn("Batch get operation failed. Triggering replay.", e);
- throw new FailedException(e);
- }
- return batchRetrieveResult;
- }
-
- public static class Options implements Serializable {
- private TridentHBaseMapper mapper;
- private Durability durability = Durability.SKIP_WAL;
- private HBaseProjectionCriteria projectionCriteria;
- private HBaseValueMapper rowToStormValueMapper;
- private String configKey;
- private String tableName;
-
- public Options withDurability(Durability durability) {
- this.durability = durability;
- return this;
- }
-
- public Options withProjectionCriteria(HBaseProjectionCriteria projectionCriteria) {
- this.projectionCriteria = projectionCriteria;
- return this;
- }
-
- public Options withConfigKey(String configKey) {
- this.configKey = configKey;
- return this;
- }
-
- public Options withTableName(String tableName) {
- this.tableName = tableName;
- return this;
- }
-
- public Options withRowToStormValueMapper(HBaseValueMapper rowToStormValueMapper) {
- this.rowToStormValueMapper = rowToStormValueMapper;
- return this;
- }
-
- public Options withMapper(TridentHBaseMapper mapper) {
- this.mapper = mapper;
- return this;
- }
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseStateFactory.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseStateFactory.java
deleted file mode 100644
index 2c5cf94e6..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseStateFactory.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.hbase.trident.state;
-
-import java.util.Map;
-import org.apache.storm.task.IMetricsContext;
-import org.apache.storm.trident.state.State;
-import org.apache.storm.trident.state.StateFactory;
-
-public class HBaseStateFactory implements StateFactory {
-
- private HBaseState.Options options;
-
- public HBaseStateFactory(HBaseState.Options options) {
- this.options = options;
- }
-
- @Override
- public State makeState(Map<String, Object> map, IMetricsContext metricsContext, int partitionIndex, int numPartitions) {
- HBaseState state = new HBaseState(map, partitionIndex, numPartitions, options);
- state.prepare();
- return state;
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseUpdater.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseUpdater.java
deleted file mode 100644
index 2cde3c062..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseUpdater.java
+++ /dev/null
@@ -1,27 +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.hbase.trident.state;
-
-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;
-
-public class HBaseUpdater extends BaseStateUpdater<HBaseState> {
-
- @Override
- @SuppressWarnings("checkstyle:ParameterName")
- public void updateState(HBaseState hBaseState, List<TridentTuple> tuples, TridentCollector collector) {
- hBaseState.updateState(tuples, collector);
- }
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStore.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStore.java
deleted file mode 100644
index 3fe4bc4c4..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStore.java
+++ /dev/null
@@ -1,267 +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.hbase.trident.windowing;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
-import org.apache.storm.trident.windowing.WindowKryoSerializer;
-import org.apache.storm.trident.windowing.WindowsStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class stores entries into hbase instance of the given configuration.
- */
-public class HBaseWindowsStore implements WindowsStore {
- public static final String UTF_8 = "utf-8";
- private static final Logger LOG = LoggerFactory.getLogger(HBaseWindowsStore.class);
- private final ThreadLocal<Table> threadLocalHtable;
- private final ThreadLocal<WindowKryoSerializer> threadLocalWindowKryoSerializer;
- private final Queue<Table> htables = new ConcurrentLinkedQueue<>();
- private final byte[] family;
- private final byte[] qualifier;
-
- public HBaseWindowsStore(final Map<String, Object> topoConf, final Configuration config, final String tableName, byte[] family,
- byte[] qualifier) {
- this.family = family;
- this.qualifier = qualifier;
-
- threadLocalHtable = new ThreadLocal<Table>() {
- @Override
- protected Table initialValue() {
- try {
- Table table = ConnectionFactory.createConnection(config).getTable(TableName.valueOf(tableName));
- htables.add(table);
- return table;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
-
- threadLocalWindowKryoSerializer = new ThreadLocal<WindowKryoSerializer>() {
- @Override
- protected WindowKryoSerializer initialValue() {
- return new WindowKryoSerializer(topoConf);
- }
- };
-
- }
-
- private Table htable() {
- return threadLocalHtable.get();
- }
-
- private WindowKryoSerializer windowKryoSerializer() {
- return threadLocalWindowKryoSerializer.get();
- }
-
- private byte[] effectiveKey(String key) {
- try {
- return key.getBytes(UTF_8);
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public Object get(String key) {
- WindowsStore.Entry.nonNullCheckForKey(key);
-
- byte[] effectiveKey = effectiveKey(key);
- Get get = new Get(effectiveKey);
- Result result = null;
- try {
- result = htable().get(get);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- if (result.isEmpty()) {
- return null;
- }
-
- return windowKryoSerializer().deserialize(result.getValue(family, qualifier));
- }
-
- @Override
- public Iterable<Object> get(List<String> keys) {
- List<Get> gets = new ArrayList<>();
- for (String key : keys) {
- WindowsStore.Entry.nonNullCheckForKey(key);
-
- byte[] effectiveKey = effectiveKey(key);
- gets.add(new Get(effectiveKey));
- }
-
- Result[] results = null;
- try {
- results = htable().get(gets);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- List<Object> values = new ArrayList<>();
- for (int i = 0; i < results.length; i++) {
- Result result = results[i];
- if (result.isEmpty()) {
- LOG.error("Got empty result for key [{}]", keys.get(i));
- throw new RuntimeException("Received empty result for key: " + keys.get(i));
- }
- Object resultObject = windowKryoSerializer().deserialize(result.getValue(family, qualifier));
- values.add(resultObject);
- }
-
- return values;
- }
-
- @Override
- public Iterable<String> getAllKeys() {
- Scan scan = new Scan();
- // this filter makes sure to receive only Key or row but not values associated with those rows.
- scan.setFilter(new FirstKeyOnlyFilter());
- //scan.setCaching(1000);
-
- final Iterator<Result> resultIterator;
- try {
- resultIterator = htable().getScanner(scan).iterator();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- final Iterator<String> iterator = new Iterator<String>() {
- @Override
- public boolean hasNext() {
- return resultIterator.hasNext();
- }
-
- @Override
- public String next() {
- Result result = resultIterator.next();
- String key = null;
- try {
- key = new String(result.getRow(), UTF_8);
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- return key;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException("remove operation is not supported");
- }
- };
-
- return new Iterable<String>() {
- @Override
- public Iterator<String> iterator() {
- return iterator;
- }
- };
- }
-
- @Override
- public void put(String key, Object value) {
- WindowsStore.Entry.nonNullCheckForKey(key);
- WindowsStore.Entry.nonNullCheckForValue(value);
-
- if (value == null) {
- throw new IllegalArgumentException("Invalid value of null with key: " + key);
- }
- Put put = new Put(effectiveKey(key));
- put.addColumn(family, ByteBuffer.wrap(qualifier), System.currentTimeMillis(), windowKryoSerializer().serializeToByteBuffer(value));
- try {
- htable().put(put);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void putAll(Collection<Entry> entries) {
- List<Put> list = new ArrayList<>();
- for (Entry entry : entries) {
- Put put = new Put(effectiveKey(entry.key));
- put.addColumn(family, ByteBuffer.wrap(qualifier), System.currentTimeMillis(),
- windowKryoSerializer().serializeToByteBuffer(entry.value));
- list.add(put);
- }
-
- try {
- htable().put(list);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void remove(String key) {
- WindowsStore.Entry.nonNullCheckForKey(key);
-
- Delete delete = new Delete(effectiveKey(key), System.currentTimeMillis());
- try {
- htable().delete(delete);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void removeAll(Collection<String> keys) {
- List<Delete> deleteBatch = new ArrayList<>();
- for (String key : keys) {
- WindowsStore.Entry.nonNullCheckForKey(key);
-
- Delete delete = new Delete(effectiveKey(key), System.currentTimeMillis());
- deleteBatch.add(delete);
- }
- try {
- htable().delete(deleteBatch);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void shutdown() {
- // close all the created hTable instances
- for (Table htable : htables) {
- try {
- htable.close();
- } catch (IOException e) {
- LOG.error(e.getMessage(), e);
- }
- }
- }
-
-}
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStoreFactory.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStoreFactory.java
deleted file mode 100644
index c4030b204..000000000
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/windowing/HBaseWindowsStoreFactory.java
+++ /dev/null
@@ -1,48 +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.hbase.trident.windowing;
-
-import java.util.Map;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.storm.trident.windowing.WindowsStore;
-import org.apache.storm.trident.windowing.WindowsStoreFactory;
-
-/**
- * Factory to create {@link HBaseWindowsStore} instances.
- */
-public class HBaseWindowsStoreFactory implements WindowsStoreFactory {
- private final Map<String, Object> config;
- private final String tableName;
- private final byte[] family;
- private final byte[] qualifier;
-
- public HBaseWindowsStoreFactory(Map<String, Object> config, String tableName, byte[] family, byte[] qualifier) {
- this.config = config;
- this.tableName = tableName;
- this.family = family;
- this.qualifier = qualifier;
- }
-
- @Override
- public WindowsStore create(Map<String, Object> topoConf) {
- Configuration configuration = HBaseConfiguration.create();
- for (Map.Entry<String, Object> entry : config.entrySet()) {
- if (entry.getValue() != null) {
- configuration.set(entry.getKey(), entry.getValue().toString());
- }
- }
- return new HBaseWindowsStore(topoConf, configuration, tableName, family, qualifier);
- }
-
-}
diff --git a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseClientTestUtil.java b/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseClientTestUtil.java
deleted file mode 100644
index ab960d9c8..000000000
--- a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseClientTestUtil.java
+++ /dev/null
@@ -1,387 +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.hbase.state;
-
-import com.google.common.primitives.UnsignedBytes;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.NavigableSet;
-import java.util.TreeMap;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
-import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-
-public class HBaseClientTestUtil {
- private HBaseClientTestUtil() {
- }
-
- public static HBaseClient mockedHBaseClient() throws Exception {
- return mockedHBaseClient(new ConcurrentSkipListMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>>(
- UnsignedBytes.lexicographicalComparator()));
- }
-
- public static HBaseClient mockedHBaseClient(
- ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> internalMap)
- throws Exception {
- HBaseClient mockClient = mock(HBaseClient.class);
-
- Mockito.doNothing().when(mockClient).close();
-
- Mockito.when(mockClient.constructGetRequests(any(byte[].class), any(HBaseProjectionCriteria.class)))
- .thenCallRealMethod();
-
- Mockito.when(mockClient.constructMutationReq(any(byte[].class), any(ColumnList.class), any(Durability.class)))
- .thenCallRealMethod();
-
- Mockito.when(mockClient.exists(any(Get.class))).thenAnswer(new ExistsAnswer(internalMap));
- Mockito.when(mockClient.batchGet(any(List.class))).thenAnswer(new BatchGetAnswer(internalMap));
- Mockito.doAnswer(new BatchMutateAnswer(internalMap)).when(mockClient).batchMutate(any(List.class));
- Mockito.when(mockClient.scan(any(byte[].class), any(byte[].class))).thenAnswer(new ScanAnswer(internalMap));
-
- return mockClient;
- }
-
- static class BuildCellsHelper {
- public static void addMatchingColumnFamilies(byte[] rowKey, Map<byte[], NavigableSet<byte[]>> familyMap,
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValueMap,
- List<Cell> cells) {
- for (Map.Entry<byte[], NavigableSet<byte[]>> entry : familyMap.entrySet()) {
- byte[] columnFamily = entry.getKey();
-
- NavigableMap<byte[], byte[]> qualifierToValueMap = cfToQualifierToValueMap.get(columnFamily);
- if (qualifierToValueMap != null) {
- if (entry.getValue() == null || entry.getValue().size() == 0) {
- addAllQualifiers(rowKey, columnFamily, qualifierToValueMap, cells);
- } else {
- addMatchingQualifiers(rowKey, columnFamily, entry, qualifierToValueMap, cells);
- }
- }
- }
- }
-
- public static void addMatchingQualifiers(byte[] rowKey, byte[] columnFamily,
- Map.Entry<byte[], NavigableSet<byte[]>> qualifierSet,
- NavigableMap<byte[], byte[]> qualifierToValueMap,
- List<Cell> cells) {
- for (byte[] qualifier : qualifierSet.getValue()) {
- byte[] value = qualifierToValueMap.get(qualifier);
- if (value != null) {
- cells.add(new KeyValue(rowKey, columnFamily, qualifier, value));
- }
- }
- }
-
- public static void addAllColumnFamilies(byte[] rowKey, NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValueMap,
- List<Cell> cells) {
- for (Map.Entry<byte[], NavigableMap<byte[], byte[]>> entry : cfToQualifierToValueMap.entrySet()) {
- byte[] columnFamily = entry.getKey();
- addAllQualifiers(rowKey, columnFamily, entry.getValue(), cells);
- }
- }
-
- public static void addAllQualifiers(byte[] rowKey, byte[] columnFamily,
- NavigableMap<byte[], byte[]> qualifierToValueMap, List<Cell> cells) {
- for (Map.Entry<byte[], byte[]> entry2 : qualifierToValueMap.entrySet()) {
- byte[] qualifier = entry2.getKey();
- byte[] value = entry2.getValue();
- cells.add(new KeyValue(rowKey, columnFamily, qualifier, value));
- }
- }
-
- }
-
- static class BatchGetAnswer implements Answer<Result[]> {
- private final ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap;
-
- public BatchGetAnswer(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap) {
- this.mockMap = mockMap;
- }
-
- @Override
- public Result[] answer(InvocationOnMock invocationOnMock) throws Throwable {
- Object[] args = invocationOnMock.getArguments();
- List<Get> param = (List<Get>) args[0];
-
- List<Result> results = new ArrayList<>(param.size());
-
- for (Get get : param) {
- byte[] rowKey = get.getRow();
-
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValueMap =
- mockMap.get(rowKey);
-
- if (cfToQualifierToValueMap != null) {
- Map<byte[], NavigableSet<byte[]>> familyMap = get.getFamilyMap();
-
- List<Cell> cells = new ArrayList<>();
- if (familyMap == null || familyMap.size() == 0) {
- // all column families
- BuildCellsHelper.addAllColumnFamilies(rowKey, cfToQualifierToValueMap, cells);
- } else {
- // one or more column families
- BuildCellsHelper.addMatchingColumnFamilies(rowKey, familyMap, cfToQualifierToValueMap, cells);
- }
-
- // Result.create() states that "You must ensure that the keyvalues are already sorted."
- Collections.sort(cells, new KeyValue.KVComparator());
- results.add(Result.create(cells));
- } else {
- results.add(Result.EMPTY_RESULT);
- }
- }
-
- return results.toArray(new Result[0]);
- }
- }
-
- static class BatchMutateAnswer implements Answer {
- private final ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap;
-
- public BatchMutateAnswer(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap) {
- this.mockMap = mockMap;
- }
-
- @Override
- public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- Object[] args = invocationOnMock.getArguments();
- List<Mutation> param = (List<Mutation>) args[0];
-
- // assumption: there're no put and delete for same target in parameter list
- for (Mutation mutation : param) {
- byte[] rowKey = mutation.getRow();
-
- NavigableMap<byte[], List<Cell>> familyCellMap = mutation.getFamilyCellMap();
- if (familyCellMap == null || familyCellMap.size() == 0) {
- if (mutation instanceof Delete) {
- deleteRow(mockMap, rowKey);
- } else {
- throw new IllegalStateException("Not supported in mocked mutate.");
- }
- }
-
- for (Map.Entry<byte[], List<Cell>> entry : familyCellMap.entrySet()) {
- byte[] columnFamily = entry.getKey();
- List<Cell> cells = entry.getValue();
-
- if (cells == null || cells.size() == 0) {
- if (mutation instanceof Delete) {
- deleteColumnFamily(mockMap, rowKey, columnFamily);
- } else {
- throw new IllegalStateException("Not supported in mocked mutate.");
- }
- } else {
- for (Cell cell : cells) {
- byte[] qualifier = CellUtil.cloneQualifier(cell);
-
- if (mutation instanceof Put) {
- byte[] value = CellUtil.cloneValue(cell);
-
- putCell(mockMap, rowKey, columnFamily, qualifier, value);
- } else if (mutation instanceof Delete) {
- deleteCell(mockMap, rowKey, columnFamily, qualifier);
- } else {
- throw new IllegalStateException("Not supported in mocked mutate.");
- }
- }
- }
- }
- }
-
- return null;
- }
-
- private void putCell(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap,
- byte[] rowKey, byte[] columnFamily, byte[] qualifier, byte[] value) {
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValue = mockMap.get(rowKey);
- if (cfToQualifierToValue == null) {
- cfToQualifierToValue = new TreeMap<>(UnsignedBytes.lexicographicalComparator());
- mockMap.put(rowKey, cfToQualifierToValue);
- }
-
- NavigableMap<byte[], byte[]> qualifierToValue = cfToQualifierToValue.get(columnFamily);
- if (qualifierToValue == null) {
- qualifierToValue = new TreeMap<>(UnsignedBytes.lexicographicalComparator());
- cfToQualifierToValue.put(columnFamily, qualifierToValue);
- }
-
- qualifierToValue.put(qualifier, value);
- }
-
- private void deleteRow(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap,
- byte[] rowKey) {
- mockMap.remove(rowKey);
- }
-
- private void deleteColumnFamily(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap,
- byte[] rowKey, byte[] columnFamily) {
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValue = mockMap.get(rowKey);
- if (cfToQualifierToValue != null) {
- cfToQualifierToValue.remove(columnFamily);
- }
- }
-
- private void deleteCell(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap,
- byte[] rowKey, byte[] columnFamily, byte[] qualifier) {
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValue = mockMap.get(rowKey);
- if (cfToQualifierToValue != null) {
- NavigableMap<byte[], byte[]> qualifierToValue = cfToQualifierToValue.get(columnFamily);
- if (qualifierToValue != null) {
- qualifierToValue.remove(qualifier);
- }
- }
- }
- }
-
- static class ExistsAnswer implements Answer<Boolean> {
- private final ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap;
-
- public ExistsAnswer(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap) {
- this.mockMap = mockMap;
- }
-
- @Override
- public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
- Object[] args = invocationOnMock.getArguments();
- Get param = (Get) args[0];
-
- // assume that Get doesn't have any families defined. this is for not digging deeply...
- byte[] rowKey = param.getRow();
- Map<byte[], NavigableSet<byte[]>> familyMap = param.getFamilyMap();
- if (familyMap.size() > 0) {
- throw new IllegalStateException("Not supported in mocked exists.");
- }
-
- return mockMap.containsKey(rowKey);
- }
- }
-
- static class ScanAnswer implements Answer<ResultScanner> {
- private final ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap;
-
- public ScanAnswer(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> internalMap) {
- this.mockMap = internalMap;
- }
-
- @Override
- public ResultScanner answer(InvocationOnMock invocationOnMock) throws Throwable {
- Object[] args = invocationOnMock.getArguments();
- byte[] startKey = (byte[]) args[0];
- byte[] endKey = (byte[]) args[1];
-
- final ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> subMap =
- mockMap.subMap(startKey, true, endKey, false);
-
- final List<Result> results = buildResults(subMap);
-
- return new MockedResultScanner(results);
- }
-
- private List<Result> buildResults(ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> subMap) {
- final List<Result> results = new ArrayList<>();
- for (Map.Entry<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> entry : subMap.entrySet()) {
- byte[] rowKey = entry.getKey();
- NavigableMap<byte[], NavigableMap<byte[], byte[]>> cfToQualifierToValueMap = entry.getValue();
- List<Cell> cells = new ArrayList<>();
- // all column families
- BuildCellsHelper.addAllColumnFamilies(rowKey, cfToQualifierToValueMap, cells);
-
- // Result.create() states that "You must ensure that the keyvalues are already sorted."
- Collections.sort(cells, new KeyValue.KVComparator());
- results.add(Result.create(cells));
- }
- return results;
- }
-
- static class MockedResultScanner implements ResultScanner {
-
- private final List<Result> results;
- private int position = 0;
-
- MockedResultScanner(List<Result> results) {
- this.results = results;
- }
-
- @Override
- public Result next() {
- if (results.size() <= position) {
- return null;
- }
- return results.get(position++);
- }
-
- @Override
- public Result[] next(int nbRows) {
- List<Result> bulkResult = new ArrayList<>();
- for (int i = 0; i < nbRows; i++) {
- Result result = next();
- if (result == null) {
- break;
- }
-
- bulkResult.add(result);
- }
- return bulkResult.toArray(new Result[0]);
- }
-
- @Override
- public void close() {
- //NO-OP
- }
-
- @Override
- public boolean renewLease() {
- return true;
- }
-
- @Override
- public ScanMetrics getScanMetrics() {
- return null;
- }
-
- @Override
- public Iterator<Result> iterator() {
- return results.iterator();
- }
- }
- }
-}
diff --git a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateIteratorTest.java b/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateIteratorTest.java
deleted file mode 100644
index 96cca53ec..000000000
--- a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateIteratorTest.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * Copyright 2016 The Apache Software Foundation.
- *
- * Licensed 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.hbase.state;
-
-import com.google.common.primitives.UnsignedBytes;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.NavigableMap;
-import java.util.TreeMap;
-import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.Durability;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.storm.hbase.common.ColumnList;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.state.DefaultStateEncoder;
-import org.apache.storm.state.DefaultStateSerializer;
-import org.apache.storm.state.Serializer;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.apache.storm.hbase.state.HBaseKeyValueState.STATE_QUALIFIER;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Test for HBaseKeyValueStateIterator.
- */
-public class HBaseKeyValueStateIteratorTest {
-
- private String namespace;
- private byte[] keyNamespace;
- private byte[] columnFamily;
- private HBaseClient mockHBaseClient;
- private final int chunkSize = 1000;
- private final Serializer<byte[]> keySerializer = new DefaultStateSerializer<>();
- private final Serializer<byte[]> valueSerializer = new DefaultStateSerializer<>();
- private DefaultStateEncoder<byte[], byte[]> encoder;
-
- @BeforeEach
- public void setUp() throws Exception {
- namespace = "namespace";
- keyNamespace = (namespace + "$key:").getBytes();
- columnFamily = "cf".getBytes();
- mockHBaseClient = HBaseClientTestUtil.mockedHBaseClient();
- encoder = new DefaultStateEncoder<>(keySerializer, valueSerializer);
- }
-
- @Test
- public void testGetEntriesInHBase() throws Exception {
- // pendingPrepare has no entries
- final NavigableMap<byte[], byte[]> pendingPrepare = getBinaryTreeMap();
-
- // pendingCommit has no entries
- final NavigableMap<byte[], byte[]> pendingCommit = getBinaryTreeMap();
-
- // HBase has some entries
- NavigableMap<byte[], byte[]> chunkMap = getBinaryTreeMap();
- putEncodedKeyValueToMap(chunkMap, "key0".getBytes(), "value0".getBytes());
- putEncodedKeyValueToMap(chunkMap, "key2".getBytes(), "value2".getBytes());
-
- applyPendingStateToHBase(chunkMap);
-
- HBaseKeyValueStateIterator<byte[], byte[]> kvIterator =
- new HBaseKeyValueStateIterator<>(namespace, columnFamily, mockHBaseClient, pendingPrepare.entrySet().iterator(),
- pendingCommit.entrySet().iterator(), chunkSize, keySerializer, valueSerializer);
-
- assertNextEntry(kvIterator, "key0".getBytes(), "value0".getBytes());
-
- // key1 shouldn't in iterator
-
- assertNextEntry(kvIterator, "key2".getBytes(), "value2".getBytes());
-
- assertFalse(kvIterator.hasNext());
- }
-
- @Test
- public void testGetEntriesRemovingDuplicationKeys() throws Exception {
- NavigableMap<byte[], byte[]> pendingPrepare = getBinaryTreeMap();
- putEncodedKeyValueToMap(pendingPrepare, "key0".getBytes(), "value0".getBytes());
- putTombstoneToMap(pendingPrepare, "key1".getBytes());
-
- NavigableMap<byte[], byte[]> pendingCommit = getBinaryTreeMap();
- putEncodedKeyValueToMap(pendingCommit, "key1".getBytes(), "value1".getBytes());
- putEncodedKeyValueToMap(pendingCommit, "key2".getBytes(), "value2".getBytes());
-
- NavigableMap<byte[], byte[]> chunkMap = getBinaryTreeMap();
- putEncodedKeyValueToMap(chunkMap, "key2".getBytes(), "value2".getBytes());
- putEncodedKeyValueToMap(chunkMap, "key3".getBytes(), "value3".getBytes());
- putEncodedKeyValueToMap(chunkMap, "key4".getBytes(), "value4".getBytes());
-
- applyPendingStateToHBase(chunkMap);
-
- HBaseKeyValueStateIterator<byte[], byte[]> kvIterator =
- new HBaseKeyValueStateIterator<>(namespace, columnFamily, mockHBaseClient, pendingPrepare.entrySet().iterator(),
- pendingCommit.entrySet().iterator(), chunkSize, keySerializer, valueSerializer);
-
- // keys shouldn't appear twice
-
- assertNextEntry(kvIterator, "key0".getBytes(), "value0".getBytes());
-
- // key1 shouldn't be in iterator since it's marked as deleted
-
- assertNextEntry(kvIterator, "key2".getBytes(), "value2".getBytes());
- assertNextEntry(kvIterator, "key3".getBytes(), "value3".getBytes());
- assertNextEntry(kvIterator, "key4".getBytes(), "value4".getBytes());
-
- assertFalse(kvIterator.hasNext());
- }
-
- @Test
- public void testGetEntryNotAvailable() {
- NavigableMap<byte[], byte[]> pendingPrepare = getBinaryTreeMap();
-
- NavigableMap<byte[], byte[]> pendingCommit = getBinaryTreeMap();
-
- HBaseKeyValueStateIterator<byte[], byte[]> kvIterator =
- new HBaseKeyValueStateIterator<>(namespace, columnFamily, mockHBaseClient, pendingPrepare.entrySet().iterator(),
- pendingCommit.entrySet().iterator(), chunkSize, keySerializer, valueSerializer);
-
- assertFalse(kvIterator.hasNext());
- }
-
- private void assertNextEntry(HBaseKeyValueStateIterator<byte[], byte[]> kvIterator, byte[] expectedKey,
- byte[] expectedValue) {
- assertTrue(kvIterator.hasNext());
- Map.Entry<byte[], byte[]> entry = kvIterator.next();
- assertArrayEquals(expectedKey, entry.getKey());
- assertArrayEquals(expectedValue, entry.getValue());
- }
-
- private void putEncodedKeyValueToMap(NavigableMap<byte[], byte[]> map, byte[] key, byte[] value) {
- map.put(encoder.encodeKey(key), encoder.encodeValue(value));
- }
-
- private void putTombstoneToMap(NavigableMap<byte[], byte[]> map, byte[] key) {
- map.put(encoder.encodeKey(key), encoder.getTombstoneValue());
- }
-
- private TreeMap<byte[], byte[]> getBinaryTreeMap() {
- return new TreeMap<>(UnsignedBytes.lexicographicalComparator());
- }
-
- private void applyPendingStateToHBase(NavigableMap<byte[], byte[]> pendingMap) throws Exception {
- List<Mutation> mutations = new ArrayList<>();
- for (Map.Entry<byte[], byte[]> entry : pendingMap.entrySet()) {
- byte[] rowKey = entry.getKey();
- byte[] value = entry.getValue();
-
- if (Arrays.equals(value, encoder.getTombstoneValue())) {
- mutations.add(new Delete(getRowKeyForStateKey(rowKey)));
- } else {
- List<Mutation> mutationsForRow = prepareMutateRow(getRowKeyForStateKey(rowKey), columnFamily,
- Collections.singletonMap(STATE_QUALIFIER, value));
- mutations.addAll(mutationsForRow);
- }
- }
-
- mockHBaseClient.batchMutate(mutations);
- }
-
- private byte[] getRowKeyForStateKey(byte[] columnKey) {
- byte[] rowKey = new byte[keyNamespace.length + columnKey.length];
- System.arraycopy(keyNamespace, 0, rowKey, 0, keyNamespace.length);
- System.arraycopy(columnKey, 0, rowKey, keyNamespace.length, columnKey.length);
- return rowKey;
- }
-
- private ColumnList buildColumnList(byte[] columnFamily, Map<byte[], byte[]> map) {
- ColumnList columnList = new ColumnList();
- for (Map.Entry<byte[], byte[]> entry : map.entrySet()) {
- columnList.addColumn(columnFamily, entry.getKey(), entry.getValue());
- }
- return columnList;
- }
-
- private List<Mutation> prepareMutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map) {
- return prepareMutateRow(rowKey, columnFamily, map, Durability.USE_DEFAULT);
- }
-
- private List<Mutation> prepareMutateRow(byte[] rowKey, byte[] columnFamily, Map<byte[], byte[]> map,
- Durability durability) {
- ColumnList columnList = buildColumnList(columnFamily, map);
- return mockHBaseClient.constructMutationReq(rowKey, columnList, durability);
- }
-}
\ No newline at end of file
diff --git a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateProviderTest.java b/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateProviderTest.java
deleted file mode 100644
index f9b5977ad..000000000
--- a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateProviderTest.java
+++ /dev/null
@@ -1,86 +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.hbase.state;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Unit tests for {@link HBaseKeyValueStateProvider}
- */
-public class HBaseKeyValueStateProviderTest {
-
- @Test
- public void testConfigHBaseConfigKeyIsEmpty() {
- HBaseKeyValueStateProvider provider = new HBaseKeyValueStateProvider();
- Map<String, String> stormConf = new HashMap<>();
- stormConf.put(Config.TOPOLOGY_STATE_PROVIDER_CONFIG, "{\"keyClass\":\"String\", \"valueClass\":\"String\"," +
- " \"tableName\": \"table\", \"columnFamily\": \"cf\"}");
-
- IllegalArgumentException e =
- assertThrows(IllegalArgumentException.class, () -> provider.getStateConfig(stormConf));
- assertTrue(e.getMessage().contains("hbaseConfigKey"));
- }
-
- @Test
- public void testConfigTableNameIsEmpty() {
- HBaseKeyValueStateProvider provider = new HBaseKeyValueStateProvider();
- Map<String, String> stormConf = new HashMap<>();
- stormConf.put(Config.TOPOLOGY_STATE_PROVIDER_CONFIG, "{\"keyClass\":\"String\", \"valueClass\":\"String\"," +
- " \"hbaseConfigKey\": \"hbaseConfKey\", \"columnFamily\": \"cf\"}");
-
- IllegalArgumentException e =
- assertThrows(IllegalArgumentException.class, () -> provider.getStateConfig(stormConf));
- assertTrue(e.getMessage().contains("tableName"));
- }
-
- @Test
- public void testConfigColumnFamilyIsEmpty() {
- HBaseKeyValueStateProvider provider = new HBaseKeyValueStateProvider();
- Map<String, String> stormConf = new HashMap<>();
- stormConf.put(Config.TOPOLOGY_STATE_PROVIDER_CONFIG, "{\"keyClass\":\"String\", \"valueClass\":\"String\"," +
- " \"hbaseConfigKey\": \"hbaseConfKey\", \"tableName\": \"table\"}");
-
- IllegalArgumentException e =
- assertThrows(IllegalArgumentException.class, () -> provider.getStateConfig(stormConf));
- assertTrue(e.getMessage().contains("columnFamily"));
- }
-
- @Test
- public void testValidProviderConfig() throws Exception {
- HBaseKeyValueStateProvider provider = new HBaseKeyValueStateProvider();
- Map<String, String> stormConf = new HashMap<>();
- stormConf.put(Config.TOPOLOGY_STATE_PROVIDER_CONFIG, "{\"keyClass\":\"String\", \"valueClass\":\"String\"," +
- " \"hbaseConfigKey\": \"hbaseConfKey\", \"tableName\": \"table\"," +
- " \"columnFamily\": \"columnFamily\"}");
-
- HBaseKeyValueStateProvider.StateConfig config = provider.getStateConfig(stormConf);
- assertEquals("String", config.keyClass);
- assertEquals("String", config.valueClass);
- assertEquals("hbaseConfKey", config.hbaseConfigKey);
- assertEquals("table", config.tableName);
- assertEquals("columnFamily", config.columnFamily);
- }
-}
\ No newline at end of file
diff --git a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateTest.java b/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateTest.java
deleted file mode 100644
index 59de89a7f..000000000
--- a/external/storm-hbase/src/test/java/org/apache/storm/hbase/state/HBaseKeyValueStateTest.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
- *
- * 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.hbase.state;
-
-import com.google.common.primitives.UnsignedBytes;
-import java.util.NavigableMap;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
-import org.apache.storm.hbase.common.HBaseClient;
-import org.apache.storm.state.DefaultStateSerializer;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-/**
- * Unit tests for {@link HBaseKeyValueState}
- */
-public class HBaseKeyValueStateTest {
- private static final String COLUMN_FAMILY = "cf";
- private static final String NAMESPACE = "namespace";
-
- HBaseClient mockClient;
- HBaseKeyValueState<String, String> keyValueState;
- ConcurrentNavigableMap<byte[], NavigableMap<byte[], NavigableMap<byte[], byte[]>>> mockMap;
-
- @BeforeEach
- public void setUp() throws Exception {
- mockMap = new ConcurrentSkipListMap<>(UnsignedBytes.lexicographicalComparator());
- mockClient = HBaseClientTestUtil.mockedHBaseClient(mockMap);
- keyValueState = new HBaseKeyValueState<>(mockClient, COLUMN_FAMILY, NAMESPACE,
- new DefaultStateSerializer<>(), new DefaultStateSerializer<>());
- }
-
- @Test
- public void testPutAndGet() {
- keyValueState.put("a", "1");
- keyValueState.put("b", "2");
- assertEquals("1", keyValueState.get("a"));
- assertEquals("2", keyValueState.get("b"));
- assertNull(keyValueState.get("c"));
- }
-
- @Test
- public void testPutAndDelete() {
- keyValueState.put("a", "1");
- keyValueState.put("b", "2");
- assertEquals("1", keyValueState.get("a"));
- assertEquals("2", keyValueState.get("b"));
- assertNull(keyValueState.get("c"));
- assertEquals("1", keyValueState.delete("a"));
- assertNull(keyValueState.get("a"));
- assertEquals("2", keyValueState.get("b"));
- assertNull(keyValueState.get("c"));
- }
-
- @Test
- public void testPrepareCommitRollback() {
- keyValueState.put("a", "1");
- keyValueState.put("b", "2");
- keyValueState.prepareCommit(1);
- keyValueState.put("c", "3");
- assertArrayEquals(new String[]{ "1", "2", "3" }, getValues());
- keyValueState.rollback();
- assertArrayEquals(new String[]{ null, null, null }, getValues());
- keyValueState.put("a", "1");
- keyValueState.put("b", "2");
- keyValueState.prepareCommit(1);
- keyValueState.commit(1);
- keyValueState.put("c", "3");
- assertArrayEquals(new String[]{ "1", "2", "3" }, getValues());
- keyValueState.rollback();
- assertArrayEquals(new String[]{ "1", "2", null }, getValues());
- keyValueState.put("c", "3");
- assertEquals("2", keyValueState.delete("b"));
- assertEquals("3", keyValueState.delete("c"));
- assertArrayEquals(new String[]{ "1", null, null }, getValues());
- keyValueState.prepareCommit(2);
- assertArrayEquals(new String[]{ "1", null, null }, getValues());
- keyValueState.commit(2);
- assertArrayEquals(new String[]{ "1", null, null }, getValues());
- keyValueState.put("b", "2");
- keyValueState.prepareCommit(3);
- keyValueState.put("c", "3");
- assertArrayEquals(new String[]{ "1", "2", "3" }, getValues());
- keyValueState.rollback();
- assertArrayEquals(new String[]{ "1", null, null }, getValues());
- }
-
- private String[] getValues() {
- return new String[]{
- keyValueState.get("a"),
- keyValueState.get("b"),
- keyValueState.get("c")
- };
- }
-}
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 2e5799a08..160afc6ac 100644
--- a/pom.xml
+++ b/pom.xml
@@ -489,7 +489,6 @@
<module>external/storm-hdfs</module>
<module>external/storm-hdfs-blobstore</module>
<module>external/storm-hdfs-oci</module>
- <module>external/storm-hbase</module>
<module>external/storm-hive</module>
<module>external/storm-jdbc</module>
<module>external/storm-redis</module>
@@ -529,7 +528,6 @@
<module>examples/storm-kafka-client-examples</module>
<module>examples/storm-jdbc-examples</module>
<module>examples/storm-hdfs-examples</module>
- <module>examples/storm-hbase-examples</module>
<module>examples/storm-hive-examples</module>
<module>examples/storm-elasticsearch-examples</module>
<module>examples/storm-mqtt-examples</module>