content: /**
   * 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 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;
  
  import backtype.storm.Config;
  import backtype.storm.spout.SpoutOutputCollector;
  import backtype.storm.task.ICollectorCallback;
  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;
  
  /**
   * Batch spout trigger with strong sequential order
   *
   * @author zhongyan.feng
   */
  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;
  
      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<>();
          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("unknown BatchStatus type: " + 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())) {
              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());
          collector.emit(streamId, new Values(msgId.getBatchId()), msgId, new EmitCb(msgId));
      }
  
      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 {
                  // a larger batchId fails, when the earlier msgId finishes 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);
          } else {
              LOG.warn("Unknown msgId type " + msgId.getClass().getName() + ":" + msgId);
          }
      }
  
      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 msgId type " + msgId.getClass().getName() + ":" + msgId);
          }
      }
  
      @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<>();
          ConfigExtension.setSpoutSingleThread(map, true);
          return map;
      }
  
      class EmitCb implements ICollectorCallback {
          private BatchSpoutMsgId msgId;
  
          public EmitCb(BatchSpoutMsgId msgId) {
              this.msgId = msgId;
          }
  
          @Override
          public void execute(String stream, List<Integer> outTasks, List values) {
              if (outTasks.isEmpty()) {
                  forward(msgId);
              }
          }
      }
  }
  