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:41 UTC
[storm] 10/13: STORM-3988 - Remove "storm-rocketmq"
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 e655c6d88ebbdfd13d4c9c23dee5a698751fd2b7
Author: Richard Zowalla <ri...@hs-heilbronn.de>
AuthorDate: Thu Oct 19 09:00:22 2023 +0200
STORM-3988 - Remove "storm-rocketmq"
---
examples/storm-rocketmq-examples/pom.xml | 96 ---------
.../storm/rocketmq/topology/WordCountTopology.java | 89 --------
.../storm/rocketmq/topology/WordCounter.java | 68 ------
.../storm/rocketmq/trident/WordCountTrident.java | 89 --------
external/storm-rocketmq/README.md | 150 --------------
external/storm-rocketmq/pom.xml | 90 --------
.../storm/rocketmq/ConsumerBatchMessage.java | 65 ------
.../rocketmq/DefaultMessageBodySerializer.java | 38 ----
.../storm/rocketmq/MessageBodySerializer.java | 28 ---
.../org/apache/storm/rocketmq/RocketMqConfig.java | 178 ----------------
.../org/apache/storm/rocketmq/RocketMqUtils.java | 80 --------
.../org/apache/storm/rocketmq/SpoutConfig.java | 29 ---
.../apache/storm/rocketmq/bolt/RocketMqBolt.java | 216 -------------------
.../mapper/FieldNameBasedTupleToMessageMapper.java | 70 -------
.../common/mapper/TupleToMessageMapper.java | 32 ---
.../common/selector/DefaultTopicSelector.java | 45 ----
.../selector/FieldNameBasedTopicSelector.java | 70 -------
.../rocketmq/common/selector/TopicSelector.java | 29 ---
.../apache/storm/rocketmq/spout/RocketMqSpout.java | 228 ---------------------
.../rocketmq/spout/scheme/KeyValueScheme.java | 27 ---
.../spout/scheme/StringKeyValueScheme.java | 40 ----
.../storm/rocketmq/spout/scheme/StringScheme.java | 56 -----
.../rocketmq/trident/state/RocketMqState.java | 132 ------------
.../trident/state/RocketMqStateFactory.java | 43 ----
.../trident/state/RocketMqStateUpdater.java | 35 ----
.../rocketmq/DefaultMessageBodySerializerTest.java | 36 ----
.../java/org/apache/storm/rocketmq/TestUtils.java | 45 ----
.../storm/rocketmq/bolt/RocketMqBoltTest.java | 67 ------
.../FieldNameBasedTupleToMessageMapperTest.java | 36 ----
.../common/selector/DefaultTopicSelectorTest.java | 40 ----
.../selector/FieldNameBasedTopicSelectorTest.java | 40 ----
.../storm/rocketmq/spout/RocketMqSpoutTest.java | 87 --------
.../spout/scheme/StringKeyValueSchemeTest.java | 63 ------
pom.xml | 2 -
34 files changed, 2439 deletions(-)
diff --git a/examples/storm-rocketmq-examples/pom.xml b/examples/storm-rocketmq-examples/pom.xml
deleted file mode 100644
index c0ae8eb48..000000000
--- a/examples/storm-rocketmq-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-rocketmq-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-rocketmq</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-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCountTopology.java b/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCountTopology.java
deleted file mode 100644
index 3b40b51fc..000000000
--- a/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCountTopology.java
+++ /dev/null
@@ -1,89 +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.rocketmq.topology;
-
-import java.util.Properties;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.rocketmq.RocketMqConfig;
-import org.apache.storm.rocketmq.SpoutConfig;
-import org.apache.storm.rocketmq.bolt.RocketMqBolt;
-import org.apache.storm.rocketmq.common.mapper.FieldNameBasedTupleToMessageMapper;
-import org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper;
-import org.apache.storm.rocketmq.common.selector.DefaultTopicSelector;
-import org.apache.storm.rocketmq.common.selector.TopicSelector;
-import org.apache.storm.rocketmq.spout.RocketMqSpout;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-
-public class WordCountTopology {
- private static final String WORD_SPOUT = "WORD_SPOUT";
- private static final String COUNT_BOLT = "COUNT_BOLT";
- private static final String INSERT_BOLT = "INSERT_BOLT";
-
- private static final String CONSUMER_GROUP = "wordcount";
- private static final String CONSUMER_TOPIC = "source";
-
- public static StormTopology buildTopology(String nameserverAddr, String topic) {
- Properties properties = new Properties();
- properties.setProperty(SpoutConfig.NAME_SERVER_ADDR, nameserverAddr);
- properties.setProperty(SpoutConfig.CONSUMER_GROUP, CONSUMER_GROUP);
- properties.setProperty(SpoutConfig.CONSUMER_TOPIC, CONSUMER_TOPIC);
-
- RocketMqSpout spout = new RocketMqSpout(properties);
-
- TupleToMessageMapper mapper = new FieldNameBasedTupleToMessageMapper("word", "count");
- TopicSelector selector = new DefaultTopicSelector(topic);
-
- properties = new Properties();
- properties.setProperty(RocketMqConfig.NAME_SERVER_ADDR, nameserverAddr);
-
- RocketMqBolt insertBolt = new RocketMqBolt()
- .withMapper(mapper)
- .withSelector(selector)
- .withProperties(properties);
-
- // wordSpout ==> countBolt ==> insertBolt
- TopologyBuilder builder = new TopologyBuilder();
-
- WordCounter bolt = new WordCounter();
- builder.setSpout(WORD_SPOUT, spout, 1);
- builder.setBolt(COUNT_BOLT, bolt, 1).fieldsGrouping(WORD_SPOUT, new Fields("str"));
- builder.setBolt(INSERT_BOLT, insertBolt, 1).shuffleGrouping(COUNT_BOLT);
-
- return builder.createTopology();
- }
-
- public static void main(String[] args) throws Exception {
- Config conf = new Config();
- conf.setMaxSpoutPending(5);
- conf.setNumWorkers(3);
-
- String topologyName = "wordCounter";
- if (args.length < 2) {
- System.out.println("Usage: WordCountTopology <nameserver addr> <topic> [topology name]");
- } else {
- if (args.length > 3) {
- topologyName = args[2];
- }
- StormSubmitter.submitTopology(topologyName, conf, buildTopology(args[0], args[1]));
- }
- }
-}
diff --git a/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCounter.java b/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCounter.java
deleted file mode 100644
index dd6820036..000000000
--- a/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/topology/WordCounter.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.rocketmq.topology;
-
-import java.util.HashMap;
-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;
-import org.apache.storm.tuple.Values;
-
-public class WordCounter implements IBasicBolt {
- private Map<String, Integer> wordCounter = new HashMap<>();
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context) {
-
- }
-
- @Override
- public void execute(Tuple input, BasicOutputCollector collector) {
- String word = input.getStringByField("str");
- int count;
- if (wordCounter.containsKey(word)) {
- count = wordCounter.get(word) + 1;
- } else {
- count = 1;
- }
- wordCounter.put(word, count);
- collector.emit(new Values(word, count));
- }
-
- @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-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/trident/WordCountTrident.java b/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/trident/WordCountTrident.java
deleted file mode 100644
index dbdee9355..000000000
--- a/examples/storm-rocketmq-examples/src/main/java/org/apache/storm/rocketmq/trident/WordCountTrident.java
+++ /dev/null
@@ -1,89 +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.rocketmq.trident;
-
-import java.util.Properties;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.rocketmq.RocketMqConfig;
-import org.apache.storm.rocketmq.common.mapper.FieldNameBasedTupleToMessageMapper;
-import org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper;
-import org.apache.storm.rocketmq.common.selector.DefaultTopicSelector;
-import org.apache.storm.rocketmq.common.selector.TopicSelector;
-import org.apache.storm.rocketmq.trident.state.RocketMqState;
-import org.apache.storm.rocketmq.trident.state.RocketMqStateFactory;
-import org.apache.storm.rocketmq.trident.state.RocketMqStateUpdater;
-import org.apache.storm.trident.Stream;
-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 nameserverAddr, String topic) {
- 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);
-
- TupleToMessageMapper mapper = new FieldNameBasedTupleToMessageMapper("word", "count");
- TopicSelector selector = new DefaultTopicSelector(topic);
-
- Properties properties = new Properties();
- properties.setProperty(RocketMqConfig.NAME_SERVER_ADDR, nameserverAddr);
-
- RocketMqState.Options options = new RocketMqState.Options()
- .withMapper(mapper)
- .withSelector(selector)
- .withProperties(properties);
-
- StateFactory factory = new RocketMqStateFactory(options);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- stream.partitionPersist(factory, fields,
- new RocketMqStateUpdater(), new Fields());
-
- return topology.build();
- }
-
- public static void main(String[] args) throws Exception {
- Config conf = new Config();
- conf.setMaxSpoutPending(5);
- conf.setNumWorkers(3);
-
- String topologyName = "wordCounter";
- if (args.length < 2) {
- System.out.println("Usage: WordCountTrident <nameserver addr> <topic> [topology name]");
- } else {
- if (args.length > 3) {
- topologyName = args[2];
- }
- StormSubmitter.submitTopology(topologyName, conf, buildTopology(args[0], args[1]));
- }
- }
-}
diff --git a/external/storm-rocketmq/README.md b/external/storm-rocketmq/README.md
deleted file mode 100644
index 4934cf0e8..000000000
--- a/external/storm-rocketmq/README.md
+++ /dev/null
@@ -1,150 +0,0 @@
-# Storm RocketMQ
-
-Storm/Trident integration for [RocketMQ](https://rocketmq.apache.org/). This package includes the core spout, bolt and trident states that allows a storm topology to either write storm tuples into a topic or read from topics in a storm topology.
-
-
-## Read from Topic
-The spout included in this package for reading data from a topic.
-
-### RocketMqSpout
-To use the `RocketMqSpout`, you construct an instance of it by specifying a Properties instance which including rocketmq configs.
-RocketMqSpout uses RocketMQ MQPushConsumer as the default implementation. PushConsumer is a high level consumer API, wrapping the pulling details. Looks like broker push messages to consumer.
-RocketMqSpout's messages retrying depends on RocketMQ's push mode retry policy.
-
- ```java
- Properties properties = new Properties();
- properties.setProperty(SpoutConfig.NAME_SERVER_ADDR, nameserverAddr);
- properties.setProperty(SpoutConfig.CONSUMER_GROUP, group);
- properties.setProperty(SpoutConfig.CONSUMER_TOPIC, topic);
-
- RocketMqSpout spout = new RocketMqSpout(properties);
- ```
-
-
-## Write into Topic
-The bolt and trident state included in this package for write data into a topic.
-
-### TupleToMessageMapper
-The main API for mapping Storm tuple to a RocketMQ Message is the `org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper` interface:
-
-```java
-public interface TupleToMessageMapper extends Serializable {
- String getKeyFromTuple(ITuple tuple);
- byte[] getValueFromTuple(ITuple tuple);
-}
-```
-
-### FieldNameBasedTupleToMessageMapper
-`storm-rocketmq` includes a general purpose `TupleToMessageMapper` implementation called `FieldNameBasedTupleToMessageMapper`.
-
-### TopicSelector
-The main API for selecting topic and tags is the `org.apache.storm.rocketmq.common.selector.TopicSelector` interface:
-
-```java
-public interface TopicSelector extends Serializable {
- String getTopic(ITuple tuple);
- String getTag(ITuple tuple);
-}
-```
-
-### DefaultTopicSelector/FieldNameBasedTopicSelector
-`storm-rocketmq` includes general purpose `TopicSelector` implementations called `DefaultTopicSelector` and `FieldNameBasedTopicSelector`.
-
-
-### RocketMqBolt
-To use the `RocketMqBolt`, you construct an instance of it by specifying TupleToMessageMapper, TopicSelector and Properties instances.
-RocketMqBolt send messages async by default. You can change this by invoking `withAsync(false)`.
-
- ```java
- TupleToMessageMapper mapper = new FieldNameBasedTupleToMessageMapper("word", "count");
- TopicSelector selector = new DefaultTopicSelector(topic);
-
- properties = new Properties();
- properties.setProperty(RocketMqConfig.NAME_SERVER_ADDR, nameserverAddr);
-
- RocketMqBolt insertBolt = new RocketMqBolt()
- .withMapper(mapper)
- .withSelector(selector)
- .withProperties(properties);
- ```
-
-### Trident State
-We support trident persistent state that can be used with trident topologies. To create a RocketMQ persistent trident state you need to initialize it with the TupleToMessageMapper, TopicSelector, Properties instances. See the example below:
-
- ```java
- TupleToMessageMapper mapper = new FieldNameBasedTupleToMessageMapper("word", "count");
- TopicSelector selector = new DefaultTopicSelector(topic);
-
- Properties properties = new Properties();
- properties.setProperty(RocketMqConfig.NAME_SERVER_ADDR, nameserverAddr);
-
- RocketMqState.Options options = new RocketMqState.Options()
- .withMapper(mapper)
- .withSelector(selector)
- .withProperties(properties);
-
- StateFactory factory = new RocketMqStateFactory(options);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- stream.partitionPersist(factory, fields,
- new RocketMqStateUpdater(), new Fields());
- ```
-
-## Configurations
-
-### Producer Configurations
-| NAME | DESCRIPTION | DEFAULT |
-| ------------- |:-------------:|:------:|
-| nameserver.address | name server address *Required* | null |
-| nameserver.poll.interval | name server poll topic info interval | 30000 |
-| brokerserver.heartbeat.interval | broker server heartbeat interval | 30000 |
-| producer.group | producer group | $UUID |
-| producer.retry.times | producer send messages retry times | 3 |
-| producer.timeout | producer send messages timeout | 3000 |
-
-
-### Consumer Configurations
-| NAME | DESCRIPTION | DEFAULT |
-| ------------- |:-------------:|:------:|
-| nameserver.address | name server address *Required* | null |
-| nameserver.poll.interval | name server poll topic info interval | 30000 |
-| brokerserver.heartbeat.interval | broker server heartbeat interval | 30000 |
-| consumer.group | consumer group *Required* | null |
-| consumer.topic | consumer topic *Required* | null |
-| consumer.tag | consumer topic tag | * |
-| consumer.offset.reset.to | what to do when there is no initial offset on the server | latest/earliest/timestamp |
-| consumer.offset.from.timestamp | the timestamp when `consumer.offset.reset.to=timestamp` was set | $TIMESTAMP |
-| consumer.messages.orderly | if the consumer topic is ordered | false |
-| consumer.offset.persist.interval | auto commit offset interval | 5000 |
-| consumer.min.threads | consumer min threads | 20 |
-| consumer.max.threads | consumer max threads | 64 |
-| consumer.callback.executor.threads | client callback executor threads | $availableProcessors |
-| consumer.batch.size | consumer messages batch size | 32 |
-| consumer.batch.process.timeout | consumer messages batch process timeout | $TOPOLOGY_MESSAGE_TIMEOUT_SECS + 10s|
-
-
-## 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
-
- * Xin Wang ([xinwang@apache.org](mailto:xinwang@apache.org))
-
diff --git a/external/storm-rocketmq/pom.xml b/external/storm-rocketmq/pom.xml
deleted file mode 100644
index 969ac1d30..000000000
--- a/external/storm-rocketmq/pom.xml
+++ /dev/null
@@ -1,90 +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-rocketmq</artifactId>
- <name>storm-rocketmq</name>
-
- <packaging>jar</packaging>
-
- <developers>
- <developer>
- <id>vesense</id>
- <name>Xin Wang</name>
- <email>xinwang@apache.org</email>
- </developer>
- </developers>
-
- <dependencies>
- <!--parent module dependency-->
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <scope>${provided.scope}</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-client</artifactId>
- <version>${rocketmq.version}</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </dependency>
- <dependency>
- <groupId>commons-lang</groupId>
- <artifactId>commons-lang</artifactId>
- </dependency>
- <!--test dependencies -->
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-server</artifactId>
- <version>${project.version}</version>
- <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-rocketmq/src/main/java/org/apache/storm/rocketmq/ConsumerBatchMessage.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/ConsumerBatchMessage.java
deleted file mode 100644
index c32a18a35..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/ConsumerBatchMessage.java
+++ /dev/null
@@ -1,65 +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.rocketmq;
-
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-public class ConsumerBatchMessage<T> {
- private final List<T> data;
- private CountDownLatch latch;
- private boolean hasFailure = false;
-
- public ConsumerBatchMessage(List<T> data) {
- this.data = data;
- latch = new CountDownLatch(data.size());
- }
-
- public boolean waitFinish(long timeout) throws InterruptedException {
- return latch.await(timeout, TimeUnit.MILLISECONDS);
- }
-
- public boolean isSuccess() {
- return !hasFailure;
- }
-
- public List<T> getData() {
- return data;
- }
-
- /**
- * Countdown if the sub message is successful.
- */
- public void ack() {
- latch.countDown();
- }
-
- /**
- * Countdown and fail-fast if the sub message is failed.
- */
- public void fail() {
- hasFailure = true;
- // fail fast
- long count = latch.getCount();
- for (int i = 0; i < count; i++) {
- latch.countDown();
- }
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/DefaultMessageBodySerializer.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/DefaultMessageBodySerializer.java
deleted file mode 100644
index 7cfaac6c0..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/DefaultMessageBodySerializer.java
+++ /dev/null
@@ -1,38 +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.rocketmq;
-
-import java.nio.charset.StandardCharsets;
-
-public class DefaultMessageBodySerializer implements MessageBodySerializer {
-
- /**
- * Currently, we just convert string to bytes using UTF-8 charset.
- * Note: in this way, object.toString() method is invoked.
- * @param body RocketMQ Message body
- * @return serialized byte array
- */
- @Override
- public byte[] serialize(Object body) {
- if (body == null) {
- return null;
- }
- return body.toString().getBytes(StandardCharsets.UTF_8);
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/MessageBodySerializer.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/MessageBodySerializer.java
deleted file mode 100644
index fbf16cd6c..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/MessageBodySerializer.java
+++ /dev/null
@@ -1,28 +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.rocketmq;
-
-import java.io.Serializable;
-
-/**
- * RocketMQ message body serializer.
- */
-public interface MessageBodySerializer extends Serializable {
- byte[] serialize(Object body);
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqConfig.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqConfig.java
deleted file mode 100644
index 9a239beae..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqConfig.java
+++ /dev/null
@@ -1,178 +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.rocketmq;
-
-import static org.apache.storm.rocketmq.RocketMqUtils.getInteger;
-
-import java.util.Properties;
-import java.util.UUID;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.Validate;
-import org.apache.rocketmq.client.ClientConfig;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-
-/**
- * RocketMqConfig for Consumer/Producer.
- */
-public class RocketMqConfig {
- // common
- public static final String NAME_SERVER_ADDR = "nameserver.address"; // Required
-
- public static final String NAME_SERVER_POLL_INTERVAL = "nameserver.poll.interval";
- public static final int DEFAULT_NAME_SERVER_POLL_INTERVAL = 30000; // 30 seconds
-
- public static final String BROKER_HEART_BEAT_INTERVAL = "brokerserver.heartbeat.interval";
- public static final int DEFAULT_BROKER_HEART_BEAT_INTERVAL = 30000; // 30 seconds
-
-
- // producer
- public static final String PRODUCER_GROUP = "producer.group";
-
- public static final String PRODUCER_RETRY_TIMES = "producer.retry.times";
- public static final int DEFAULT_PRODUCER_RETRY_TIMES = 3;
-
- public static final String PRODUCER_TIMEOUT = "producer.timeout";
- public static final int DEFAULT_PRODUCER_TIMEOUT = 3000; // 3 seconds
-
-
- // consumer
- public static final String CONSUMER_GROUP = "consumer.group"; // Required
-
- public static final String CONSUMER_TOPIC = "consumer.topic"; // Required
-
- public static final String CONSUMER_TAG = "consumer.tag";
- public static final String DEFAULT_CONSUMER_TAG = "*";
-
- public static final String CONSUMER_OFFSET_RESET_TO = "consumer.offset.reset.to";
- public static final String CONSUMER_OFFSET_LATEST = "latest";
- public static final String CONSUMER_OFFSET_EARLIEST = "earliest";
- public static final String CONSUMER_OFFSET_TIMESTAMP = "timestamp";
- public static final String CONSUMER_OFFSET_FROM_TIMESTAMP = "consumer.offset.from.timestamp";
-
- public static final String CONSUMER_MESSAGES_ORDERLY = "consumer.messages.orderly";
-
- public static final String CONSUMER_OFFSET_PERSIST_INTERVAL = "consumer.offset.persist.interval";
- public static final int DEFAULT_CONSUMER_OFFSET_PERSIST_INTERVAL = 5000; // 5 seconds
-
- public static final String CONSUMER_MIN_THREADS = "consumer.min.threads";
- public static final int DEFAULT_CONSUMER_MIN_THREADS = 20;
-
- public static final String CONSUMER_MAX_THREADS = "consumer.max.threads";
- public static final int DEFAULT_CONSUMER_MAX_THREADS = 64;
-
- public static final String CONSUMER_CALLBACK_EXECUTOR_THREADS = "consumer.callback.executor.threads";
- public static final int DEFAULT_CONSUMER_CALLBACK_EXECUTOR_THREADS = Runtime.getRuntime().availableProcessors();
-
- public static final String CONSUMER_BATCH_SIZE = "consumer.batch.size";
- public static final int DEFAULT_CONSUMER_BATCH_SIZE = 32;
-
- public static final String CONSUMER_BATCH_PROCESS_TIMEOUT = "consumer.batch.process.timeout";
-
- /**
- * Build Producer Configs.
- * @param props Properties
- * @param producer DefaultMQProducer
- */
- public static void buildProducerConfigs(Properties props, DefaultMQProducer producer) {
- buildCommonConfigs(props, producer);
-
- String group = props.getProperty(PRODUCER_GROUP);
- if (StringUtils.isEmpty(group)) {
- group = UUID.randomUUID().toString();
- }
- producer.setProducerGroup(props.getProperty(PRODUCER_GROUP, group));
-
- producer.setRetryTimesWhenSendFailed(getInteger(props,
- PRODUCER_RETRY_TIMES, DEFAULT_PRODUCER_RETRY_TIMES));
- producer.setRetryTimesWhenSendAsyncFailed(getInteger(props,
- PRODUCER_RETRY_TIMES, DEFAULT_PRODUCER_RETRY_TIMES));
- producer.setSendMsgTimeout(getInteger(props,
- PRODUCER_TIMEOUT, DEFAULT_PRODUCER_TIMEOUT));
- }
-
- /**
- * Build Consumer Configs.
- * @param props Properties
- * @param consumer DefaultMQPushConsumer
- */
- public static void buildConsumerConfigs(Properties props, DefaultMQPushConsumer consumer) {
- buildCommonConfigs(props, consumer);
-
- String group = props.getProperty(CONSUMER_GROUP);
- Validate.notEmpty(group);
- consumer.setConsumerGroup(group);
-
- consumer.setPullBatchSize(getInteger(props,
- CONSUMER_BATCH_SIZE, DEFAULT_CONSUMER_BATCH_SIZE));
-
- consumer.setPersistConsumerOffsetInterval(getInteger(props,
- CONSUMER_OFFSET_PERSIST_INTERVAL, DEFAULT_CONSUMER_OFFSET_PERSIST_INTERVAL));
- consumer.setConsumeThreadMin(getInteger(props,
- CONSUMER_MIN_THREADS, DEFAULT_CONSUMER_MIN_THREADS));
- consumer.setConsumeThreadMax(getInteger(props,
- CONSUMER_MAX_THREADS, DEFAULT_CONSUMER_MAX_THREADS));
-
- consumer.setClientCallbackExecutorThreads(getInteger(props,
- CONSUMER_CALLBACK_EXECUTOR_THREADS, DEFAULT_CONSUMER_CALLBACK_EXECUTOR_THREADS));
-
- String initOffset = props.getProperty(CONSUMER_OFFSET_RESET_TO, CONSUMER_OFFSET_LATEST);
- switch (initOffset) {
- case CONSUMER_OFFSET_EARLIEST:
- consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
- break;
- case CONSUMER_OFFSET_LATEST:
- consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
- break;
- case CONSUMER_OFFSET_TIMESTAMP:
- String timestamp = props.getProperty(CONSUMER_OFFSET_FROM_TIMESTAMP);
- consumer.setConsumeTimestamp(timestamp);
- break;
- default:
- consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
- }
-
- String topic = props.getProperty(CONSUMER_TOPIC);
- Validate.notEmpty(topic);
- try {
- consumer.subscribe(topic, props.getProperty(CONSUMER_TAG, DEFAULT_CONSUMER_TAG));
- } catch (MQClientException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- /**
- * Build Common Configs.
- * @param props Properties
- * @param client ClientConfig
- */
- public static void buildCommonConfigs(Properties props, ClientConfig client) {
- String nameServers = props.getProperty(NAME_SERVER_ADDR);
- Validate.notEmpty(nameServers);
- client.setNamesrvAddr(nameServers);
-
- client.setPollNameServerInterval(getInteger(props,
- NAME_SERVER_POLL_INTERVAL, DEFAULT_NAME_SERVER_POLL_INTERVAL));
- client.setHeartbeatBrokerInterval(getInteger(props,
- BROKER_HEART_BEAT_INTERVAL, DEFAULT_BROKER_HEART_BEAT_INTERVAL));
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqUtils.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqUtils.java
deleted file mode 100644
index a2d7e7a6d..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMqUtils.java
+++ /dev/null
@@ -1,80 +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.rocketmq;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.Properties;
-
-import org.apache.rocketmq.common.message.Message;
-import org.apache.storm.rocketmq.spout.scheme.KeyValueScheme;
-import org.apache.storm.spout.Scheme;
-
-public final class RocketMqUtils {
-
- public static int getInteger(Properties props, String key, int defaultValue) {
- return Integer.parseInt(props.getProperty(key, String.valueOf(defaultValue)));
- }
-
- public static long getLong(Properties props, String key, long defaultValue) {
- return Long.parseLong(props.getProperty(key, String.valueOf(defaultValue)));
- }
-
- public static boolean getBoolean(Properties props, String key, boolean defaultValue) {
- return Boolean.parseBoolean(props.getProperty(key, String.valueOf(defaultValue)));
- }
-
- /**
- * Create Scheme by Properties.
- * @param props Properties
- * @return Scheme
- */
- public static Scheme createScheme(Properties props) {
- String schemeString = props.getProperty(SpoutConfig.SCHEME, SpoutConfig.DEFAULT_SCHEME);
- Scheme scheme;
- try {
- Class clazz = Class.forName(schemeString);
- scheme = (Scheme) clazz.newInstance();
- } catch (Exception e) {
- throw new IllegalArgumentException("Cannot create Scheme for " + schemeString
- + " due to " + e.getMessage());
- }
- return scheme;
- }
-
- /**
- * Generate Storm tuple values by Message and Scheme.
- * @param msg RocketMQ Message
- * @param scheme Scheme for deserializing
- * @return tuple values
- */
- public static List<Object> generateTuples(Message msg, Scheme scheme) {
- List<Object> tup;
- String rawKey = msg.getKeys();
- ByteBuffer body = ByteBuffer.wrap(msg.getBody());
- if (rawKey != null && scheme instanceof KeyValueScheme) {
- ByteBuffer key = ByteBuffer.wrap(rawKey.getBytes(StandardCharsets.UTF_8));
- tup = ((KeyValueScheme) scheme).deserializeKeyAndValue(key, body);
- } else {
- tup = scheme.deserialize(body);
- }
- return tup;
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/SpoutConfig.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/SpoutConfig.java
deleted file mode 100644
index 83e956d53..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/SpoutConfig.java
+++ /dev/null
@@ -1,29 +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.rocketmq;
-
-import org.apache.storm.rocketmq.spout.scheme.StringScheme;
-
-public class SpoutConfig extends RocketMqConfig {
-
- public static final String SCHEME = "spout.scheme";
- public static final String DEFAULT_SCHEME = StringScheme.class.getName();
-
-
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/bolt/RocketMqBolt.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/bolt/RocketMqBolt.java
deleted file mode 100644
index 4cfa9f1d9..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/bolt/RocketMqBolt.java
+++ /dev/null
@@ -1,216 +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.rocketmq.bolt;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import org.apache.commons.lang.Validate;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendCallback;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.storm.Config;
-import org.apache.storm.rocketmq.RocketMqConfig;
-import org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper;
-import org.apache.storm.rocketmq.common.selector.TopicSelector;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.IRichBolt;
-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;
-
-public class RocketMqBolt implements IRichBolt {
- private static final long serialVersionUID = 1L;
- private static final Logger LOG = LoggerFactory.getLogger(RocketMqBolt.class);
-
- private static final int DEFAULT_FLUSH_INTERVAL_SECS = 5;
- private static final int DEFAULT_BATCH_SIZE = 20;
-
- private DefaultMQProducer producer;
- private OutputCollector collector;
- private TopicSelector selector;
- private TupleToMessageMapper mapper;
- private Properties properties;
-
- private boolean async = true;
- private boolean batch = false;
- private int batchSize = DEFAULT_BATCH_SIZE;
- private int flushIntervalSecs = DEFAULT_FLUSH_INTERVAL_SECS;
- private BatchHelper batchHelper;
- private List<Message> messages;
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
- Validate.notEmpty(properties, "Producer properties can not be empty");
- Validate.notNull(selector, "TopicSelector can not be null");
- Validate.notNull(mapper, "TupleToMessageMapper can not be null");
-
- producer = new DefaultMQProducer();
- producer.setInstanceName(String.valueOf(context.getThisTaskId()));
- RocketMqConfig.buildProducerConfigs(properties, producer);
-
- try {
- producer.start();
- } catch (MQClientException e) {
- throw new RuntimeException(e);
- }
-
- this.collector = collector;
- this.batchHelper = new BatchHelper(batchSize, collector);
- this.messages = new LinkedList<>();
- }
-
- public RocketMqBolt withSelector(TopicSelector selector) {
- this.selector = selector;
- return this;
- }
-
- public RocketMqBolt withMapper(TupleToMessageMapper mapper) {
- this.mapper = mapper;
- return this;
- }
-
- public RocketMqBolt withAsync(boolean async) {
- this.async = async;
- return this;
- }
-
- public RocketMqBolt withBatch(boolean batch) {
- this.batch = batch;
- return this;
- }
-
- public RocketMqBolt withBatchSize(int batchSize) {
- this.batchSize = batchSize;
- return this;
- }
-
- public RocketMqBolt withFlushIntervalSecs(int flushIntervalSecs) {
- this.flushIntervalSecs = flushIntervalSecs;
- return this;
- }
-
- public RocketMqBolt withProperties(Properties properties) {
- this.properties = properties;
- return this;
- }
-
- @Override
- public void execute(Tuple input) {
- if (!batch && TupleUtils.isTick(input)) {
- return;
- }
-
- String topic = selector.getTopic(input);
- if (topic == null) {
- LOG.warn("skipping Message due to topic selector returned null.");
- collector.ack(input);
- return;
- }
-
- if (batch) {
- // batch sync sending
- try {
- if (batchHelper.shouldHandle(input)) {
- batchHelper.addBatch(input);
- messages.add(prepareMessage(input));
- }
-
- if (batchHelper.shouldFlush()) {
- producer.send(messages);
- batchHelper.ack();
- messages.clear();
- }
- } catch (Exception e) {
- LOG.error("Batch send messages failure!", e);
- batchHelper.fail(e);
- messages.clear();
- }
- } else {
- if (async) {
- // async sending
- try {
- producer.send(prepareMessage(input), new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- collector.ack(input);
- }
-
- @Override
- public void onException(Throwable throwable) {
- if (throwable != null) {
- LOG.error("Async send messages failure!", throwable);
- collector.reportError(throwable);
- collector.fail(input);
- }
- }
- });
- } catch (Exception e) {
- LOG.error("Async send messages failure!", e);
- collector.reportError(e);
- collector.fail(input);
- }
- } else {
- // sync sending, will return a SendResult
- try {
- producer.send(prepareMessage(input));
- collector.ack(input);
- } catch (Exception e) {
- LOG.error("Sync send messages failure!", e);
- collector.reportError(e);
- collector.fail(input);
- }
- }
- }
- }
-
- // Mapping: from storm tuple -> rocketmq Message
- private Message prepareMessage(Tuple input) {
- String topic = selector.getTopic(input);
- String tag = selector.getTag(input);
- String key = mapper.getKeyFromTuple(input);
- byte[] value = mapper.getValueFromTuple(input);
-
- return new Message(topic, tag, key, value);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
-
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return TupleUtils.putTickFrequencyIntoComponentConfig(new Config(), flushIntervalSecs);
- }
-
- @Override
- public void cleanup() {
- if (producer != null) {
- producer.shutdown();
- }
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapper.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapper.java
deleted file mode 100644
index a38a36539..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapper.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.rocketmq.common.mapper;
-
-import org.apache.storm.rocketmq.DefaultMessageBodySerializer;
-import org.apache.storm.rocketmq.MessageBodySerializer;
-import org.apache.storm.tuple.ITuple;
-
-public class FieldNameBasedTupleToMessageMapper implements TupleToMessageMapper {
- public static final String BOLT_KEY = "key";
- public static final String BOLT_MESSAGE = "message";
- public String boltKeyField;
- public String boltMessageField;
- private MessageBodySerializer messageBodySerializer;
-
- public FieldNameBasedTupleToMessageMapper() {
- this(BOLT_KEY, BOLT_MESSAGE);
- }
-
- /**
- * FieldNameBasedTupleToMessageMapper Constructor.
- * @param boltKeyField tuple field for selecting the key
- * @param boltMessageField tuple field for selecting the value
- */
- public FieldNameBasedTupleToMessageMapper(String boltKeyField, String boltMessageField) {
- this.boltKeyField = boltKeyField;
- this.boltMessageField = boltMessageField;
- this.messageBodySerializer = new DefaultMessageBodySerializer();
- }
-
- @Override
- public String getKeyFromTuple(ITuple tuple) {
- return tuple.getStringByField(boltKeyField);
- }
-
- @Override
- public byte[] getValueFromTuple(ITuple tuple) {
- Object obj = tuple.getValueByField(boltMessageField);
- if (obj == null) {
- return null;
- }
- return messageBodySerializer.serialize(obj);
- }
-
- /**
- * using this method can override the default MessageBodySerializer.
- * @param serializer MessageBodySerializer
- * @return this object
- */
- public FieldNameBasedTupleToMessageMapper withMessageBodySerializer(MessageBodySerializer serializer) {
- this.messageBodySerializer = serializer;
- return this;
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/TupleToMessageMapper.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/TupleToMessageMapper.java
deleted file mode 100644
index ef716a4c4..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/mapper/TupleToMessageMapper.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.rocketmq.common.mapper;
-
-import java.io.Serializable;
-import org.apache.storm.tuple.ITuple;
-
-/**
- * Interface defining a mapping from storm tuple to rocketmq key and message.
- */
-public interface TupleToMessageMapper extends Serializable {
-
- String getKeyFromTuple(ITuple tuple);
-
- byte[] getValueFromTuple(ITuple tuple);
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelector.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelector.java
deleted file mode 100644
index e0192da68..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelector.java
+++ /dev/null
@@ -1,45 +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.rocketmq.common.selector;
-
-import org.apache.storm.tuple.ITuple;
-
-public class DefaultTopicSelector implements TopicSelector {
- private final String topicName;
- private final String tagName;
-
- public DefaultTopicSelector(final String topicName, final String tagName) {
- this.topicName = topicName;
- this.tagName = tagName;
- }
-
- public DefaultTopicSelector(final String topicName) {
- this(topicName, "");
- }
-
- @Override
- public String getTopic(ITuple tuple) {
- return topicName;
- }
-
- @Override
- public String getTag(ITuple tuple) {
- return tagName;
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelector.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelector.java
deleted file mode 100644
index 253221ee5..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelector.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.storm.rocketmq.common.selector;
-
-import org.apache.storm.tuple.ITuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Uses field name to select topic and tag name from tuple.
- */
-public class FieldNameBasedTopicSelector implements TopicSelector {
- private static final Logger LOG = LoggerFactory.getLogger(FieldNameBasedTopicSelector.class);
-
- private final String topicFieldName;
- private final String defaultTopicName;
-
- private final String tagFieldName;
- private final String defaultTagName;
-
- /**
- * FieldNameBasedTopicSelector Constructor.
- * @param topicFieldName field name used for selecting topic
- * @param defaultTopicName default field name used for selecting topic
- * @param tagFieldName field name used for selecting tag
- * @param defaultTagName default field name used for selecting tag
- */
- public FieldNameBasedTopicSelector(String topicFieldName, String defaultTopicName, String tagFieldName, String defaultTagName) {
- this.topicFieldName = topicFieldName;
- this.defaultTopicName = defaultTopicName;
- this.tagFieldName = tagFieldName;
- this.defaultTagName = defaultTagName;
- }
-
- @Override
- public String getTopic(ITuple tuple) {
- if (tuple.contains(topicFieldName)) {
- return tuple.getStringByField(topicFieldName);
- } else {
- LOG.warn("Field {} Not Found. Returning default topic {}", topicFieldName, defaultTopicName);
- return defaultTopicName;
- }
- }
-
- @Override
- public String getTag(ITuple tuple) {
- if (tuple.contains(tagFieldName)) {
- return tuple.getStringByField(tagFieldName);
- } else {
- LOG.warn("Field {} Not Found. Returning default tag {}", tagFieldName, defaultTagName);
- return defaultTagName;
- }
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/TopicSelector.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/TopicSelector.java
deleted file mode 100644
index 2940920fe..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/common/selector/TopicSelector.java
+++ /dev/null
@@ -1,29 +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.rocketmq.common.selector;
-
-import java.io.Serializable;
-import org.apache.storm.tuple.ITuple;
-
-public interface TopicSelector extends Serializable {
-
- String getTopic(ITuple tuple);
-
- String getTag(ITuple tuple);
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMqSpout.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMqSpout.java
deleted file mode 100644
index c08dd6263..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMqSpout.java
+++ /dev/null
@@ -1,228 +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.rocketmq.spout;
-
-import static org.apache.storm.rocketmq.RocketMqUtils.getBoolean;
-import static org.apache.storm.rocketmq.RocketMqUtils.getLong;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.commons.lang.Validate;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.storm.Config;
-import org.apache.storm.rocketmq.ConsumerBatchMessage;
-import org.apache.storm.rocketmq.RocketMqConfig;
-import org.apache.storm.rocketmq.RocketMqUtils;
-import org.apache.storm.spout.Scheme;
-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.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * RocketMqSpout uses MQPushConsumer as the default implementation.
- * PushConsumer is a high level consumer API, wrapping the pulling details
- * Looks like broker push messages to consumer
- */
-public class RocketMqSpout implements IRichSpout {
- // TODO add metrics
- private static final long serialVersionUID = 1L;
- private static final Logger LOG = LoggerFactory.getLogger(RocketMqSpout.class);
-
- private DefaultMQPushConsumer consumer;
- private SpoutOutputCollector collector;
- private BlockingQueue<ConsumerBatchMessage<List<Object>>> queue;
- private Map<String, ConsumerBatchMessage<List<Object>>> cache;
-
- private Properties properties;
- private Scheme scheme;
- private long batchProcessTimeout;
-
- /**
- * RocketMqSpout Constructor.
- * @param properties Properties Config
- */
- public RocketMqSpout(Properties properties) {
- Validate.notEmpty(properties, "Consumer properties can not be empty");
- this.properties = properties;
- scheme = RocketMqUtils.createScheme(properties);
- }
-
- @Override
- public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
- consumer = new DefaultMQPushConsumer();
- consumer.setInstanceName(String.valueOf(context.getThisTaskId()));
- RocketMqConfig.buildConsumerConfigs(properties, consumer);
-
- boolean ordered = getBoolean(properties, RocketMqConfig.CONSUMER_MESSAGES_ORDERLY, false);
- if (ordered) {
- consumer.registerMessageListener(new MessageListenerOrderly() {
- @Override
- public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
- ConsumeOrderlyContext context) {
- if (process(msgs)) {
- return ConsumeOrderlyStatus.SUCCESS;
- } else {
- return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
- }
- }
- });
- } else {
- consumer.registerMessageListener(new MessageListenerConcurrently() {
- @Override
- public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
- ConsumeConcurrentlyContext context) {
- if (process(msgs)) {
- return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
- } else {
- return ConsumeConcurrentlyStatus.RECONSUME_LATER;
- }
- }
- });
- }
-
- try {
- consumer.start();
- } catch (MQClientException e) {
- LOG.error("Failed to start RocketMQ consumer.", e);
- throw new RuntimeException(e);
- }
-
- long defaultBatchProcessTimeout = (long) conf.getOrDefault(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 30) * 1000 + 10000;
- batchProcessTimeout = getLong(properties, RocketMqConfig.CONSUMER_BATCH_PROCESS_TIMEOUT, defaultBatchProcessTimeout);
-
- queue = new LinkedBlockingQueue<>();
- cache = new ConcurrentHashMap<>();
- this.collector = collector;
- }
-
- /**
- * Process pushed messages.
- * @param msgs messages
- * @return the boolean flag processed result
- */
- protected boolean process(List<MessageExt> msgs) {
- if (msgs.isEmpty()) {
- return true;
- }
-
- List<List<Object>> list = new ArrayList<>(msgs.size());
- for (MessageExt msg : msgs) {
- List<Object> data = RocketMqUtils.generateTuples(msg, scheme);
- if (data != null) {
- list.add(data);
- }
- }
- ConsumerBatchMessage<List<Object>> batchMessage = new ConsumerBatchMessage<>(list);
- try {
- queue.put(batchMessage);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
-
- boolean isCompleted;
- try {
- isCompleted = batchMessage.waitFinish(batchProcessTimeout);
- } catch (InterruptedException e) {
- LOG.error("Interrupted when waiting messages to be finished.", e);
- throw new RuntimeException(e);
- }
-
- boolean isSuccess = batchMessage.isSuccess();
-
- return isCompleted && isSuccess;
- }
-
- @Override
- public void nextTuple() {
- ConsumerBatchMessage<List<Object>> batchMessage = queue.poll();
- if (batchMessage != null) {
- List<List<Object>> list = batchMessage.getData();
- for (List<Object> data : list) {
- String messageId = UUID.randomUUID().toString();
- cache.put(messageId, batchMessage);
- collector.emit(data, messageId);
- }
- }
- }
-
- @Override
- public void ack(Object msgId) {
- ConsumerBatchMessage batchMessage = cache.get(msgId);
- batchMessage.ack();
- cache.remove(msgId);
- LOG.debug("Message acked {}", batchMessage);
- }
-
- @Override
- public void fail(Object msgId) {
- ConsumerBatchMessage batchMessage = cache.get(msgId);
- batchMessage.fail();
- cache.remove(msgId);
- LOG.debug("Message failed {}", batchMessage);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(scheme.getOutputFields());
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
- @Override
- public void close() {
- if (consumer != null) {
- consumer.shutdown();
- }
- }
-
- @Override
- public void activate() {
- if (consumer != null) {
- consumer.resume();
- }
- }
-
- @Override
- public void deactivate() {
- if (consumer != null) {
- consumer.suspend();
- }
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/KeyValueScheme.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/KeyValueScheme.java
deleted file mode 100644
index f4cf803bb..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/KeyValueScheme.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.rocketmq.spout.scheme;
-
-import java.nio.ByteBuffer;
-import java.util.List;
-import org.apache.storm.spout.Scheme;
-
-public interface KeyValueScheme extends Scheme {
- List<Object> deserializeKeyAndValue(ByteBuffer key, ByteBuffer value);
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueScheme.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueScheme.java
deleted file mode 100644
index 21a762e1f..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueScheme.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.rocketmq.spout.scheme;
-
-import com.google.common.collect.ImmutableMap;
-
-import java.nio.ByteBuffer;
-import java.util.List;
-
-import org.apache.storm.tuple.Values;
-
-public class StringKeyValueScheme extends StringScheme implements KeyValueScheme {
-
- @Override
- public List<Object> deserializeKeyAndValue(ByteBuffer key, ByteBuffer value) {
- if (key == null) {
- return deserialize(value);
- }
- String keyString = StringScheme.deserializeString(key);
- String valueString = StringScheme.deserializeString(value);
- return new Values(ImmutableMap.of(keyString, valueString));
- }
-
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringScheme.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringScheme.java
deleted file mode 100644
index 53fdc57aa..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/scheme/StringScheme.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.rocketmq.spout.scheme;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import org.apache.storm.spout.Scheme;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.Utils;
-
-public class StringScheme implements Scheme {
- public static final String STRING_SCHEME_KEY = "str";
-
- @Override
- public List<Object> deserialize(ByteBuffer bytes) {
- return new Values(deserializeString(bytes));
- }
-
- /**
- * Deserialize ByteBuffer to String.
- * @param byteBuffer input ByteBuffer
- * @return deserialized string
- */
- public static String deserializeString(ByteBuffer byteBuffer) {
- if (byteBuffer.hasArray()) {
- int base = byteBuffer.arrayOffset();
- return new String(byteBuffer.array(), base + byteBuffer.position(), byteBuffer.remaining(),
- StandardCharsets.UTF_8);
- } else {
- return new String(Utils.toByteArray(byteBuffer), StandardCharsets.UTF_8);
- }
- }
-
- @Override
- public Fields getOutputFields() {
- return new Fields(STRING_SCHEME_KEY);
- }
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqState.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqState.java
deleted file mode 100644
index c42bd4711..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqState.java
+++ /dev/null
@@ -1,132 +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.rocketmq.trident.state;
-
-import java.io.Serializable;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-import org.apache.commons.lang.Validate;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.storm.rocketmq.RocketMqConfig;
-import org.apache.storm.rocketmq.common.mapper.TupleToMessageMapper;
-import org.apache.storm.rocketmq.common.selector.TopicSelector;
-import org.apache.storm.topology.FailedException;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.state.State;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RocketMqState implements State {
-
- private static final Logger LOG = LoggerFactory.getLogger(RocketMqState.class);
-
- private Options options;
- private DefaultMQProducer producer;
-
- protected RocketMqState(Map<String, Object> map, Options options) {
- this.options = options;
- }
-
- public static class Options implements Serializable {
- private TopicSelector selector;
- private TupleToMessageMapper mapper;
- private Properties properties;
-
- public Options withSelector(TopicSelector selector) {
- this.selector = selector;
- return this;
- }
-
- public Options withMapper(TupleToMessageMapper mapper) {
- this.mapper = mapper;
- return this;
- }
-
- public Options withProperties(Properties properties) {
- this.properties = properties;
- return this;
- }
- }
-
- protected void prepare() {
- Validate.notEmpty(options.properties, "Producer properties can not be empty");
- Validate.notNull(options.selector, "TopicSelector can not be null");
- Validate.notNull(options.mapper, "TupleToMessageMapper can not be null");
-
- producer = new DefaultMQProducer();
- producer.setInstanceName(UUID.randomUUID().toString());
- RocketMqConfig.buildProducerConfigs(options.properties, producer);
-
- try {
- producer.start();
- } catch (MQClientException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void beginCommit(Long txid) {
- LOG.debug("beginCommit is noop.");
- }
-
- @Override
- public void commit(Long txid) {
- LOG.debug("commit is noop.");
- }
-
- /**
- * Update the RocketMQ state.
- * @param tuples trident tuples
- * @param collector trident collector
- */
- public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
- List<Message> messages = new LinkedList<>();
-
- for (TridentTuple tuple : tuples) {
- String topic = options.selector.getTopic(tuple);
- String tag = options.selector.getTag(tuple);
- String key = options.mapper.getKeyFromTuple(tuple);
- byte[] value = options.mapper.getValueFromTuple(tuple);
-
- if (topic == null) {
- LOG.warn("skipping Message with Key = " + key + ", topic selector returned null.");
- continue;
- }
-
- Message msg = new Message(topic, tag, key, value);
- messages.add(msg);
- }
-
- try {
- this.producer.send(messages);
- } catch (Exception e) {
- LOG.warn("Batch write failed. Triggering replay.", e);
- collector.reportError(e);
- throw new FailedException(e);
- }
- }
-
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateFactory.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateFactory.java
deleted file mode 100644
index 64b8277f4..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateFactory.java
+++ /dev/null
@@ -1,43 +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.rocketmq.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 RocketMqStateFactory implements StateFactory {
-
- private RocketMqState.Options options;
-
- public RocketMqStateFactory(RocketMqState.Options options) {
- this.options = options;
- }
-
- @Override
- public State makeState(Map<String, Object> conf, IMetricsContext metrics,
- int partitionIndex, int numPartitions) {
- RocketMqState state = new RocketMqState(conf, options);
- state.prepare();
- return state;
- }
-
-}
diff --git a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateUpdater.java b/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateUpdater.java
deleted file mode 100644
index 589ac5e3f..000000000
--- a/external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/trident/state/RocketMqStateUpdater.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.rocketmq.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 RocketMqStateUpdater extends BaseStateUpdater<RocketMqState> {
-
- @Override
- public void updateState(RocketMqState state, List<TridentTuple> tuples,
- TridentCollector collector) {
- state.updateState(tuples, collector);
- }
-
-}
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/DefaultMessageBodySerializerTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/DefaultMessageBodySerializerTest.java
deleted file mode 100644
index dd8ed06f3..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/DefaultMessageBodySerializerTest.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.rocketmq;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-
-public class DefaultMessageBodySerializerTest {
- @Test
- public void serialize() {
- DefaultMessageBodySerializer messageBodySerializer = new DefaultMessageBodySerializer();
- String body = "this is message body data";
- assertArrayEquals(body.getBytes(), messageBodySerializer.serialize(body));
-
- body = null;
- assertArrayEquals(null, messageBodySerializer.serialize(body));
- }
-
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/TestUtils.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/TestUtils.java
deleted file mode 100644
index 399445594..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/TestUtils.java
+++ /dev/null
@@ -1,45 +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.rocketmq;
-
-import java.lang.reflect.Field;
-import java.util.Arrays;
-
-import org.apache.storm.Testing;
-import org.apache.storm.testing.MkTupleParam;
-import org.apache.storm.tuple.Tuple;
-
-public class TestUtils {
- public static void setFieldValue(Object obj, String fieldName, Object value) {
- try {
- Field field = obj.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- field.set(obj, value);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- public static Tuple generateTestTuple(String field1, String field2, Object value1, Object value2) {
- MkTupleParam param = new MkTupleParam();
- param.setFields(field1, field2);
- Tuple testTuple = Testing.testTuple(Arrays.asList(value1, value2), param);
- return testTuple;
- }
-}
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/bolt/RocketMqBoltTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/bolt/RocketMqBoltTest.java
deleted file mode 100644
index 76b016764..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/bolt/RocketMqBoltTest.java
+++ /dev/null
@@ -1,67 +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.rocketmq.bolt;
-
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendCallback;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.storm.rocketmq.TestUtils;
-import org.apache.storm.rocketmq.common.mapper.FieldNameBasedTupleToMessageMapper;
-import org.apache.storm.rocketmq.common.selector.DefaultTopicSelector;
-import org.apache.storm.tuple.Tuple;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-
-public class RocketMqBoltTest {
-
- private RocketMqBolt rocketMqBolt;
- private DefaultMQProducer producer;
-
- @BeforeEach
- public void setUp() {
- rocketMqBolt = new RocketMqBolt();
- rocketMqBolt.withSelector(new DefaultTopicSelector("tpc"));
- rocketMqBolt.withMapper(new FieldNameBasedTupleToMessageMapper("f1", "f2"));
- rocketMqBolt.withBatch(false);
- rocketMqBolt.withBatchSize(5);
- rocketMqBolt.withAsync(true);
- rocketMqBolt.withFlushIntervalSecs(5);
-
- producer = mock(DefaultMQProducer.class);
- TestUtils.setFieldValue(rocketMqBolt, "producer", producer);
- }
-
- @Test
- public void execute() throws Exception {
- Tuple tuple = TestUtils.generateTestTuple("f1", "f2", "v1", "v2");
- rocketMqBolt.execute(tuple);
-
- verify(producer).send(any(Message.class), any(SendCallback.class));
- }
-
- @Test
- public void cleanup() {
- rocketMqBolt.cleanup();
- verify(producer).shutdown();
- }
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapperTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapperTest.java
deleted file mode 100644
index 885eb5ca9..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/mapper/FieldNameBasedTupleToMessageMapperTest.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.rocketmq.common.mapper;
-
-import org.apache.storm.rocketmq.TestUtils;
-import org.apache.storm.tuple.Tuple;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class FieldNameBasedTupleToMessageMapperTest {
- @Test
- public void withMessageBodySerializer() throws Exception {
- FieldNameBasedTupleToMessageMapper messageMapper = new FieldNameBasedTupleToMessageMapper("f1", "f2");
- Tuple tuple = TestUtils.generateTestTuple("f1", "f2", "v1", "v2");
- assertEquals("v1", messageMapper.getKeyFromTuple(tuple));
- assertArrayEquals("v2".getBytes(), messageMapper.getValueFromTuple(tuple));
- }
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelectorTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelectorTest.java
deleted file mode 100644
index 841d5cd9f..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/DefaultTopicSelectorTest.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.rocketmq.common.selector;
-
-import org.apache.storm.tuple.Tuple;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class DefaultTopicSelectorTest {
-
- @Test
- public void getTopic() {
- DefaultTopicSelector topicSelector = new DefaultTopicSelector("tpc", "tg");
- Tuple tuple = null;
- assertEquals("tpc", topicSelector.getTopic(tuple));
- assertEquals("tg", topicSelector.getTag(tuple));
-
- topicSelector = new DefaultTopicSelector("tpc2");
- assertEquals("tpc2", topicSelector.getTopic(tuple));
- assertEquals("", topicSelector.getTag(tuple));
- }
-
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelectorTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelectorTest.java
deleted file mode 100644
index 0d2365688..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/common/selector/FieldNameBasedTopicSelectorTest.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.rocketmq.common.selector;
-
-import org.apache.storm.rocketmq.TestUtils;
-import org.apache.storm.tuple.Tuple;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class FieldNameBasedTopicSelectorTest {
- @Test
- public void getTopic() {
- FieldNameBasedTopicSelector topicSelector = new FieldNameBasedTopicSelector("f1", "tpc", "f2", "tg");
- Tuple tuple = TestUtils.generateTestTuple("f1", "fn", "v1", "vn");
- assertEquals("v1", topicSelector.getTopic(tuple));
- assertEquals("tg", topicSelector.getTag(tuple));
-
- tuple = TestUtils.generateTestTuple("fn", "f2", "vn", "v2");
- assertEquals("tpc", topicSelector.getTopic(tuple));
- assertEquals("v2", topicSelector.getTag(tuple));
- }
-
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/RocketMqSpoutTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/RocketMqSpoutTest.java
deleted file mode 100644
index f036be134..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/RocketMqSpoutTest.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.rocketmq.spout;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.storm.rocketmq.ConsumerBatchMessage;
-import org.apache.storm.rocketmq.RocketMqUtils;
-import org.apache.storm.rocketmq.SpoutConfig;
-import org.apache.storm.rocketmq.TestUtils;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-
-public class RocketMqSpoutTest {
- private RocketMqSpout spout;
- private DefaultMQPushConsumer consumer;
- private SpoutOutputCollector collector;
- private BlockingQueue<ConsumerBatchMessage> queue;
- private Properties properties;
-
- @BeforeEach
- public void setUp() {
- properties = new Properties();
- properties.setProperty(SpoutConfig.NAME_SERVER_ADDR, "address");
- properties.setProperty(SpoutConfig.CONSUMER_GROUP, "group");
- properties.setProperty(SpoutConfig.CONSUMER_TOPIC, "topic");
-
- spout = new RocketMqSpout(properties);
- consumer = mock(DefaultMQPushConsumer.class);
- TestUtils.setFieldValue(spout, "consumer", consumer);
- collector = mock(SpoutOutputCollector.class);
- TestUtils.setFieldValue(spout, "collector", collector);
-
- queue = new LinkedBlockingQueue<>();
- TestUtils.setFieldValue(spout, "queue", queue);
-
- Map<String, ConsumerBatchMessage> cache = mock(Map.class);
- TestUtils.setFieldValue(spout, "cache", cache);
- }
-
- @Test
- public void nextTuple() throws Exception {
- List<List<Object>> list = new ArrayList<>();
- list.add(RocketMqUtils.generateTuples(new Message("tpc", "body".getBytes()),
- RocketMqUtils.createScheme(properties)));
- ConsumerBatchMessage<List<Object>> batchMessage = new ConsumerBatchMessage<>(list);
- queue.put(batchMessage);
-
- spout.nextTuple();
- verify(collector).emit(anyList(), anyString());
- }
-
- @Test
- public void close() {
- spout.close();
- verify(consumer).shutdown();
- }
-}
\ No newline at end of file
diff --git a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueSchemeTest.java b/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueSchemeTest.java
deleted file mode 100644
index dbb59a51c..000000000
--- a/external/storm-rocketmq/src/test/java/org/apache/storm/rocketmq/spout/scheme/StringKeyValueSchemeTest.java
+++ /dev/null
@@ -1,63 +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.rocketmq.spout.scheme;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
-import java.util.Collections;
-
-import com.google.common.collect.ImmutableMap;
-import org.apache.storm.tuple.Fields;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class StringKeyValueSchemeTest {
-
- private StringKeyValueScheme scheme = new StringKeyValueScheme();
-
- @Test
- public void testDeserialize() throws Exception {
- assertEquals(Collections.singletonList("test"), scheme.deserialize(wrapString("test")));
- }
-
- @Test
- public void testGetOutputFields() throws Exception {
- Fields outputFields = scheme.getOutputFields();
- assertTrue(outputFields.contains(StringScheme.STRING_SCHEME_KEY));
- assertEquals(1, outputFields.size());
- }
-
- @Test
- public void testDeserializeWithNullKeyAndValue() throws Exception {
- assertEquals(Collections.singletonList("test"),
- scheme.deserializeKeyAndValue(null, wrapString("test")));
- }
-
- @Test
- public void testDeserializeWithKeyAndValue() throws Exception {
- assertEquals(Collections.singletonList(ImmutableMap.of("key", "test")),
- scheme.deserializeKeyAndValue(wrapString("key"), wrapString("test")));
- }
-
- private static ByteBuffer wrapString(String s) {
- return ByteBuffer.wrap(s.getBytes(Charset.defaultCharset()));
- }
-}
diff --git a/pom.xml b/pom.xml
index 583e94607..efedc0ed1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -499,7 +499,6 @@
<module>external/storm-kafka-migration</module>
<module>external/storm-kafka-monitor</module>
<module>external/storm-jms</module>
- <module>external/storm-rocketmq</module>
<module>external/storm-blobstore-migration</module>
<module>integration-test</module>
@@ -524,7 +523,6 @@
<module>examples/storm-hive-examples</module>
<module>examples/storm-elasticsearch-examples</module>
<module>examples/storm-jms-examples</module>
- <module>examples/storm-rocketmq-examples</module>
<module>examples/storm-perf</module>
</modules>
</profile>