You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by vesense <gi...@git.apache.org> on 2017/03/22 06:58:10 UTC
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
GitHub user vesense opened a pull request:
https://github.com/apache/storm/pull/2024
STORM-2349: Add one RocketMQ plugin for the Apache Storm
https://issues.apache.org/jira/browse/STORM-2349
This is the init version for code review.
Current phase(fetatures included in this PR):
* RocketMQ Bolt
* RocketMQ Spout
* RocketMQ Trident State
* documents and examples
Local tests passed.
Next phase(in the plan):
RocketMQ Trident Spout.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/vesense/storm STORM-2349
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2024.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2024
----
commit eff76dc5210d957e4d793cd983c53a878e49533b
Author: vesense <be...@163.com>
Date: 2017-03-20T08:50:34Z
STORM-2349: Add one RocketMQ plugin for the Apache Storm
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r110601839
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/RocketMQConfig.java ---
@@ -0,0 +1,172 @@
+/**
+ * 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.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;
+import org.apache.rocketmq.remoting.common.RemotingUtil;
+import org.apache.storm.task.TopologyContext;
+
+import java.util.Properties;
+import java.util.UUID;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * RocketMQConfig for Consumer/Producer
+ */
+public class RocketMQConfig {
+ // common
+ public static final String NAME_SERVER_ADDR = "nameserver.addr"; // Required
+
+ public static final String CLIENT_NAME = "client.name";
+
+ public static final String CLIENT_IP = "client.ip";
+ public static final String DEFAULT_CLIENT_IP = RemotingUtil.getLocalAddress();
+
+ public static final String CLIENT_CALLBACK_EXECUTOR_THREADS = "client.callback.executor.threads";
+ public static final int DEFAULT_CLIENT_CALLBACK_EXECUTOR_THREADS = Runtime.getRuntime().availableProcessors();;
+
+ 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 = 2;
+
+ 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_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_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 void buildProducerConfigs(Properties props, DefaultMQProducer producer, TopologyContext context) {
+ buildCommonConfigs(props, producer, context);
+
+ // According to the RocketMQ official docs, "only one instance is allowed per producer group"
+ // So, we use taskID/UUID as the producer group by default
+ String defaultGroup;
+ if (context != null) {
+ defaultGroup = String.valueOf(context.getThisTaskId());
+ } else {
+ defaultGroup = UUID.randomUUID().toString();
+ }
+ producer.setProducerGroup(props.getProperty(PRODUCER_GROUP, defaultGroup));
+
+ 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));
+ }
+
+ public static void buildConsumerConfigs(Properties props, DefaultMQPushConsumer consumer, TopologyContext context) {
+ buildCommonConfigs(props, consumer, context);
+
+ String group = props.getProperty(CONSUMER_GROUP);
+ Validate.notEmpty(group);
+ consumer.setConsumerGroup(group);
+
+ 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));
+
+ 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:
+ consumer.setConsumeTimestamp(initOffset);
+ 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_TAG));
+ } catch (MQClientException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ public static void buildCommonConfigs(Properties props, ClientConfig client, TopologyContext context) {
+ String namesvr = props.getProperty(NAME_SERVER_ADDR);
+ Validate.notEmpty(namesvr);
+ client.setNamesrvAddr(namesvr);
+
+ client.setClientIP(props.getProperty(CLIENT_IP, DEFAULT_CLIENT_IP));
+ // use taskID/UUID for client name by default
+ String defaultClientName;
+ if (context != null) {
+ defaultClientName = String.valueOf(context.getThisTaskId());
--- End diff --
Cool~
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111345560
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,177 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Integer.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING).toString()));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
--- End diff --
Currently, only push mode is supported which is auto-commit. We can add pull mode in the next stage.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111867114
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ 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) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ int maxRetry = getInteger(properties, SpoutConfig.MESSAGES_MAX_RETRY, SpoutConfig.DEFAULT_MESSAGES_MAX_RETRY);
+ int ttl = getInteger(properties, SpoutConfig.MESSAGES_TTL, SpoutConfig.DEFAULT_MESSAGES_TTL);
+ this.messageRetryManager = new DefaultMessageRetryManager(queue, maxRetry, ttl);
+ this.collector = collector;
+ }
+
+ public boolean process(List<MessageExt> msgs) {
+ if (msgs.isEmpty()) {
+ return true;
+ }
+ MessageSet messageSet = new MessageSet(msgs);
--- End diff --
Updated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111346264
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,177 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Integer.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING).toString()));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
--- End diff --
@hustfxj Thanks for your reminding.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111524719
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ 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) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ int maxRetry = getInteger(properties, SpoutConfig.MESSAGES_MAX_RETRY, SpoutConfig.DEFAULT_MESSAGES_MAX_RETRY);
+ int ttl = getInteger(properties, SpoutConfig.MESSAGES_TTL, SpoutConfig.DEFAULT_MESSAGES_TTL);
+ this.messageRetryManager = new DefaultMessageRetryManager(queue, maxRetry, ttl);
+ this.collector = collector;
+ }
+
+ public boolean process(List<MessageExt> msgs) {
+ if (msgs.isEmpty()) {
+ return true;
+ }
+ MessageSet messageSet = new MessageSet(msgs);
--- End diff --
any plans of offering a TupleMapper to flatten the schema or you want to preserve the same data in rocketmq and send it downstream
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r109309759
--- Diff: external/storm-rocketmq/pom.xml ---
@@ -0,0 +1,89 @@
+<?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.0.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-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.0.0-incubating</version>
+ </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>
+ <version>${mockito.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>test</scope>
--- End diff --
Does the latest storm core still dependency log4j\uff1f
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111533311
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ if (ordered) {
+ consumer.registerMessageListener(new MessageListenerOrderly() {
--- End diff --
yes
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111551156
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ 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) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ int maxRetry = getInteger(properties, SpoutConfig.MESSAGES_MAX_RETRY, SpoutConfig.DEFAULT_MESSAGES_MAX_RETRY);
+ int ttl = getInteger(properties, SpoutConfig.MESSAGES_TTL, SpoutConfig.DEFAULT_MESSAGES_TTL);
+ this.messageRetryManager = new DefaultMessageRetryManager(queue, maxRetry, ttl);
+ this.collector = collector;
+ }
+
+ public boolean process(List<MessageExt> msgs) {
+ if (msgs.isEmpty()) {
+ return true;
+ }
+ MessageSet messageSet = new MessageSet(msgs);
--- End diff --
Yes, the work is in progress. I will update the PR later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
POM files updated & Rebased.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r109309713
--- Diff: external/storm-rocketmq/pom.xml ---
@@ -0,0 +1,89 @@
+<?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.0.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-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.0.0-incubating</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
--- End diff --
No need\uff0c RocketMQ client has dependency commons-lang3, which is a next generation for commons lang package~
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
@vongosling @hustfxj Updated. Can you take a look again?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on the issue:
https://github.com/apache/storm/pull/2024
@vesense thank you. +1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
@vongosling Thanks for your comments. And I will rebase the code on master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r109309839
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,179 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private boolean ordered;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ ordered = getBoolean(properties,RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Utils.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
+ }
+ }
--- End diff --
Nowadays, RocketMQ does not supported many instances in one machine. If we wish it , we must set the instanceID, as the 75 line listed in the https://github.com/rocketmq/rocketmq-storm/blob/master/src/main/java/org/apache/rocketmq/integration/storm/spout/SimpleMessageSpout.java.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111462177
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
--- End diff --
even if its thread-safe shouldn't we consider making per spout instance its own consumer. That way it will more performant instead of one consumer making a call to the rocketmq-servers?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
@vongosling @hustfxj @harshach Updated. Can you take a look again?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111294519
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,177 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Integer.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING).toString()));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
--- End diff --
The messages maybe lost because the consumer is based the automatic commit mode of RocketMq here. It had better not commit the message until the storm handle the message successfully. Of course, this is only a consumption strategy.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111697081
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,177 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Integer.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING).toString()));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
--- End diff --
@vesense The push mode can also send messages with at-least-once. You can commit the message at "ack(Object msgId)" , which means that spout don't commit the message until the storm handle the message successfully.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
I will update the POM files accordingly based on STORM-2416(Reduce release package size) later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2024
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111292663
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,179 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private boolean ordered;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ ordered = getBoolean(properties,RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Utils.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
+ }
+ }
--- End diff --
@vesense In fact I don't suggest that per task has a consumer by setting the instanceId, thus per worker maybe have lots of consumer. Why not consider that per worker only has a consumer by singleton pattern.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
Any comments are welcome.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the issue:
https://github.com/apache/storm/pull/2024
LGTM @vesense . +1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r110599801
--- Diff: external/storm-rocketmq/pom.xml ---
@@ -0,0 +1,89 @@
+<?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.0.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-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.0.0-incubating</version>
+ </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>
+ <version>${mockito.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>test</scope>
--- End diff --
This is not used. Will remove.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111551137
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ if (ordered) {
+ consumer.registerMessageListener(new MessageListenerOrderly() {
--- End diff --
In fact, the RocketMQ "push" mode is still pulling data from broker. PushConsumer is a high level consumer API, wrapping the pulling details, looks like broker push messages to consumer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vongosling <gi...@git.apache.org>.
Github user vongosling commented on the issue:
https://github.com/apache/storm/pull/2024
Great ~
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111551133
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
--- End diff --
Maybe my code comment is not so clear. thread-safe is just precondition, the important is that this is related to the RocketMQ internal implementation(sharing queue, threads, etc.), "Consumer concurrency / Only one consumer instance per process" is the way official recommend.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111344417
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,179 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private boolean ordered;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ ordered = getBoolean(properties,RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Utils.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
+ }
+ }
--- End diff --
Yes, I see the RocketMQ Consumer is thread-safe, sharing a single instance across threads should generally be faster than having multiple instances, as well as Producer. I will refactor them.
@hustfxj Thanks for your suggestion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111882508
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,177 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Integer.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING).toString()));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
--- End diff --
@hustfxj I updated the code. Now all consumed messages will be put into queue and managed by `MessageRetryManager`. Of course, this is in storm side to implement the "at-least-once".
BTW, I'm not sure if I misunderstand the `ConsumeConcurrentlyStatus`:
```
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
if (process(msgs)) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else {
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
```
When returning CONSUME_SUCCESS, the consumer will commit the offset automatically. And when returning RECONSUME_LATER, the consumer will consume the failed messages again? Do you mind explaining what the consumer will do after receiving the `ConsumeConcurrentlyStatus.RECONSUME_LATER`?
Please correct me if I misunderstand. Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r111462620
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,189 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.ObjectReader;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ boolean ordered = getBoolean(properties, RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ // Since RocketMQ Consumer is thread-safe, RocketMQSpout uses a single
+ // consumer instance across threads to improve the performance.
+ synchronized (RocketMQSpout.class) {
+ if (consumer == null) {
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer);
+
+ if (ordered) {
+ consumer.registerMessageListener(new MessageListenerOrderly() {
--- End diff --
is this a push model from server instead of spout polling?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2024: STORM-2349: Add one RocketMQ plugin for the Apache Storm
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2024
Hi @hustfxj I guess you know Apache RocketMQ a lot, please take a look if you have time.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r110599823
--- Diff: external/storm-rocketmq/src/main/java/org/apache/storm/rocketmq/spout/RocketMQSpout.java ---
@@ -0,0 +1,179 @@
+/**
+ * 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 org.apache.commons.lang.Validate;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MQPushConsumer;
+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.DefaultMessageRetryManager;
+import org.apache.storm.rocketmq.MessageRetryManager;
+import org.apache.storm.rocketmq.MessageSet;
+import org.apache.storm.rocketmq.RocketMQConfig;
+import org.apache.storm.rocketmq.SpoutConfig;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.storm.rocketmq.RocketMQUtils.getBoolean;
+import static org.apache.storm.rocketmq.RocketMQUtils.getInteger;
+
+/**
+ * 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 MQPushConsumer consumer;
+ private SpoutOutputCollector collector;
+ private boolean ordered;
+ private BlockingQueue<MessageSet> queue;
+
+ private Properties properties;
+ private MessageRetryManager messageRetryManager;
+
+ public RocketMQSpout(Properties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ Validate.notEmpty(properties, "Consumer properties can not be empty");
+ ordered = getBoolean(properties,RocketMQConfig.CONSUMER_MESSAGES_ORDERLY, false);
+
+ int queueSize = getInteger(properties, SpoutConfig.QUEUE_SIZE, Utils.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING)));
+ queue = new LinkedBlockingQueue<>(queueSize);
+
+ consumer = new DefaultMQPushConsumer();
+ RocketMQConfig.buildConsumerConfigs(properties, (DefaultMQPushConsumer)consumer, context);
+
+ 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;
+ }
+ }
--- End diff --
Yes, I noticed this. Please see: [line 156 - 162](https://github.com/apache/storm/pull/2024/files#diff-fa1922acb0c89979e1f3107ee844485aR156) in `RocketMQConfig`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2024: STORM-2349: Add one RocketMQ plugin for the Apache...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/2024#discussion_r110599786
--- Diff: external/storm-rocketmq/pom.xml ---
@@ -0,0 +1,89 @@
+<?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.0.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-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>4.0.0-incubating</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
--- End diff --
This is used for validating parameters. It will cause `ClassNotFound` exception without the dependency.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---