jstorm-core/src/main/java/com/alibaba/jstorm/batch/impl/BatchSpoutTrigger.java - storm - Git at Google

 /**
  * 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 com.alibaba.jstorm.batch.impl;

 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import backtype.storm.Config;
 import backtype.storm.spout.SpoutOutputCollector;
 import backtype.storm.task.TopologyContext;
 import backtype.storm.topology.IRichSpout;
 import backtype.storm.topology.OutputFieldsDeclarer;
 import backtype.storm.tuple.Fields;
 import backtype.storm.tuple.Values;

 import com.alibaba.jstorm.batch.BatchId;
 import com.alibaba.jstorm.batch.util.BatchCommon;
 import com.alibaba.jstorm.batch.util.BatchDef;
 import com.alibaba.jstorm.batch.util.BatchStatus;
 import com.alibaba.jstorm.client.ConfigExtension;
 import com.alibaba.jstorm.cluster.ClusterState;
 import com.alibaba.jstorm.utils.IntervalCheck;
 import com.alibaba.jstorm.utils.JStormUtils;

 /**
  * Strong Sequence
  *
  * @author zhongyan.feng
  * @version
  */
 public class BatchSpoutTrigger implements IRichSpout {
     /**  */
     private static final long serialVersionUID = 7215109169247425954L;

     private static final Logger LOG = LoggerFactory.getLogger(BatchSpoutTrigger.class);

     private LinkedBlockingQueue<BatchSpoutMsgId> batchQueue;

     private transient ClusterState zkClient;

     private transient SpoutOutputCollector collector;

     private static final String ZK_NODE_PATH = "/trigger";

     private static BatchId currentBatchId = null;

     private Map conf;

     private String taskName;

     private IntervalCheck intervalCheck;

     /**
      * @throws Exception
      *
      */
     public void initMsgId() throws Exception {
         Long zkMsgId = null;
         byte[] data = zkClient.get_data(ZK_NODE_PATH, false);
         if (data != null) {
             String value = new String(data);
             try {
                 zkMsgId = Long.valueOf(value);
                 LOG.info("ZK msgId:" + zkMsgId);
             } catch (Exception e) {
                 LOG.warn("Failed to get msgId ", e);

             }

         }

         if (zkMsgId != null) {
             BatchId.updateId(zkMsgId);
         }

         int max_spout_pending = JStormUtils.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 1);

         for (int i = 0; i < max_spout_pending; i++) {
             BatchSpoutMsgId msgId = BatchSpoutMsgId.mkInstance();
             if (currentBatchId == null) {
                 currentBatchId = msgId.getBatchId();
             }
             batchQueue.offer(msgId);
             LOG.info("Push into queue," + msgId);
         }

     }

     @Override
     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
         batchQueue = new LinkedBlockingQueue<BatchSpoutMsgId>();
         this.collector = collector;
         this.conf = conf;
         taskName = context.getThisComponentId() + "_" + context.getThisTaskId();

         intervalCheck = new IntervalCheck();

         try {
             zkClient = BatchCommon.getZkClient(conf);

             initMsgId();

         } catch (Exception e) {
             LOG.error("", e);
             throw new RuntimeException("Failed to init");
         }
         LOG.info("Successfully open " + taskName);
     }

     @Override
     public void close() {
         zkClient.close();
     }

     @Override
     public void activate() {
         LOG.info("Activate " + taskName);
     }

     @Override
     public void deactivate() {
         LOG.info("Deactivate " + taskName);
     }

     protected String getStreamId(BatchStatus batchStatus) {
         if (batchStatus == BatchStatus.COMPUTING) {
             return BatchDef.COMPUTING_STREAM_ID;
         } else if (batchStatus == BatchStatus.PREPARE_COMMIT) {
             return BatchDef.PREPARE_STREAM_ID;
         } else if (batchStatus == BatchStatus.COMMIT) {
             return BatchDef.COMMIT_STREAM_ID;
         } else if (batchStatus == BatchStatus.POST_COMMIT) {
             return BatchDef.POST_STREAM_ID;
         } else if (batchStatus == BatchStatus.REVERT_COMMIT) {
             return BatchDef.REVERT_STREAM_ID;
         } else {
             LOG.error("Occur unkonw type BatchStatus " + batchStatus);
             throw new RuntimeException();
         }
     }

     protected boolean isCommitStatus(BatchStatus batchStatus) {
         if (batchStatus == BatchStatus.COMMIT) {
             return true;
         } else if (batchStatus == BatchStatus.REVERT_COMMIT) {
             return true;
         } else {
             return false;
         }
     }

     protected boolean isCommitWait(BatchSpoutMsgId msgId) {

         if (isCommitStatus(msgId.getBatchStatus()) == false) {
             return false;
         }

         // left status is commit status
         if (currentBatchId.getId() >= msgId.getBatchId().getId()) {
             return false;
         }

         return true;
     }

     @Override
     public void nextTuple() {
         BatchSpoutMsgId msgId = null;
         try {
             msgId = batchQueue.poll(10, TimeUnit.MILLISECONDS);
         } catch (InterruptedException e) {
             LOG.error("", e);
         }
         if (msgId == null) {
             return;
         }

         if (isCommitWait(msgId)) {

             batchQueue.offer(msgId);
             if (intervalCheck.check()) {
                 LOG.info("Current msgId " + msgId + ", but current commit BatchId is " + currentBatchId);
             } else {
                 LOG.debug("Current msgId " + msgId + ", but current commit BatchId is " + currentBatchId);
             }

             return;
         }

         String streamId = getStreamId(msgId.getBatchStatus());
         List<Integer> outTasks = collector.emit(streamId, new Values(msgId.getBatchId()), msgId);
         if (outTasks.isEmpty()) {
             forward(msgId);
         }
         return;

     }

     protected void mkMsgId(BatchSpoutMsgId oldMsgId) {
         synchronized (BatchSpoutMsgId.class) {
             if (currentBatchId.getId() <= oldMsgId.getBatchId().getId()) {
                 // this is normal case

                 byte[] data = String.valueOf(currentBatchId.getId()).getBytes();
                 try {
                     zkClient.set_data(ZK_NODE_PATH, data);
                 } catch (Exception e) {
                     LOG.error("Failed to update to ZK " + oldMsgId, e);
                 }

                 currentBatchId = BatchId.incBatchId(oldMsgId.getBatchId());

             } else {
                 // bigger batchId has been failed, when old msgId finish
                 // it will go here

             }

         }

         BatchSpoutMsgId newMsgId = BatchSpoutMsgId.mkInstance();
         batchQueue.offer(newMsgId);
         StringBuilder sb = new StringBuilder();
         sb.append("Create new BatchId,");
         sb.append("old:").append(oldMsgId);
         sb.append("new:").append(newMsgId);
         sb.append("currentBatchId:").append(currentBatchId);
         LOG.info(sb.toString());
     }

     protected void forward(BatchSpoutMsgId msgId) {
         BatchStatus status = msgId.getBatchStatus();

         BatchStatus newStatus = status.forward();
         if (newStatus == null) {
             // create new status
             mkMsgId(msgId);
             LOG.info("Finish old batch " + msgId);

         } else {
             msgId.setBatchStatus(newStatus);
             batchQueue.offer(msgId);
             LOG.info("Forward batch " + msgId);
         }
     }

     @Override
     public void ack(Object msgId) {
         if (msgId instanceof BatchSpoutMsgId) {
             forward((BatchSpoutMsgId) msgId);
             return;
         } else {
             LOG.warn("Unknown type msgId " + msgId.getClass().getName() + ":" + msgId);
             return;
         }
     }

     protected void handleFail(BatchSpoutMsgId msgId) {
         LOG.info("Failed batch " + msgId);
         BatchStatus status = msgId.getBatchStatus();

         BatchStatus newStatus = status.error();
         if (newStatus == BatchStatus.ERROR) {
             // create new status
             mkMsgId(msgId);

         } else {

             msgId.setBatchStatus(newStatus);
             batchQueue.offer(msgId);

         }
     }

     @Override
     public void fail(Object msgId) {
         if (msgId instanceof BatchSpoutMsgId) {
             handleFail((BatchSpoutMsgId) msgId);
         } else {
             LOG.warn("Unknown type msgId " + msgId.getClass().getName() + ":" + msgId);
             return;
         }
     }

     @Override
     public void declareOutputFields(OutputFieldsDeclarer declarer) {
         declarer.declareStream(BatchDef.COMPUTING_STREAM_ID, new Fields("BatchId"));
         declarer.declareStream(BatchDef.PREPARE_STREAM_ID, new Fields("BatchId"));
         declarer.declareStream(BatchDef.COMMIT_STREAM_ID, new Fields("BatchId"));
         declarer.declareStream(BatchDef.REVERT_STREAM_ID, new Fields("BatchId"));
         declarer.declareStream(BatchDef.POST_STREAM_ID, new Fields("BatchId"));
     }

     @Override
     public Map<String, Object> getComponentConfiguration() {
         Map<String, Object> map = new HashMap<String, Object>();
         ConfigExtension.setSpoutSingleThread(map, true);
         return map;
     }

 }
	/**
	* 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 com.alibaba.jstorm.batch.impl;

	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.TimeUnit;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import backtype.storm.Config;
	import backtype.storm.spout.SpoutOutputCollector;
	import backtype.storm.task.TopologyContext;
	import backtype.storm.topology.IRichSpout;
	import backtype.storm.topology.OutputFieldsDeclarer;
	import backtype.storm.tuple.Fields;
	import backtype.storm.tuple.Values;

	import com.alibaba.jstorm.batch.BatchId;
	import com.alibaba.jstorm.batch.util.BatchCommon;
	import com.alibaba.jstorm.batch.util.BatchDef;
	import com.alibaba.jstorm.batch.util.BatchStatus;
	import com.alibaba.jstorm.client.ConfigExtension;
	import com.alibaba.jstorm.cluster.ClusterState;
	import com.alibaba.jstorm.utils.IntervalCheck;
	import com.alibaba.jstorm.utils.JStormUtils;

	/**
	* Strong Sequence
	*
	* @author zhongyan.feng
	* @version
	*/
	public class BatchSpoutTrigger implements IRichSpout {
	/** */
	private static final long serialVersionUID = 7215109169247425954L;

	private static final Logger LOG = LoggerFactory.getLogger(BatchSpoutTrigger.class);

	private LinkedBlockingQueue<BatchSpoutMsgId> batchQueue;

	private transient ClusterState zkClient;

	private transient SpoutOutputCollector collector;

	private static final String ZK_NODE_PATH = "/trigger";

	private static BatchId currentBatchId = null;

	private Map conf;

	private String taskName;

	private IntervalCheck intervalCheck;

	/**
	* @throws Exception
	*
	*/
	public void initMsgId() throws Exception {
	Long zkMsgId = null;
	byte[] data = zkClient.get_data(ZK_NODE_PATH, false);
	if (data != null) {
	String value = new String(data);
	try {
	zkMsgId = Long.valueOf(value);
	LOG.info("ZK msgId:" + zkMsgId);
	} catch (Exception e) {
	LOG.warn("Failed to get msgId ", e);

	}

	}

	if (zkMsgId != null) {
	BatchId.updateId(zkMsgId);
	}

	int max_spout_pending = JStormUtils.parseInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 1);

	for (int i = 0; i < max_spout_pending; i++) {
	BatchSpoutMsgId msgId = BatchSpoutMsgId.mkInstance();
	if (currentBatchId == null) {
	currentBatchId = msgId.getBatchId();
	}
	batchQueue.offer(msgId);
	LOG.info("Push into queue," + msgId);
	}

	}

	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
	batchQueue = new LinkedBlockingQueue<BatchSpoutMsgId>();
	this.collector = collector;
	this.conf = conf;
	taskName = context.getThisComponentId() + "_" + context.getThisTaskId();

	intervalCheck = new IntervalCheck();

	try {
	zkClient = BatchCommon.getZkClient(conf);

	initMsgId();

	} catch (Exception e) {
	LOG.error("", e);
	throw new RuntimeException("Failed to init");
	}
	LOG.info("Successfully open " + taskName);
	}

	@Override
	public void close() {
	zkClient.close();
	}

	@Override
	public void activate() {
	LOG.info("Activate " + taskName);
	}

	@Override
	public void deactivate() {
	LOG.info("Deactivate " + taskName);
	}

	protected String getStreamId(BatchStatus batchStatus) {
	if (batchStatus == BatchStatus.COMPUTING) {
	return BatchDef.COMPUTING_STREAM_ID;
	} else if (batchStatus == BatchStatus.PREPARE_COMMIT) {
	return BatchDef.PREPARE_STREAM_ID;
	} else if (batchStatus == BatchStatus.COMMIT) {
	return BatchDef.COMMIT_STREAM_ID;
	} else if (batchStatus == BatchStatus.POST_COMMIT) {
	return BatchDef.POST_STREAM_ID;
	} else if (batchStatus == BatchStatus.REVERT_COMMIT) {
	return BatchDef.REVERT_STREAM_ID;
	} else {
	LOG.error("Occur unkonw type BatchStatus " + batchStatus);
	throw new RuntimeException();
	}
	}

	protected boolean isCommitStatus(BatchStatus batchStatus) {
	if (batchStatus == BatchStatus.COMMIT) {
	return true;
	} else if (batchStatus == BatchStatus.REVERT_COMMIT) {
	return true;
	} else {
	return false;
	}
	}

	protected boolean isCommitWait(BatchSpoutMsgId msgId) {

	if (isCommitStatus(msgId.getBatchStatus()) == false) {
	return false;
	}

	// left status is commit status
	if (currentBatchId.getId() >= msgId.getBatchId().getId()) {
	return false;
	}

	return true;
	}

	@Override
	public void nextTuple() {
	BatchSpoutMsgId msgId = null;
	try {
	msgId = batchQueue.poll(10, TimeUnit.MILLISECONDS);
	} catch (InterruptedException e) {
	LOG.error("", e);
	}
	if (msgId == null) {
	return;
	}

	if (isCommitWait(msgId)) {

	batchQueue.offer(msgId);
	if (intervalCheck.check()) {
	LOG.info("Current msgId " + msgId + ", but current commit BatchId is " + currentBatchId);
	} else {
	LOG.debug("Current msgId " + msgId + ", but current commit BatchId is " + currentBatchId);
	}

	return;
	}

	String streamId = getStreamId(msgId.getBatchStatus());
	List<Integer> outTasks = collector.emit(streamId, new Values(msgId.getBatchId()), msgId);
	if (outTasks.isEmpty()) {
	forward(msgId);
	}
	return;

	}

	protected void mkMsgId(BatchSpoutMsgId oldMsgId) {
	synchronized (BatchSpoutMsgId.class) {
	if (currentBatchId.getId() <= oldMsgId.getBatchId().getId()) {
	// this is normal case

	byte[] data = String.valueOf(currentBatchId.getId()).getBytes();
	try {
	zkClient.set_data(ZK_NODE_PATH, data);
	} catch (Exception e) {
	LOG.error("Failed to update to ZK " + oldMsgId, e);
	}

	currentBatchId = BatchId.incBatchId(oldMsgId.getBatchId());

	} else {
	// bigger batchId has been failed, when old msgId finish
	// it will go here

	}

	}

	BatchSpoutMsgId newMsgId = BatchSpoutMsgId.mkInstance();
	batchQueue.offer(newMsgId);
	StringBuilder sb = new StringBuilder();
	sb.append("Create new BatchId,");
	sb.append("old:").append(oldMsgId);
	sb.append("new:").append(newMsgId);
	sb.append("currentBatchId:").append(currentBatchId);
	LOG.info(sb.toString());
	}

	protected void forward(BatchSpoutMsgId msgId) {
	BatchStatus status = msgId.getBatchStatus();

	BatchStatus newStatus = status.forward();
	if (newStatus == null) {
	// create new status
	mkMsgId(msgId);
	LOG.info("Finish old batch " + msgId);

	} else {
	msgId.setBatchStatus(newStatus);
	batchQueue.offer(msgId);
	LOG.info("Forward batch " + msgId);
	}
	}

	@Override
	public void ack(Object msgId) {
	if (msgId instanceof BatchSpoutMsgId) {
	forward((BatchSpoutMsgId) msgId);
	return;
	} else {
	LOG.warn("Unknown type msgId " + msgId.getClass().getName() + ":" + msgId);
	return;
	}
	}

	protected void handleFail(BatchSpoutMsgId msgId) {
	LOG.info("Failed batch " + msgId);
	BatchStatus status = msgId.getBatchStatus();

	BatchStatus newStatus = status.error();
	if (newStatus == BatchStatus.ERROR) {
	// create new status
	mkMsgId(msgId);

	} else {

	msgId.setBatchStatus(newStatus);
	batchQueue.offer(msgId);

	}
	}

	@Override
	public void fail(Object msgId) {
	if (msgId instanceof BatchSpoutMsgId) {
	handleFail((BatchSpoutMsgId) msgId);
	} else {
	LOG.warn("Unknown type msgId " + msgId.getClass().getName() + ":" + msgId);
	return;
	}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	declarer.declareStream(BatchDef.COMPUTING_STREAM_ID, new Fields("BatchId"));
	declarer.declareStream(BatchDef.PREPARE_STREAM_ID, new Fields("BatchId"));
	declarer.declareStream(BatchDef.COMMIT_STREAM_ID, new Fields("BatchId"));
	declarer.declareStream(BatchDef.REVERT_STREAM_ID, new Fields("BatchId"));
	declarer.declareStream(BatchDef.POST_STREAM_ID, new Fields("BatchId"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
	Map<String, Object> map = new HashMap<String, Object>();
	ConfigExtension.setSpoutSingleThread(map, true);
	return map;
	}

	}