fix: update error parsing of eloqua

router/batchrouter/asyncdestinationmanager/common/common.go

@@ -149,12 +149,12 @@ func GetBatchRouterConfigInt64(key, destType string, defaultValue int64) int64 {
  return config.GetInt64("BatchRouter."+key, defaultValue)
 }
 
-func GetBatchRouterConfigBool(key, destType string, defaultValue bool) bool {
+func GetBatchRouterConfigStringMap(key, destType string, defaultValue []string) []string {
  destOverrideFound := config.IsSet("BatchRouter." + destType + "." + key)
  if destOverrideFound {
- return config.GetBool("BatchRouter."+destType+"."+key, defaultValue)
+ return config.GetStringSlice("BatchRouter."+destType+"."+key, defaultValue)
  }
- return config.GetBool("BatchRouter."+key, defaultValue)
+ return config.GetStringSlice("BatchRouter."+key, defaultValue)
 }
 
 /*

router/batchrouter/asyncdestinationmanager/eloqua/bulk_uploader.go

@@ -51,6 +51,8 @@ func (b *EloquaBulkUploader) Upload(asyncDestStruct *common.AsyncDestinationStru
  return b.createAsyncUploadErrorOutput("got error while fetching fields. ", err, destination.ID, asyncDestStruct)
  }
 
+ uniqueKeys := getUniqueKeys(eloquaFields)
+ b.uniqueKeys = uniqueKeys
  uploadJobInfo := JobInfo{
  fileSizeLimit: b.fileSizeLimit,
  importingJobs: asyncDestStruct.ImportingJobIDs,
@@ -227,7 +229,7 @@ func (b *EloquaBulkUploader) GetUploadStats(UploadStatsInput common.GetUploadSta
  DynamicPart: UploadStatsInput.WarningJobURLs,
  Authorization: b.authorization,
  }
- eventStatMetaWithRejectedSucceededJobs, err := parseRejectedData(&checkRejectedData, UploadStatsInput.ImportingList, b.service, b.jobToCSVMap)
+ eventStatMetaWithRejectedSucceededJobs, err := parseRejectedData(&checkRejectedData, UploadStatsInput.ImportingList, b)
  if err != nil {
  b.logger.Error("Error while parsing rejected data", err)
  return common.GetUploadStatsResponse{

router/batchrouter/asyncdestinationmanager/eloqua/manager.go

@@ -44,13 +44,14 @@ func NewManager(destination *backendconfig.DestinationT) (*EloquaBulkUploader, e
 
 func NewEloquaBulkUploader(destinationName, authorization, baseEndpoint string, eloqua EloquaService) *EloquaBulkUploader {
  return &EloquaBulkUploader{
- destName: destinationName,
- logger: logger.NewLogger().Child("batchRouter").Child("AsyncDestinationManager").Child("Eloqua").Child("EloquaBulkUploader"),
- authorization: authorization,
- baseEndpoint: baseEndpoint,
- fileSizeLimit: common.GetBatchRouterConfigInt64("MaxUploadLimit", destinationName, 32*bytesize.MB),
- eventsLimit: common.GetBatchRouterConfigInt64("MaxEventsLimit", destinationName, 1000000),
- service: eloqua,
- jobToCSVMap: map[int64]int64{},
+ destName: destinationName,
+ logger: logger.NewLogger().Child("batchRouter").Child("AsyncDestinationManager").Child("Eloqua").Child("EloquaBulkUploader"),
+ authorization: authorization,
+ baseEndpoint: baseEndpoint,
+ fileSizeLimit: common.GetBatchRouterConfigInt64("MaxUploadLimit", destinationName, 32*bytesize.MB),
+ eventsLimit: common.GetBatchRouterConfigInt64("MaxEventsLimit", destinationName, 1000000),
+ successStatusCode: common.GetBatchRouterConfigStringMap("SuccessStatusCode", destinationName, []string{"ELQ-00040"}),
+ service: eloqua,
+ jobToCSVMap: map[int64]int64{},
  }
 }

router/batchrouter/asyncdestinationmanager/eloqua/types.go

@@ -19,14 +19,16 @@ type EloquaService interface {
 }
 
 type EloquaBulkUploader struct {
- destName string
- logger logger.Logger
- authorization string
- baseEndpoint string
- fileSizeLimit int64
- eventsLimit int64
- service EloquaService
- jobToCSVMap map[int64]int64
+ destName string
+ logger logger.Logger
+ authorization string
+ baseEndpoint string
+ fileSizeLimit int64
+ eventsLimit int64
+ service EloquaService
+ jobToCSVMap map[int64]int64
+ uniqueKeys []string
+ successStatusCode []string
 }
 type DestinationConfig struct {
  CompanyName string `json:"companyName"`

router/batchrouter/asyncdestinationmanager/eloqua/utils.go

@@ -93,7 +93,12 @@ func createCSVFile(fields []string, file *os.File, uploadJobInfo *JobInfo, jobId
  }
  var values []string
  for _, field := range fields {
- values = append(values, data.Message.Data[field].(string))
+ val, ok := data.Message.Data[field].(string)
+ if !ok || val == "null" {
+ values = append(values, "")
+ } else {
+ values = append(values, val)
+ }
  }
  fileInfo, err := csvFile.Stat()
  if err != nil {
@@ -158,12 +163,12 @@ func generateErrorString(item RejectedItem) string {
  return item.StatusCode + " : " + item.Message + " " + invalidItems
 }
 
-func parseRejectedData(data *HttpRequestData, importingList []*jobsdb.JobT, service EloquaService, jobToCSVMap map[int64]int64) (*common.EventStatMeta, error) {
+func parseRejectedData(data *HttpRequestData, importingList []*jobsdb.JobT, eloqua *EloquaBulkUploader) (*common.EventStatMeta, error) {
  jobIDs := []int64{}
  for _, job := range importingList {
  jobIDs = append(jobIDs, job.JobID)
  }
- rejectResponse, err := service.CheckRejectedData(data)
+ rejectResponse, err := eloqua.service.CheckRejectedData(data)
  if err != nil {
  return nil, err
  }
@@ -179,14 +184,19 @@ func parseRejectedData(data *HttpRequestData, importingList []*jobsdb.JobT, serv
  offset = i * 1000
  data.Offset = offset
  if offset != 0 {
- rejectResponse, err = service.CheckRejectedData(data)
+ rejectResponse, err = eloqua.service.CheckRejectedData(data)
  if err != nil {
  return nil, err
  }
  }
  for _, val := range rejectResponse.Items {
- failedJobIDs = append(failedJobIDs, jobToCSVMap[val.RecordIndex])
- failedReasons[jobToCSVMap[val.RecordIndex]] = generateErrorString(val)
+ uniqueInvalidFields := lo.Intersect(eloqua.uniqueKeys, val.InvalidFields)
+ successStatusCode := lo.Intersect(eloqua.successStatusCode, []string{val.StatusCode})
+ if len(successStatusCode) != 0 && len(uniqueInvalidFields) == 0 {
+ continue
+ }
+ failedJobIDs = append(failedJobIDs, eloqua.jobToCSVMap[val.RecordIndex])
+ failedReasons[eloqua.jobToCSVMap[val.RecordIndex]] = generateErrorString(val)
  }
  }
  }
@@ -214,3 +224,13 @@ func parseFailedData(syncId string, importingList []*jobsdb.JobT) *common.EventS
  }
  return &eventStatMeta
 }
+
+func getUniqueKeys(eloquaFields *Fields) []string {
+ uniqueKeys := []string{}
+ for _, item := range eloquaFields.Items {
+ if item.HasUniquenessConstraint {
+ uniqueKeys = append(uniqueKeys, item.InternalName)
+ }
+ }
+ return uniqueKeys
+}