content: package com.alibaba.otter.canal.example.kafka;
  
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.util.Assert;
  
  import com.alibaba.otter.canal.client.kafka.KafkaOffsetCanalConnector;
  import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
  
  /**
   * KafkaOffsetCanalConnector 使用示例
   * <p>
   * KafkaOffsetCanalConnector 与 KafkaCanalConnector 的另一区别是 auto.offset.reset
   * 默认值不同；
   * </p>
   * <p>
   * KafkaOffsetCanalConnector 默认为
   * earliest；canal-kafka-client重启后从未被消费的记录开始拉取消息，同时提供了修改 auto.offset.reset 的方法
   * setAutoOffsetReset
   * </p>
   *
   * @author panjianping @ 2018-12-18
   * @version 1.1.3
   */
  public class CanalKafkaOffsetClientExample {
  
      protected final static Logger           logger  = LoggerFactory.getLogger(CanalKafkaOffsetClientExample.class);
  
      private KafkaOffsetCanalConnector       connector;
  
      private static volatile boolean         running = false;
  
      private Thread                          thread  = null;
  
      private Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", e);
  
      public CanalKafkaOffsetClientExample(String servers, String topic, Integer partition, String groupId){
          connector = new KafkaOffsetCanalConnector(servers, topic, partition, groupId, false);
      }
  
      public static void main(String[] args) {
          try {
              final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample(
                  AbstractKafkaTest.servers,
                  AbstractKafkaTest.topic,
                  AbstractKafkaTest.partition,
                  AbstractKafkaTest.groupId);
              logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
              kafkaCanalClientExample.start();
              logger.info("## the canal kafka consumer is running now ......");
              Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                  try {
                      logger.info("## stop the kafka consumer");
                      kafkaCanalClientExample.stop();
                  } catch (Throwable e) {
                      logger.warn("##something goes wrong when stopping kafka consumer:", e);
                  } finally {
                      logger.info("## kafka consumer is down.");
                  }
              }));
              while (running)
                  ;
          } catch (Throwable e) {
              logger.error("## Something goes wrong when starting up the kafka consumer:", e);
              System.exit(0);
          }
      }
  
      public void start() {
          Assert.notNull(connector, "connector is null");
          thread = new Thread(this::process);
          thread.setUncaughtExceptionHandler(handler);
          thread.start();
          running = true;
      }
  
      public void stop() {
          if (!running) {
              return;
          }
          running = false;
          if (thread != null) {
              try {
                  thread.join();
              } catch (InterruptedException e) {
                  // ignore
              }
          }
      }
  
      private void process() {
          while (!running) {
              try {
                  Thread.sleep(1000);
              } catch (InterruptedException e) {
              }
          }
  
          while (running) {
              try {
                  // 修改 AutoOffsetReset 的值，默认（earliest）
                  // connector.setAutoOffsetReset(null);
                  connector.connect();
                  connector.subscribe();
                  // 消息起始偏移地址
                  long offset = -1;
                  // 错误次数
                  int errorCount = 0;
                  while (running) {
                      try {
                          // 错误重试次数超过3次后，每30秒递增重试
                          if (errorCount > 2) {
                              Thread.sleep((errorCount - 2) * 1000 * 30);
                          }
  
                          List<KafkaMessage> messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS, offset); // 获取message
                          if (messages == null) {
                              continue;
                          }
                          for (KafkaMessage message : messages) {
                              long batchId = message.getId();
                              int size = message.getEntries().size();
  
                              if (batchId == -1 || size == 0) {
                                  continue;
                              }
  
                              // 记录第一条消息的offset，用于处理数据异常时重新从此位置获取消息
                              if (offset < 0) {
                                  offset = message.getOffset();
                              }
  
                              // printSummary(message, batchId, size);
                              // printEntry(message.getEntries());
                              logger.info(message.toString());
                          }
  
                          connector.ack(); // 提交确认
                          // 还原offset
                          offset = -1;
                          errorCount = 0;
                      } catch (Exception e) {
                          errorCount++;
                          logger.error(e.getMessage(), e);
                          if (errorCount == 3) {
                              // 重试3次后发送邮件提醒异常
                              // mailService.sendMail("同步数据异常，请及时处理", "错误消息");
                          }
                      }
                  }
              } catch (Exception e) {
                  logger.error(e.getMessage(), e);
              }
          }
  
          connector.unsubscribe();
          connector.disconnect();
      }
  }
  