content: package com.alibaba.otter.canal.client.adapter.kudu.service;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.kudu.client.KuduException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.fastjson2.JSON;
  import com.alibaba.fastjson2.JSONWriter.Feature;
  import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
  import com.alibaba.otter.canal.client.adapter.kudu.support.KuduTemplate;
  import com.alibaba.otter.canal.client.adapter.support.Dml;
  
  /**
   * @author liuyadong
   * @description kudu实时同步
   */
  public class KuduSyncService {
  
      private static Logger logger = LoggerFactory.getLogger(KuduSyncService.class);
  
      private KuduTemplate  kuduTemplate;
  
      // 源库表字段类型缓存: instance.schema.table -> <columnName, jdbcType>
      // private Map<String, Map<String, Integer>> columnsTypeCache = new
      // ConcurrentHashMap<>();
  
      public KuduSyncService(KuduTemplate kuduTemplate){
          this.kuduTemplate = kuduTemplate;
      }
  
      // public Map<String, Map<String, Integer>> getColumnsTypeCache() {
      // return columnsTypeCache;
      // }
  
      /**
       * 同步事件处理
       *
       * @param config
       * @param dml
       */
      public void sync(KuduMappingConfig config, Dml dml) {
          if (config != null) {
              String type = dml.getType();
              if (type != null && type.equalsIgnoreCase("INSERT")) {
                  insert(config, dml);
              } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
                  upsert(config, dml);
              } else if (type != null && type.equalsIgnoreCase("DELETE")) {
                  delete(config, dml);
              }
              if (logger.isDebugEnabled()) {
                  logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
              }
          }
      }
  
      /**
       * 删除事件
       *
       * @param config
       * @param dml
       */
      private void delete(KuduMappingConfig config, Dml dml) {
          KuduMappingConfig.KuduMapping kuduMapping = config.getKuduMapping();
          String configTable = kuduMapping.getTable();
          String configDatabase = kuduMapping.getDatabase();
          String table = dml.getTable();
          String database = dml.getDatabase();
          if (configTable.equals(table) && configDatabase.equals(database)) {
              List<Map<String, Object>> data = dml.getData();
              if (data == null || data.isEmpty()) {
                  return;
              }
              // 判定主键映射
              String pkId = "";
              Map<String, String> targetPk = kuduMapping.getTargetPk();
              for (Map.Entry<String, String> entry : targetPk.entrySet()) {
                  String mysqlID = entry.getKey().toLowerCase();
                  String kuduID = entry.getValue();
                  if (kuduID == null) {
                      pkId = mysqlID;
                  } else {
                      pkId = kuduID;
                  }
              }
              // 切割联合主键
              List<String> pkIds = Arrays.asList(pkId.split(","));
              try {
                  int idx = 1;
                  boolean completed = false;
                  List<Map<String, Object>> dataList = new ArrayList<>();
  
                  for (Map<String, Object> item : data) {
                      Map<String, Object> primaryKeyMap = new HashMap<>();
                      for (Map.Entry<String, Object> entry : item.entrySet()) {
                          String columnName = entry.getKey().toLowerCase();
                          Object value = entry.getValue();
                          if (pkIds.contains(columnName)) {
                              primaryKeyMap.put(columnName, value);
                          }
                      }
                      dataList.add(primaryKeyMap);
                      idx++;
                      if (idx % kuduMapping.getCommitBatch() == 0) {
                          kuduTemplate.delete(kuduMapping.getTargetTable(), dataList);
                          dataList.clear();
                          completed = true;
                      }
                  }
                  if (!completed) {
                      kuduTemplate.delete(kuduMapping.getTargetTable(), dataList);
                  }
              } catch (KuduException e) {
                  logger.error(e.getMessage());
                  logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
              }
          }
      }
  
      /**
       * 更新插入事件
       *
       * @param config
       * @param dml
       */
      private void upsert(KuduMappingConfig config, Dml dml) {
          KuduMappingConfig.KuduMapping kuduMapping = config.getKuduMapping();
          String configTable = kuduMapping.getTable();
          String configDatabase = kuduMapping.getDatabase();
          String table = dml.getTable();
          String database = dml.getDatabase();
          if (configTable.equals(table) && configDatabase.equals(database)) {
              List<Map<String, Object>> data = dml.getData();
              if (data == null || data.isEmpty()) {
                  return;
              }
              try {
                  int idx = 1;
                  boolean completed = false;
                  List<Map<String, Object>> dataList = new ArrayList<>();
  
                  for (Map<String, Object> entry : data) {
                      dataList.add(entry);
                      idx++;
                      if (idx % kuduMapping.getCommitBatch() == 0) {
                          kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList);
                          dataList.clear();
                          completed = true;
                      }
                  }
                  if (!completed) {
                      kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList);
                  }
              } catch (KuduException e) {
                  logger.error(e.getMessage());
                  logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
              }
          }
  
      }
  
      /**
       * 插入事件
       *
       * @param config
       * @param dml
       */
      private void insert(KuduMappingConfig config, Dml dml) {
          KuduMappingConfig.KuduMapping kuduMapping = config.getKuduMapping();
          String configTable = kuduMapping.getTable();
          String configDatabase = kuduMapping.getDatabase();
          String table = dml.getTable();
          String database = dml.getDatabase();
          if (configTable.equals(table) && configDatabase.equals(database)) {
              List<Map<String, Object>> data = dml.getData();
              if (data == null || data.isEmpty()) {
                  return;
              }
              try {
                  int idx = 1;
                  boolean completed = false;
                  List<Map<String, Object>> dataList = new ArrayList<>();
  
                  for (Map<String, Object> entry : data) {
                      dataList.add(entry);
                      idx++;
                      if (idx % kuduMapping.getCommitBatch() == 0) {
                          kuduTemplate.insert(kuduMapping.getTargetTable(), dataList);
                          dataList.clear();
                          completed = true;
                      }
                  }
                  if (!completed) {
                      kuduTemplate.insert(kuduMapping.getTargetTable(), dataList);
                  }
              } catch (KuduException e) {
                  logger.error(e.getMessage());
                  logger.error("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
              }
          }
      }
  
  }
  