【feat】 bulk增加日志

(cherry picked from commit 93d56c627d)

src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java

@@ -1119,6 +1119,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         Map<String, Object> map = Maps.newHashMap();
         String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField();
         int count = 0;
+        int batch = 0;
         Date lastCreatedDate = null;
         String maxId = null;
 
@@ -1156,6 +1157,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 log.info("query sql: {}", sql);
                 XxlJobLogger.log("query sql: {}", sql);
 
+                DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
+
                 job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
                 log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
 
@@ -1165,6 +1168,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
                 log.info("批次处理完成, 处理记录数: {}", completionOne);
 
+                dataLogService.save(dataLog);
+
                 if (completionOne == 0){
                     log.info("无更多数据, 结束处理");
                     BulkUtil.closeJob(bulkConnect, job.getId());
@@ -1173,6 +1178,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
 
                 List<Map<String, Object>> batchRecords = new ArrayList<>();
 
+                DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
+
                 QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
                 log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
 
@@ -1194,6 +1201,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                     }
                 }
 
+                dataLogService.save(dataLog1);
+
                 if (!batchRecords.isEmpty()){
                     log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
                     sfData = saveOrUpdate(api, batchRecords, true);
@@ -1206,6 +1215,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 }
 
                 BulkUtil.closeJob(bulkConnect, job.getId());
+                batch ++;
                 log.info("关闭作业成功, 作业ID: {}", job.getId());
             }catch (InterruptedException interruptedException){
                 return 0;

src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java

@@ -21,9 +21,6 @@ import com.celnet.datadump.util.DataUtil;
 import com.celnet.datadump.util.EmailUtil;
 import com.google.common.collect.Lists;
 import com.sforce.async.*;
-import com.sforce.soap.partner.PartnerConnection;
-import com.sforce.soap.partner.SaveResult;
-import com.sforce.soap.partner.sobject.SObject;
 import com.xxl.job.core.biz.model.ReturnT;
 import com.xxl.job.core.log.XxlJobLogger;
 import com.xxl.job.core.util.DateUtil;
@@ -81,6 +78,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
     @Autowired
     private LinkConfigService linkConfigService;
 
+    @Autowired
+    private DataLogService dataLogService;
+
     /**
      * Insert入口
      */
@@ -326,6 +326,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
+
             List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null  and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
             int size = data.size();
 
@@ -423,17 +425,30 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     break;
                 }
             }
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
+
             JobInfo salesforceInsertJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
+
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
                 BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
+                dataLog2.setEndTime(new Date());
+                dataLogService.save(dataLog2);
+
                 TimeUnit.MILLISECONDS.sleep(1);
 
             } catch (InterruptedException interruptedException){
@@ -512,10 +527,15 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
      */
     public int checkInsertResultsOne(BulkConnection connection, JobInfo job,
                               BatchInfo b,String api,String[] ids)
-            throws AsyncApiException, IOException {
+        throws Exception {
         int index = 0;
-        CSVReader rdr =
+        CSVReader rdr = null;
-                new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
+        int retryCount = 0;
+        final int maxRetries = 3;
+        
+        while (retryCount <= maxRetries) {
+            try {
+                rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
                 List<String> resultHeader = rdr.nextRecord();
                 int resultCols = resultHeader.size();
 
@@ -543,10 +563,22 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         linkMap1.put("value", JSON.toJSONString(error));
                         maps.add(linkMap1);
                         customMapper.updateById(api, maps, ids[index]);
-                log.error("Id:{},saveResults: {}",ids[index], error);
+                        log.info("Id:{},saveResults: {}",ids[index], error);
                     }
                     index ++;
                 }
+                break; // 成功执行，跳出重试循环
+            }catch (Exception e) {
+                retryCount++;
+                log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api);
+                try {
+                    Thread.sleep(1000 * retryCount); // 简单的退避策略
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new IOException("Interrupted while retrying", ie);
+                }
+            }
+        }
         return index;
     }
 
@@ -826,6 +858,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
+
             List<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000");
 
             log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -911,10 +945,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 }
                 updateList.add(account);
             }
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
 
             JobInfo salesforceUpdateJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
 
                 salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update);
 
@@ -922,10 +959,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 int updatedCount = checkUpdateResultsOne(bulkConnection, salesforceUpdateJob, batchInfo,api);
 
                 sfNum = sfNum + updatedCount;
 
+                dataLog2.setEndTime(new Date());
+                dataLogService.save(dataLog2);
+
                 TimeUnit.MILLISECONDS.sleep(1);
 
             } catch (InterruptedException interruptedException){
@@ -1100,7 +1145,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 update.setDataLock(1);
                 dataObjectService.updateById(update);
 
-                if (api.endsWith("Share") || "GroupMember".equals(api)){
+                if (dataFieldService.hasCreatedDate(api)){
                     insertSingleShareData(api,bulkConnection);
                     continue;
                 }
@@ -1248,8 +1293,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
         }
 
+        String sql = "";
+        String sql2 = "";
+
+        if (api.endsWith("Share")){
+            sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null ";
+            sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 8000";
+        }else {
+            sql = "where new_id is null ";
+            sql2 = "new_id is null limit 8000";
+        }
+
         //表内数据总量
-        Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
+        Integer count = customMapper.countBySQL(api, sql);
         if (count == 0) {
             return;
         }
@@ -1277,7 +1333,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
-            List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null  and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
+
+            List<JSONObject> data = customMapper.listJsonObject("*", api, sql2);
             int size = data.size();
 
             log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -1374,15 +1432,29 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             JobInfo salesforceInsertJob = null;
 
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
+
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
+
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
+                dataLog2.setEndTime(new Date());
+                dataLogService.save(dataLog2);
+
             } catch (Exception e) {
                 log.error("createdNewIdBatch error api:{}", api, e);
                 throw e;
@@ -1448,6 +1520,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
+
             List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') order by Id asc limit " + i * 10000 + ",10000");
             int size = data.size();
 
@@ -1523,9 +1597,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 }
             }
 
+            dataLog.setEndTime(new Date());
+
+            dataLogService.save(dataLog);
+
             JobInfo salesforceInsertJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
 
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
@@ -1533,8 +1612,17 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
                 BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
 
+                dataLog1.setEndTime(new Date());
+
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
 
+                dataLog2.setEndTime(new Date());
+
+                dataLogService.save(dataLog2);
             } catch (Exception e) {
                 log.error("insertSingleShareData error api:{}", api, e);
                 throw e;

src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java

@@ -3663,7 +3663,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
 
                             String value = String.valueOf(data.get(j - 1).get(field));
                             //根据旧sfid查找引用对象新sfid
-                            if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) {
+                            if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) {
                                 continue;
                             } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {
 

src/main/java/com/celnet/datadump/util/DataUtil.java

@@ -511,6 +511,7 @@ public class DataUtil {
 
     public static Object localDataToSfData(String fieldType, String data) throws ParseException {
         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
+        sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
         SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
         Date date;
         //date转Calendar类型