content: /*
   * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
   *
   * Licensed 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.otter.node.etl.select;
  
  import java.util.Date;
  import java.util.List;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.locks.LockSupport;
  
  import org.slf4j.MDC;
  import org.springframework.util.CollectionUtils;
  
  import com.alibaba.otter.canal.common.CanalException;
  import com.alibaba.otter.node.common.statistics.StatisticsClientService;
  import com.alibaba.otter.node.etl.OtterConstants;
  import com.alibaba.otter.node.etl.common.jmx.StageAggregation.AggregationItem;
  import com.alibaba.otter.node.etl.common.pipe.PipeKey;
  import com.alibaba.otter.node.etl.common.task.GlobalTask;
  import com.alibaba.otter.node.etl.extract.SetlFuture;
  import com.alibaba.otter.node.etl.select.exceptions.SelectException;
  import com.alibaba.otter.node.etl.select.selector.Message;
  import com.alibaba.otter.node.etl.select.selector.OtterSelector;
  import com.alibaba.otter.node.etl.select.selector.OtterSelectorFactory;
  import com.alibaba.otter.shared.arbitrate.model.EtlEventData;
  import com.alibaba.otter.shared.arbitrate.model.TerminEventData;
  import com.alibaba.otter.shared.common.model.config.channel.Channel;
  import com.alibaba.otter.shared.common.model.config.enums.StageType;
  import com.alibaba.otter.shared.common.model.statistics.delay.DelayCount;
  import com.alibaba.otter.shared.common.utils.lock.BooleanMutex;
  import com.alibaba.otter.shared.etl.model.DbBatch;
  import com.alibaba.otter.shared.etl.model.EventData;
  import com.alibaba.otter.shared.etl.model.Identity;
  import com.alibaba.otter.shared.etl.model.RowBatch;
  
  /**
   * select流处理模式的实现版本
   * 
   * <pre>
   * 调度模型：
   * 1. 正常运行调度流程 
   * 假如并行度为3
   * ----------------------------------------------------------->时间轴
   * | ProcessSelect
   * --> 1 
   *      --> 2
   *           -->3
   *               --> 1
   * | ProcessTermin
   *     ---> 1 ack
   *          ----> 2ack
   *                 ---> 3ack
   * a. ProcessSelect拿到数据后，丢入pool池进行异步处理，并通知ProcessTermin顺序接收termin信号
   * b. 同一时间在s/e/t/l流水线上的数据受并行度控制，满了就会阻塞ProcessSelect，避免取过多的数据，只会多取一份，等待其中一个s/e/t/l完成
   * c. ProcessTermin接受到termin信号
   *    i. 会严格按照发出去的batchId/processId进行对比，发现不匹配，发起rollback操作.
   *    ii. 会根据terminType判断这一批数据是否处理成功，如果发现不成功，发起rollback操作
   * 
   * 2. 异常调度流程
   * 假如并行度为3
   * |-->1 --> 2 -->3(ing)
   * a. 当第1份数据,ProcessTermin发现需要rollback，此时需要回滚2,3份数据的批次. (可能第2,3份数据还未提交到s/e/t/l调度中)
   *    i. 如果2批次数据已经提交，等待2批次termin信号的返回，此时需要阻塞ProcessSelect，避免再取新数据
   *    ii. 如果第2批次数据未提交，直接rollback数据，不再进入s/e/t/l调度流程
   * b. 当所有批次都已经处理完成，再通知ProcessSelect启动 (注意：这里会避免rollback和get并发操作，会造成数据不一致)
   * 
   * 3. 热备机制
   * a. Select主线程会一直监听mainstem的信号，一旦抢占成功，则启动ProcessSelect/ProcessTermin线程
   * b. ProcessSelect/ProcessTermin在处理过程中，会检查一下当前节点是否为抢占mainstem成功的节点，如果发现不是，立马停止，继续监听mainstem
   * c. ProcessSelect进行get数据之前，会等到ProcessTermin会读取未被处理过termin信号，对上一次的selector进行ack/rollback处理
   *      i. 注意：ProcessSelect进行get数据时，需要保证batch/termin/get操作状态保持一致，必须都处于同一个数据点上
   * </pre>
   * 
   * @author jianghang 2012-7-31 下午05:39:06
   * @version 4.1.0
   */
  public class SelectTask extends GlobalTask {
  
      // 运行调度控制
      private volatile boolean           isStart          = false;
      // 运行
      private StatisticsClientService    statisticsClientService;
      private OtterSelectorFactory       otterSelectorFactory;
      private OtterSelector<Message>     otterSelector;
      private ExecutorService            executor;
      private BlockingQueue<BatchTermin> batchBuffer      = new LinkedBlockingQueue<BatchTermin>(50); // 设置有界队列，避免小batch处理太多
      private boolean                    needCheck        = false;
      private BooleanMutex               canStartSelector = new BooleanMutex(false);                 // 非常轻量的一个阻塞式实现，调用成本低
      private AtomicInteger              rversion         = new AtomicInteger(0);
      private long                       lastResetTime    = new Date().getTime();
  
      public SelectTask(Long pipelineId){
          super(pipelineId);
      }
  
      public void run() {
          MDC.put(OtterConstants.splitPipelineLogFileKey, String.valueOf(pipelineId));
          try {
              while (running) {
                  try {
                      if (isStart) {
                          boolean working = arbitrateEventService.mainStemEvent().check(pipelineId);
                          if (!working) {
                              stopup(false);
                          }
  
                          LockSupport.parkNanos(5 * 1000 * 1000L * 1000L); // 5秒钟检查一次
                      } else {
                          startup();
                      }
                  } catch (Throwable e) {
                      if (isInterrupt(e)) {
                          logger.info("INFO ## select is interrupt", e);
                          return;
                      } else {
                          logger.warn("WARN ## select is failed.", e);
                          sendRollbackTermin(pipelineId, e);
  
                          // sleep 10秒再进行重试
                          try {
                              Thread.sleep(10 * 1000);
                          } catch (InterruptedException e1) {
                          }
                      }
                  }
              }
          } finally {
              arbitrateEventService.mainStemEvent().release(pipelineId);
          }
      }
  
      /**
       * 尝试启动，需要进行mainstem竞争，拿到锁之后才可以进行下一步
       * 
       * @throws InterruptedException
       */
      private void startup() throws InterruptedException {
          try {
              arbitrateEventService.mainStemEvent().await(pipelineId);
          } catch (Throwable e) {
              if (isInterrupt(e)) {
                  logger.info("INFO ## this node is interrupt", e);
              } else {
                  logger.warn("WARN ## this node is crashed.", e);
              }
              arbitrateEventService.mainStemEvent().release(pipelineId);
              return;
          }
  
          executor = Executors.newFixedThreadPool(2); // 启动两个线程
          // 启动selector
          otterSelector = otterSelectorFactory.getSelector(pipelineId); // 获取对应的selector
          otterSelector.start();
  
          canStartSelector.set(false);// 初始化为false
          startProcessTermin();
          startProcessSelect();
  
          isStart = true;
      }
  
      private synchronized void stopup(boolean needInterrut) throws InterruptedException {
          if (isStart) {
              if (executor != null) {
                  executor.shutdownNow();
              }
  
              if (otterSelector != null && otterSelector.isStart()) {
                  otterSelector.stop();
              }
  
              if (needInterrut) {
                  throw new InterruptedException();// 抛异常，退出自己
              }
  
              isStart = false;
          }
      }
  
      /**
       * 执行数据分发工作
       */
      private void startProcessSelect() {
          executor.submit(new Runnable() {
  
              public void run() {
                  MDC.put(OtterConstants.splitPipelineLogFileKey, String.valueOf(pipelineId));
                  String currentName = Thread.currentThread().getName();
                  Thread.currentThread().setName(createTaskName(pipelineId, "ProcessSelect"));
                  try {
                      processSelect();
                  } finally {
                      Thread.currentThread().setName(currentName);
                      MDC.remove(OtterConstants.splitPipelineLogFileKey);
                  }
              }
          });
  
      }
  
      private void processSelect() {
          while (running) {
              try {
                  // 等待ProcessTermin exhaust，会阻塞
                  // ProcessTermin发现出现rollback，会立即通知暂停，比分布式permit及时性高
                  canStartSelector.get();
  
                  // 判断当前是否为工作节点，S模块不能出现双节点工作，selector容易出现数据错乱
                  if (needCheck) {
                      checkContinueWork();
                  }
  
                  // 出现阻塞挂起时，等待mananger处理完成，解挂开启同步
                  arbitrateEventService.toolEvent().waitForPermit(pipelineId);// 出现rollback后能及时停住
  
                  // 使用startVersion要解决的一个问题：出现rollback时，尽可能判断取出来的数据是rollback前还是rollback后，想办法丢弃rollback前的数据。
                  // (因为出现rollback，之前取出去的几个批次的数据其实是没有执行成功，get取出来的数据会是其后一批数据，如果不丢弃的话，会出现后面的数据先执行，然后又回到出错的点，再执行一遍)
                  // int startVersion = rversion.get();
                  Message gotMessage = otterSelector.selector();
  
                  // modify by ljh at 2012-09-10，startVersion获取操作应该放在拿到数据之后
                  // 放在前面 : (遇到一个并发bug)
                  // // a.
                  // 先拿startVersion，再获取数据，在拿数据过程中rollback开始并完成了，导致selector返回时数据已经取到了末尾
                  // // b. 在进行version判断时发现已经有变化，导致又触发一次拿数据的过程，此时的get
                  // cursor已经到队列的末尾，拿不出任何数据，所以出现死等情况
                  // 放在后面 : (一点点瑕疵)
                  // // a.
                  // 并发操作rollback和selector时，针对拿到rollback前的老数据，此时startVersion还未初始化，导致判断不出出现过rollback操作，后面的变更数据会提前同步
                  // (概率性会比较高，取决于selector和初始化startVersion的时间间隔)
                  int startVersion = rversion.get();
  
                  if (canStartSelector.state() == false) { // 是否出现异常
                      // 回滚在出现异常的瞬间，拿出来的数据，因为otterSelector.selector()会循环，可能出现了rollback，其还未感知到
                      rollback(gotMessage.getId());
                      continue;
                  }
  
                  if (CollectionUtils.isEmpty(gotMessage.getDatas())) {// 处理下空数据，也得更新下游标，可能是回环数据被过滤掉
                      // 添加到待响应的buffer列表，不需要await termin信号，因为没启动过s/e/t/l流程
                      batchBuffer.put(new BatchTermin(gotMessage.getId(), false));
                      continue;
                  }
  
                  final EtlEventData etlEventData = arbitrateEventService.selectEvent().await(pipelineId);
                  if (rversion.get() != startVersion) {// 说明存在过变化，中间出现过rollback，需要丢弃该数据
                      logger.warn("rollback happend , should skip this data and get new message.");
                      canStartSelector.get();// 确认一下rollback是否完成
                      // 先睡眠一段时间，保证channel有足够的时间变成pause态，即使没有变成PAUSE态，***MemoryArbitrateEvent里面有回滚操作兜底。
                      Thread.sleep(10 * 1000);
                      arbitrateEventService.toolEvent().waitForPermit(pipelineId);
                      gotMessage = otterSelector.selector();// 这时不管有没有数据，都需要执行一次s/e/t/l
                  }
  
                  final Message message = gotMessage;
                  final BatchTermin batchTermin = new BatchTermin(message.getId(), etlEventData.getProcessId());
                  batchBuffer.put(batchTermin); // 添加到待响应的buffer列表
                  Runnable task = new Runnable() {
  
                      public void run() {
                          // 设置profiling信息
                          boolean profiling = isProfiling();
                          Long profilingStartTime = null;
                          if (profiling) {
                              profilingStartTime = System.currentTimeMillis();
                          }
  
                          MDC.put(OtterConstants.splitPipelineLogFileKey, String.valueOf(pipelineId));
                          String currentName = Thread.currentThread().getName();
                          Thread.currentThread().setName(createTaskName(pipelineId, "SelectWorker"));
                          try {
                              pipeline = configClientService.findPipeline(pipelineId);
                              List<EventData> eventData = message.getDatas();
                              long startTime = etlEventData.getStartTime();
                              if (!CollectionUtils.isEmpty(eventData)) {
                                  startTime = eventData.get(0).getExecuteTime();
                              }
  
                              Channel channel = configClientService.findChannelByPipelineId(pipelineId);
                              RowBatch rowBatch = new RowBatch();
                              // 构造唯一标识
                              Identity identity = new Identity();
                              identity.setChannelId(channel.getId());
                              identity.setPipelineId(pipelineId);
                              identity.setProcessId(etlEventData.getProcessId());
                              rowBatch.setIdentity(identity);
                              // 进行数据合并
                              for (EventData data : eventData) {
                                  rowBatch.merge(data);
                              }
  
                              long nextNodeId = etlEventData.getNextNid();
                              List<PipeKey> pipeKeys = rowDataPipeDelegate.put(new DbBatch(rowBatch), nextNodeId);
                              etlEventData.setDesc(pipeKeys);
                              etlEventData.setNumber((long) eventData.size());
                              etlEventData.setFirstTime(startTime); // 使用原始数据的第一条
                              etlEventData.setBatchId(message.getId());
  
                              if (profiling) {
                                  Long profilingEndTime = System.currentTimeMillis();
                                  stageAggregationCollector.push(pipelineId,
                                      StageType.SELECT,
                                      new AggregationItem(profilingStartTime, profilingEndTime));
                              }
                              arbitrateEventService.selectEvent().single(etlEventData);
                          } catch (Throwable e) {
                              if (!isInterrupt(e)) {
                                  logger.error(String.format("[%s] selectwork executor is error! data:%s",
                                      pipelineId,
                                      etlEventData), e);
                                  sendRollbackTermin(pipelineId, e);
                              } else {
                                  logger.info(String.format("[%s] selectwork executor is interrrupt! data:%s",
                                      pipelineId,
                                      etlEventData), e);
                              }
                          } finally {
                              Thread.currentThread().setName(currentName);
                              MDC.remove(OtterConstants.splitPipelineLogFileKey);
                          }
                      }
                  };
  
                  // 构造pending任务，可在关闭线程时退出任务
                  SetlFuture extractFuture = new SetlFuture(StageType.SELECT,
                      etlEventData.getProcessId(),
                      pendingFuture,
                      task);
                  executorService.execute(extractFuture);
  
              } catch (Throwable e) {
                  if (!isInterrupt(e)) {
                      logger.error(String.format("[%s] selectTask is error!", pipelineId), e);
                      sendRollbackTermin(pipelineId, e);
                  } else {
                      logger.info(String.format("[%s] selectTask is interrrupt!", pipelineId), e);
                      return;
                  }
              }
          }
      }
  
      private void startProcessTermin() {
          executor.submit(new Runnable() {
  
              public void run() {
                  MDC.put(OtterConstants.splitPipelineLogFileKey, String.valueOf(pipelineId));
                  String currentName = Thread.currentThread().getName();
                  Thread.currentThread().setName(createTaskName(pipelineId, "ProcessTermin"));
                  try {
                      boolean lastStatus = true;
                      while (running) {
                          try {
                              // 处理下上一次非正常退出后，未被处理的termin信息
                              lastStatus = true;
                              // 消费掉所有的termin，不管成功还是失败，都回滚canal中的数据，会有重复
                              arbitrateEventService.terminEvent().exhaust(pipelineId);
  
                              batchBuffer.clear();// 清空上一次的待处理的batch记录，因为所有的batch都会被rollback掉
  
                              // 开始处理新的termin数据
                              while (running) {
                                  if (batchBuffer.size() == 0) {
                                      // termin任务已经处理完成，可以通知selector重新开始拉数据
                                      if (canStartSelector.state() == false) {
                                          otterSelector.rollback();// rollback一下所有节点，确保所有的节点都被ack，包括被预取出来的数据
                                      }
  
                                      lastStatus = true;
                                      canStartSelector.set(true);
                                  }
  
                                  BatchTermin batch = batchBuffer.take();
                                  logger.info("start process termin : {}", batch.toString());
                                  if (batch.isNeedWait()) {
                                      lastStatus = processTermin(lastStatus, batch.getBatchId(), batch.getProcessId());
                                  } else {
                                      // 不需要wait的批次，直接以上一个batch的结果决定是否ack
                                      if (lastStatus) {
                                          ack(batch.getBatchId());
                                          sendDelayReset(pipelineId);
                                      } else {
                                          rollback(batch.getBatchId());// 会阻塞selector等待所有batch的rollback操作完成
                                      }
                                  }
  
                                  logger.info("end process termin : {}  result : {}", batch.toString(), lastStatus);
                              }
                          } catch (CanalException e) {// 捕获可处理的异常，进行retry,基本可自行恢复
                              logger.info(String.format("[%s] ProcessTermin has an error! retry...", pipelineId), e);
                              notifyRollback();
                          } catch (SelectException e) {// 捕获可处理的异常，进行retry,基本可自行恢复
                              logger.info(String.format("[%s] ProcessTermin has an error! retry...", pipelineId), e);
                              notifyRollback();
                          } catch (Throwable e) {
                              if (isInterrupt(e)) {
                                  logger.info(String.format("[%s] ProcessTermin is interrupted!", pipelineId), e);
                                  return;
                              } else {
                                  logger.error(String.format("[%s] ProcessTermin is error!", pipelineId), e);
                                  notifyRollback();
                                  sendRollbackTermin(pipelineId, e);
                              }
                          }
  
                          try {
                              Thread.sleep(30000); // sleep 30，等待termin信号都ready
                          } catch (InterruptedException e) {
                          }
                      }
                  } finally {
                      Thread.currentThread().setName(currentName);
                      MDC.remove(OtterConstants.splitPipelineLogFileKey);
                  }
              }
  
          });
      }
  
      private boolean processTermin(boolean lastStatus, Long batchId, Long processId) throws InterruptedException {
          int retry = 0;
          SelectException exception = null;
          TerminEventData terminData = null;
          while (retry++ < 30) {
              // 因为存在网络因素，而且在Load进行termin处理时，因为是异步处理，有一定的概率会出现termin不按顺序过来
              terminData = arbitrateEventService.terminEvent().await(pipelineId);
              Long terminBatchId = terminData.getBatchId();
              Long terminProcessId = terminData.getProcessId();
  
              if (terminBatchId == null && processId != -1L && !processId.equals(terminProcessId)) {
                  // 针对manager发起rollback，terminBatchId可能为null，需要特殊处理下
                  exception = new SelectException("unmatched processId, SelectTask batchId = " + batchId
                                                  + " processId = " + processId + " and Termin Event: "
                                                  + terminData.toString());
                  Thread.sleep(1000); // sleep 1秒，等新的数据包
              } else if (terminBatchId != null && batchId != -1L && !batchId.equals(terminBatchId)) {
                  exception = new SelectException("unmatched terminId, SelectTask batchId = " + batchId + " processId = "
                                                  + processId + " and Termin Event: " + terminData.toString());
                  Thread.sleep(1000); // sleep 1秒，等新的数据包
              } else {
                  exception = null; // batchId/processId对上了，退出
                  break;
              }
          }
  
          if (exception != null) {
              throw exception;
          }
  
          if (needCheck) {
              checkContinueWork();
          }
  
          boolean status = terminData.getType().isNormal();
          if (lastStatus == false && status == true) {
              // 上一批失败，这一批成功，说明调度有问题
              throw new SelectException(String.format("last status is rollback , but now [batchId:%d , processId:%d] is ack",
                  batchId,
                  terminData.getProcessId()));
          }
  
          if (terminData.getType().isNormal()) {
              ack(batchId);
              sendDelayStat(pipelineId, terminData.getEndTime(), terminData.getFirstTime());
          } else {
              rollback(batchId);
          }
  
          arbitrateEventService.terminEvent().ack(terminData); // 先发送对应的数据
          return status;
      }
  
      private void rollback(Long batchId) {
          notifyRollback();
          // otterSelector.rollback(batchId);
          otterSelector.rollback();// 一旦出错，rollback所有的mark，避免拿出后面的数据进行同步
      }
  
      private void ack(Long batchId) {
          canStartSelector.set(true);
          otterSelector.ack(batchId);
      }
  
      private void notifyRollback() {
          canStartSelector.set(false);
          rversion.incrementAndGet();// 变更一下版本
      }
  
      /**
       * 检查一下是否需要继续工作，因为mainstem代表一个节点是否为工作节点，一旦出现断网，另一个节点就会启用。此时会出现双节点同时工作，
       * 所以需要做一个检查
       */
      private void checkContinueWork() throws InterruptedException {
          boolean working = arbitrateEventService.mainStemEvent().check(pipelineId);
          if (!working) {
              logger.warn("mainstem is not run in this node");
              stopup(true);
          }
  
      }
  
      public void shutdown() {
          super.shutdown();
  
          if (executor != null) {
              executor.shutdownNow();
          }
  
          if (otterSelector != null && otterSelector.isStart()) {
              otterSelector.stop();
          }
      }
  
      public static class BatchTermin {
  
          private Long    batchId   = -1L;
          private Long    processId = -1L;
          private boolean needWait  = true;
  
          public BatchTermin(Long batchId, Long processId){
              this(batchId, processId, true);
          }
  
          public BatchTermin(Long batchId, boolean needWait){
              this(batchId, -1L, needWait);
          }
  
          public BatchTermin(Long batchId, Long processId, boolean needWait){
              this.batchId = batchId;
              this.processId = processId;
              this.needWait = needWait;
          }
  
          public Long getBatchId() {
              return batchId;
          }
  
          public void setBatchId(Long batchId) {
              this.batchId = batchId;
          }
  
          public Long getProcessId() {
              return processId;
          }
  
          public void setProcessId(Long processId) {
              this.processId = processId;
          }
  
          public boolean isNeedWait() {
              return needWait;
          }
  
          public void setNeedWait(boolean needWait) {
              this.needWait = needWait;
          }
  
          @Override
          public String toString() {
              return "BatchTermin [batchId=" + batchId + ", needWait=" + needWait + ", processId=" + processId + "]";
          }
  
      }
  
      private void sendDelayStat(long pipelineId, Long endTime, Long startTime) {
          DelayCount delayCount = new DelayCount();
          delayCount.setPipelineId(pipelineId);
          delayCount.setNumber(0L);// 不再统计delayNumber
          if (startTime != null && endTime != null) {
              delayCount.setTime(endTime - startTime);// 以后改造成获取数据库的sysdate/now()
          }
  
          statisticsClientService.sendResetDelayCount(delayCount);
      }
  
      private void sendDelayReset(long pipelineId) {
          long currentTime = System.currentTimeMillis();
          if (currentTime - lastResetTime > 60 * 1000) {
              // 60秒向manager推送一次配置
              lastResetTime = currentTime;
              DelayCount delayCount = new DelayCount();
              delayCount.setPipelineId(pipelineId);
              delayCount.setNumber(0L);
              long delayTime = currentTime - otterSelector.lastEntryTime();
              delayCount.setTime(delayTime);
              statisticsClientService.sendResetDelayCount(delayCount);
          }
      }
  
      // ======================= setter / getter ===================
  
      public void setOtterSelectorFactory(OtterSelectorFactory otterSelectorFactory) {
          this.otterSelectorFactory = otterSelectorFactory;
      }
  
      public void setStatisticsClientService(StatisticsClientService statisticsClientService) {
          this.statisticsClientService = statisticsClientService;
      }
  
  }
  