From 643d830939de0ba58a2cf0cdde3ec2d3dbd3b737 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 19 Nov 2025 14:13:14 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20=E5=A2=9E=E5=8A=A0BULK?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=A2=9E=E9=87=8F=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../celnet/datadump/job/DataDumpNewJob.java | 21 ++ .../datadump/service/CommonBatchService.java | 2 + .../service/DataDumpSpecialService.java | 1 + .../service/impl/CommonBatchServiceImpl.java | 352 +++++++++++++++++- .../impl/DataImportBatchServiceImpl.java | 33 +- .../datadump/util/CsvConverterUtil.java | 3 +- 6 files changed, 404 insertions(+), 8 deletions(-) diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index a47f1d5..43a999c 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -114,6 +114,27 @@ public class DataDumpNewJob { return commonBatchService.dumpBatch(param); } + /** + * 增量任务（大数据量） + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataDumpIncrementBatchJob") + public ReturnT dataDumpIncrementBatchJob(String paramStr) throws Exception { + log.info("dataDumpIncrementBatchJob execute start .................."); + SalesforceParam param = new SalesforceParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, SalesforceParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + + return commonBatchService.incrementBatch(param); + } + /** * 存量任务（BigObject） * diff --git a/src/main/java/com/celnet/datadump/service/CommonBatchService.java b/src/main/java/com/celnet/datadump/service/CommonBatchService.java index 41997f5..807fa14 100644 --- a/src/main/java/com/celnet/datadump/service/CommonBatchService.java +++ b/src/main/java/com/celnet/datadump/service/CommonBatchService.java @@ -17,4 +17,6 @@ public interface CommonBatchService { ReturnT dumpBigObjectTargetV1(SalesforceParam param) throws Exception; + ReturnT incrementBatch(SalesforceParam param) throws Exception; + } \ No newline at end of file diff --git a/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java b/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java index c74a030..97e6152 100644 --- a/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java +++ b/src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java @@ -2,6 +2,7 @@ package com.celnet.datadump.service; import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.param.DataDumpSpecialParam; +import com.sforce.async.BulkConnection; import com.sforce.soap.partner.PartnerConnection; import com.xxl.job.core.biz.model.ReturnT; diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index 5d30ebc..951d5f2 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -13,6 +13,7 @@ import com.celnet.datadump.entity.*; import com.celnet.datadump.enums.FileType; import com.celnet.datadump.global.Const; import com.celnet.datadump.global.SystemConfigCode; +import com.celnet.datadump.global.TypeCode; import com.celnet.datadump.mapper.CustomMapper; import com.celnet.datadump.param.DataDumpParam; import com.celnet.datadump.param.DataDumpSpecialParam; @@ -95,7 +96,8 @@ public class CommonBatchServiceImpl implements CommonBatchService { private DataDumpSpecialService dataDumpSpecialService; @Autowired private DataBatchHistoryService dataBatchHistoryService; - + @Autowired + private DataReportService dataReportService; // 批次处理记录数，默认为10000 private int batchProcessCount = 10000; @@ -143,6 +145,67 @@ public class CommonBatchServiceImpl implements CommonBatchService { } } + @Override + public ReturnT incrementBatch(SalesforceParam param) throws Exception { + QueryWrapper qw = new QueryWrapper<>(); + if (StringUtils.isNotBlank(param.getApi())) { + List apis = DataUtil.toIdList(param.getApi()); + qw.in("name", apis); + } + qw.eq("need_update", true) + .isNotNull("last_update_date"); + List list = dataObjectService.list(qw); + if (CollectionUtils.isEmpty(list)) { + return new ReturnT<>(500, ("表" + param.getApi() + "不存在或未开启更新")); + } + List> futures = Lists.newArrayList(); + try { + DataReport dataReport = new DataReport(); + dataReport.setType(TypeCode.INCREMENT); + dataReport.setApis(list.stream().map(DataObject::getName).collect(Collectors.joining(","))); + dataReportService.save(dataReport); + for (DataObject dataObject : list) { + //无创建时间对象 + if (!dataFieldService.hasCreatedDate(dataObject.getName())) { + DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam(); + dataDumpSpecialParam.setApi(dataObject.getName()); + Future future = getDataBatch(dataDumpSpecialParam, salesforceConnect.createBulkConnect(),dataObject); + // 等待当前所有线程执行完成 + salesforceExecutor.waitForFutures(future); + continue; + } + Future future = salesforceExecutor.execute(() -> { + try { + Date updateTime = new Date(); + SalesforceParam salesforceParam = new SalesforceParam(); + salesforceParam.setApi(dataObject.getName()); + salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate()); + salesforceParam.setType(2); + // 更新字段值不为空 按更新字段里的字段校验 + if (StringUtils.isNotBlank(dataObject.getUpdateField())) { + salesforceParam.setUpdateField(dataObject.getUpdateField()); + } + dumpIncrementDataBatch(salesforceParam); + dataObject.setLastUpdateDate(updateTime); + dataObject.setNeedUpdate( false); + dataObjectService.updateById(dataObject); + + } catch (Throwable throwable) { + log.error("salesforceExecutor error", throwable); + throw new RuntimeException(throwable); + } + }, 0, 0); + futures.add(future); + } + // 等待当前所有线程执行完成 + salesforceExecutor.waitForFutures(futures.toArray(new Future[]{})); + return ReturnT.SUCCESS; + } catch (Throwable throwable) { + salesforceExecutor.remove(futures.toArray(new Future[]{})); + throw throwable; + } + } + @Override public ReturnT dumpBigObject(SalesforceParam param) throws Exception { try { @@ -1077,6 +1140,167 @@ public class CommonBatchServiceImpl implements CommonBatchService { return bulkConnect; } + private BulkConnection dumpIncrementDataBatch(SalesforceParam param) throws Throwable { + String api = param.getApi(); + BulkConnection bulkConnect = null; + try { + DataBatchHistory dataBatchHistory = new DataBatchHistory(); + dataBatchHistory.setName(api); + dataBatchHistory.setStartDate(new Date()); + dataBatchHistory.setSyncStartDate(param.getBeginCreateDate()); + if (param.getEndCreateDate() != null) { + dataBatchHistory.setSyncEndDate(DateUtils.addSeconds(param.getEndCreateDate(), -1)); + } + dataBatchHistory.setBatch(param.getBatch()); + if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) { + log.info("NO.{} dump {}, date: {} ~ {} start", param.getBatch(), + param.getApi(), + DateFormatUtils.format(param.getBeginCreateDate(), "yyyy-MM-dd HH:mm:ss"), + DateFormatUtils.format(param.getEndCreateDate(), "yyyy-MM-dd HH:mm:ss")); + } + bulkConnect = salesforceConnect.createBulkConnect(); + // 存在isDeleted 只查询IsDeleted为false的 + if (dataFieldService.hasDeleted(param.getApi())) { + param.setIsDeleted(false); + } else { + // 不存在 过滤 + param.setIsDeleted(null); + } + + int sfData = getAllBulkV1SfIncrementData(param, bulkConnect); + + dataBatchHistory.setDbNum(sfData); + dataBatchHistory.setSfNum(sfData); + + updateDataBatch(param, dataBatchHistory); + }catch (InterruptedException e){ + return bulkConnect; + } catch (Throwable throwable) { + log.error("dataDumpJob error api:{}", api, throwable); + String type = param.getType() == 1 ? "存量" : "增量"; + String format = String.format("%s数据迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", type, api, JSON.toJSONString(param, DataDumpParam.getFilter()), throwable); + EmailUtil.send("DataDump ERROR", format); + throw throwable; + } + return bulkConnect; + } + + public Future getDataBatch(DataDumpSpecialParam param, BulkConnection bulkConnect,DataObject dataObject) { + String api = param.getApi(); + return salesforceExecutor.execute(() -> { + try { + // 检测表 不存在就生成 但不生成批次 + commonService.checkApi(api, true); + int count = 0; + boolean hasMore = true; + QueryWrapper wrapper = new QueryWrapper<>(); + wrapper.eq("api", api); + List list = dataFieldService.list(wrapper); + + Map map = Maps.newHashMap(); + SalesforceParam salesforceParam = new SalesforceParam(); + salesforceParam.setApi(api); + salesforceParam.setIdField(param.getField()); + String maxId = null; + + List fields = Lists.newArrayList(); + for (DataField field : list) { + if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){ + fields.add(field.getField()); + } + } + salesforceParam.setSelect(StringUtils.join(fields, ",")); + + while (hasMore) { + int sfData = 0; + salesforceParam.setMaxId(maxId); + salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate()); + salesforceParam.setUpdateField(dataFieldService.returnUpdateDateField(dataObject.getName())); + salesforceParam.setLimit(batchProcessCount); + map.put("param", salesforceParam); + JobInfo job = null; + try { + String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdShare", map); + + log.info("query sql: {}", sql); + XxlJobLogger.log("query sql: {}", sql); + + job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll); + log.info("创建Bulk作业成功, 作业ID: {}", job.getId()); + + BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8))); + log.info("创建批次成功, 批次ID: {}", batchInfo.getId()); + + int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo); + log.info("批次处理完成, 处理记录数: {}", completionOne); + + if (completionOne == 0){ + log.info("无更多数据, 结束处理"); + BulkUtil.closeJob(bulkConnect, job.getId()); + break; + } + + List> batchRecords = new ArrayList<>(); + + QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId()); + log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length); + + for (final String resultId : queryResultList.getResult()) { + InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId); + + CSVReader rdr = new CSVReader(resultStream); + List headers = rdr.nextRecord(); + log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size()); + + // 批量处理记录 + List record; + while ((record = rdr.nextRecord()) != null) { + Map recordMap = new HashMap<>(); + for (int i = 0; i < headers.size() && i < record.size(); i++) { + recordMap.put(headers.get(i), record.get(i)); + } + batchRecords.add(recordMap); + } + } + if (!batchRecords.isEmpty()){ + log.info("开始保存或更新数据, 记录数: {}", batchRecords.size()); + sfData = saveOrUpdate(api, batchRecords, true); + count += sfData; + maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString(); + log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}", api, sfData, maxId); + } else { + log.info("批次无数据记录"); + } + + BulkUtil.closeJob(bulkConnect, job.getId()); + log.info("关闭作业成功, 作业ID: {}", job.getId()); + }catch (InterruptedException interruptedException){ + return ; + } catch (Exception e) { + log.error("处理查询结果时异常: ", e); + throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown); + }finally { + if (sfData != batchProcessCount){ + hasMore = false; + log.info("当前批次数据不足{}条, 结束循环", batchProcessCount); + } + } + } + UpdateWrapper updateWrapper = new UpdateWrapper<>(); + updateWrapper.eq("name", param.getApi()); + updateWrapper.set("first_db_num", count); + updateWrapper.set("first_sf_num", count); + dataBatchService.update(updateWrapper); + + } catch (Throwable throwable) { + String format = String.format("数据特殊表迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", api, JSON.toJSONString(param), throwable); + EmailUtil.send("DataDump ERROR", format); + log.error("dataDumpSpecial error ", throwable); + throw new RuntimeException(throwable); + } + }, 0, 0); + } + public Future getData(DataDumpSpecialParam param, BulkConnection bulkConnect) { String api = param.getApi(); return salesforceExecutor.execute(() -> { @@ -1191,6 +1415,132 @@ public class CommonBatchServiceImpl implements CommonBatchService { }, 0, 0); } + private int getAllBulkV1SfIncrementData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable { + + boolean hasMore = true; + QueryWrapper wrapper = new QueryWrapper<>(); + String api = param.getApi(); + wrapper.eq("api", api); + List list = dataFieldService.list(wrapper); + + Map 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; + + List fields = Lists.newArrayList(); + for (DataField field : list) { + if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){ + fields.add(field.getField()); + } + } + DataObject dataObject = dataObjectService.getById(api); + if (dataObject != null && StringUtils.isNotBlank(dataObject.getExtraField())) { + fields.addAll(Arrays.asList(StringUtils.split(dataObject.getExtraField().replaceAll(StringUtils.SPACE, StringUtils.EMPTY), ","))); + } + param.setSelect(StringUtils.join(fields, ",")); + + log.info("开始处理API: {}, 类型: {}", api, param.getType() == 1 ? "按创建时间" : "按修改时间"); + + while (hasMore) { + int sfData = 0; + param.setMaxId(maxId); + param.setLimit(batchProcessCount); + // 获取创建时间 + param.setTimestamp(lastCreatedDate); + map.put("param", param); + String sql; + JobInfo job = null; + try { + if (param.getType() == 1) { + // type 1 按创建时间纬度 + sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.list", map); + } else { + // type 2 按修改时间纬度 + sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listByModifyTime", map); + } + 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()); + + BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8))); + log.info("创建批次成功, 批次ID: {}", batchInfo.getId()); + + int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo); + log.info("批次处理完成, 处理记录数: {}", completionOne); + + dataLogService.save(dataLog); + + if (completionOne == 0){ + log.info("无更多数据, 结束处理"); + BulkUtil.closeJob(bulkConnect, job.getId()); + break; + } + + List> 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); + + for (final String resultId : queryResultList.getResult()) { + InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId); + + CSVReader rdr = new CSVReader(resultStream); + List headers = rdr.nextRecord(); + log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size()); + + // 批量处理记录 + List record; + while ((record = rdr.nextRecord()) != null) { + Map recordMap = new HashMap<>(); + for (int i = 0; i < headers.size() && i < record.size(); i++) { + recordMap.put(headers.get(i), record.get(i)); + } + batchRecords.add(recordMap); + } + } + + dataLogService.save(dataLog1); + + if (!batchRecords.isEmpty()){ + log.info("开始保存或更新数据, 记录数: {}", batchRecords.size()); + sfData = saveOrUpdate(api, batchRecords, true); + count += sfData; + maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString(); + lastCreatedDate = DateUtils.addHours(DateUtils.parseDate((String) batchRecords.get(batchRecords.size() - 1).get("CreatedDate"), Const.SF_DATE_FORMAT), 8); + log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}, 当前批次最后创建时间:{}", api, sfData, maxId, lastCreatedDate); + } else { + log.info("批次无数据记录"); + } + + BulkUtil.closeJob(bulkConnect, job.getId()); + batch ++; + log.info("关闭作业成功, 作业ID: {}", job.getId()); + }catch (InterruptedException interruptedException){ + return 0; + } catch (Exception e) { + log.error("处理查询结果时异常: ", e); + throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown); + }finally { + if (sfData != batchProcessCount){ + hasMore = false; + log.info("当前批次数据不足{}条, 结束循环", batchProcessCount); + } + } + } + log.info("API: {} 数据处理完成, 总处理记录数: {}", api, count); + + return count; + } + private int getAllBulkV1SfData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable { boolean hasMore = true; diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java index 3440ca9..da1d562 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -289,8 +289,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { dbQw.and(wrapper -> wrapper.eq("is_createable", 1) .eq("is_nillable", 0) .eq("is_defaulted_on_create", 0) - .or().eq("field", "CreatedDate") - .or().eq("field", "CreatedById")); + .or().eq("field", "Id") + .or().eq("field", "CreatedDate") + .or().eq("field", "CreatedById")); List list = dataFieldService.list(dbQw); String beginDateStr = null; @@ -437,6 +438,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceInsertJob = null; + if (insertList.isEmpty()){ + log.info("当前批次无可获取SFID的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api); + } + try { DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF"); @@ -447,6 +452,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { 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"); @@ -540,19 +546,23 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { int retryCount = 0; final int maxRetries = 3; - while (retryCount <= maxRetries) { + while (retryCount < maxRetries) { try { - rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId())); + InputStream batchResultStream = connection.getBatchResultStream(job.getId(), b.getId()); + rdr = new CSVReader(batchResultStream); break; // 成功执行，跳出重试循环 }catch (Exception e) { retryCount++; log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api); try { - Thread.sleep(1000 * retryCount); // 简单的退避策略 + Thread.sleep(5000 * retryCount); // 简单的退避策略 } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while retrying", ie); } + if (maxRetries <= retryCount){ + return 0; + } } } @@ -964,6 +974,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceUpdateJob = null; + if (updateList.isEmpty()){ + log.info("当前批次无可更新的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api); + } + try { DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF"); @@ -1448,8 +1462,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceInsertJob = null; dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + if (insertList.isEmpty()){ + log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api); + } + try { DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF"); @@ -1636,6 +1655,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceInsertJob = null; + if (insertList.isEmpty()){ + log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api); + } + try { DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF"); diff --git a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java index 85d222f..b859e9c 100644 --- a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java +++ b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java @@ -259,8 +259,7 @@ public class CsvConverterUtil { if (isEditable){ header = Stream.concat( - fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable())) - .map(DataField::getField), + fields.stream().map(DataField::getField), Stream.of("old_owner_id__c", "old_sfdc_id__c") ).toArray(String[]::new); }else {