Add jstorm-utility module

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "jstorm-utility/rocket-mq"]
+	path = jstorm-utility/rocket-mq
+	url = https://github.com/rocketmq/rocketmq-storm

jstorm-utility/transaction_meta_spout/conf/topology.yaml

@@ -0,0 +1,21 @@
+
+meta.consumer.group: "S_BATCH_META_TEST"
+meta.topic: "jae_log"
+meta.subexpress: "*"
+meta.nameserver: null
+#type "yyyyMMddhhmmss"
+#meta.start.timestamp: "20140701000000"
+meta.start.timestamp: null
+meta.batch.message.num: 1024
+meta.max.fail.times: 10
+meta.rebalance.frequency.sec: 60
+
+
+storm.cluster.mode: "local"
+#storm.cluster.mode: "distributed"
+
+topology.name: "batch_test"
+topology.spout.parallel: 1
+topology.bolt.parallel: 1
+topology.message.timeout.secs: 300
+topology.workers: 10

jstorm-utility/transaction_meta_spout/pom.xml

@@ -0,0 +1,68 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>com.alibaba.aloha</groupId>
+    <artifactId>aloha-utility</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+  </parent>
+  <artifactId>transaction_meta_spout</artifactId>
+
+	<build>
+		<plugins>
+			<plugin>
+				<artifactId>maven-assembly-plugin</artifactId>
+				<configuration>
+					<descriptorRefs>
+						<descriptorRef>jar-with-dependencies</descriptorRef>
+					</descriptorRefs>
+				</configuration>
+				<executions>
+					<execution>
+						<id>make-assembly</id>
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>1.6</source>
+					<target>1.6</target>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+	<dependencies>
+		<dependency>
+			<groupId>com.taobao.metaq.final</groupId>
+			<artifactId>metaq-client</artifactId>
+			<version>3.1.3</version>
+		</dependency>
+		<!-- 
+		<dependency>
+			<groupId>com.alibaba.rocketmq</groupId>
+			<artifactId>rocketmq-common</artifactId>
+			<version>3.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.alibaba.rocketmq</groupId>
+			<artifactId>rocketmq-client</artifactId>
+			<version>3.0.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.alibaba.rocketmq</groupId>
+			<artifactId>rocketmq-remoting</artifactId>
+			<version>3.0.1</version>
+		</dependency>
+		 -->
+
+	</dependencies>
+
+</project>

jstorm-utility/transaction_meta_spout/src/main/java/com/alibaba/jstorm/batch/example/BatchMetaRebalance.java

@@ -0,0 +1,108 @@
+package com.alibaba.jstorm.batch.example;
+
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.log4j.Logger;
+
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.IBasicBolt;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import com.alibaba.jstorm.batch.BatchId;
+import com.alibaba.jstorm.batch.IPostCommit;
+import com.alibaba.jstorm.utils.JStormUtils;
+
+public class BatchMetaRebalance implements IBasicBolt, IPostCommit {
+	/**  */
+	private static final long serialVersionUID = 2991323223385556163L;
+
+	private static final Logger LOG = Logger
+			.getLogger(BatchMetaRebalance.class);
+
+	public static final String BOLT_NAME = BatchMetaRebalance.class
+			.getSimpleName();
+
+	public static final String REBALANCE_STREAM_ID = "rebalance";
+
+	private transient AtomicBoolean isNeedRebalance;
+	private transient ScheduledExecutorService scheduExec;
+
+	@Override
+	public void prepare(Map stormConf, TopologyContext context) {
+		isNeedRebalance = new AtomicBoolean(false);
+		CheckRebalanceTimer timer = new CheckRebalanceTimer();
+
+		int rebalanceTimeInterval = JStormUtils.parseInt(
+				stormConf.get("meta.rebalance.frequency.sec"), 3600);
+
+		long now = System.currentTimeMillis();
+		long next = (now/(rebalanceTimeInterval * 1000) + 1) * rebalanceTimeInterval * 1000;
+		long diff = (next - now )/1000;
+
+//		Calendar start = Calendar.getInstance();
+//
+//		start.add(Calendar.HOUR_OF_DAY, 1);
+//
+//		start.set(Calendar.MINUTE, 30);
+//		start.set(Calendar.SECOND, 0);
+//		start.set(Calendar.MILLISECOND, 0);
+//		long startMillis = start.getTimeInMillis();
+//
+//		long now = System.currentTimeMillis();
+//
+//		long diff = (startMillis - now) / (1000);
+
+		ScheduledExecutorService scheduExec = null;
+
+		scheduExec = Executors.newSingleThreadScheduledExecutor();
+		scheduExec.scheduleAtFixedRate(timer, diff, rebalanceTimeInterval, 
+				TimeUnit.SECONDS);
+
+		LOG.info("Successfully init rebalance timer");
+	}
+
+	@Override
+	public void execute(Tuple input, BasicOutputCollector collector) {
+		LOG.warn("Receive unkonw message");
+	}
+
+	@Override
+	public void cleanup() {
+		scheduExec.shutdown();
+		LOG.info("Successfully do cleanup");
+	}
+
+	@Override
+	public void postCommit(BatchId id, BasicOutputCollector collector) {
+		if (isNeedRebalance.get() == true) {
+			isNeedRebalance.set(false);
+			collector.emit(REBALANCE_STREAM_ID, new Values(id));
+			LOG.info("Emit rebalance command");
+		}
+	}
+
+	@Override
+	public void declareOutputFields(OutputFieldsDeclarer declarer) {
+		declarer.declareStream(REBALANCE_STREAM_ID, new Fields("BatchId"));
+	}
+
+	@Override
+	public Map<String, Object> getComponentConfiguration() {
+		return null;
+	}
+
+	public class CheckRebalanceTimer implements Runnable {
+		public void run() {
+			BatchMetaRebalance.this.isNeedRebalance.set(true);
+		}
+	}
+
+}

jstorm-utility/transaction_meta_spout/src/main/java/com/alibaba/jstorm/batch/example/BatchMetaSpout.java

@@ -0,0 +1,131 @@
+package com.alibaba.jstorm.batch.example;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.log4j.Logger;
+
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.FailedException;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import com.alibaba.jstorm.batch.BatchId;
+import com.alibaba.jstorm.batch.IBatchSpout;
+import com.alibaba.jstorm.batch.meta.MetaSimpleClient;
+import com.alibaba.jstorm.batch.meta.MetaSpoutConfig;
+import com.alibaba.jstorm.batch.util.BatchCommon;
+import com.alibaba.jstorm.cluster.ClusterState;
+import com.alibaba.rocketmq.common.message.MessageExt;
+
+public class BatchMetaSpout implements IBatchSpout{
+	private static final long serialVersionUID = 5720810158625748041L;
+
+	private static final Logger LOG = Logger.getLogger(BatchMetaSpout.class);
+
+	public static final String SPOUT_NAME = BatchMetaSpout.class.getSimpleName();
+
+	private Map conf;
+
+	private String taskName;
+	private int taskIndex;
+	private int taskParallel;
+
+	private transient MetaSimpleClient metaClient;
+	private MetaSpoutConfig metaSpoutConfig;
+
+
+
+	public BatchMetaSpout(MetaSpoutConfig metaSpoutConfig) {
+		this.metaSpoutConfig = metaSpoutConfig;
+	}
+
+	public void initMetaClient() throws Exception {
+		ClusterState zkClient = BatchCommon.getZkClient(conf);
+		metaClient = new MetaSimpleClient(metaSpoutConfig, zkClient, taskIndex,
+				taskParallel);
+
+		metaClient.init();
+
+		LOG.info("Successfully init meta client " + taskName);
+	}
+
+	@Override
+	public void prepare(Map stormConf, TopologyContext context) {
+		this.conf = stormConf;
+
+		taskName = context.getThisComponentId() + "_" + context.getThisTaskId();
+
+		taskIndex = context.getThisTaskIndex();
+
+		taskParallel = context.getComponentTasks(context.getThisComponentId())
+				.size();
+
+		try {
+			initMetaClient();
+		} catch (Exception e) {
+			LOG.info("Failed to init Meta Client,", e);
+			throw new RuntimeException(e);
+		}
+
+		LOG.info(taskName + " successfully do prepare ");
+	}
+
+	public void emitBatch(BatchId batchId, BasicOutputCollector collector) {
+		List<MessageExt> msgs = metaClient.fetchOneBatch();
+		for (MessageExt msgExt : msgs) {
+			collector.emit(new Values(batchId, msgExt));
+		}
+	}
+
+	@Override
+	public void execute(Tuple input, BasicOutputCollector collector) {
+
+		String streamId = input.getSourceStreamId();
+		if (streamId.equals(BatchMetaRebalance.REBALANCE_STREAM_ID)) {
+			try {
+				metaClient.rebalanceMqList();
+			} catch (Exception e) {
+				LOG.warn("Failed to do rebalance operation", e);
+			}
+		}else {
+			BatchId batchId = (BatchId) input.getValue(0);
+
+			emitBatch(batchId, collector);
+		}
+
+	}
+
+	@Override
+	public void cleanup() {
+		metaClient.cleanup();
+
+	}
+
+	@Override
+	public byte[] commit(BatchId id) throws FailedException {
+		return metaClient.commit(id);
+	}
+
+	@Override
+	public void revert(BatchId id, byte[] commitResult) {
+		metaClient.revert(id, commitResult);
+	}
+
+	@Override
+	public void declareOutputFields(OutputFieldsDeclarer declarer) {
+		declarer.declare(new Fields("BatchId", "MessageExt"));
+	}
+
+	@Override
+	public Map<String, Object> getComponentConfiguration() {
+		return null;
+	}
+
+
+
+
+}