fix: include commas in the profile size, handle failure case of profile more than 500kB in size

router/batchrouter/asyncdestinationmanager/klaviyobulkupload/klaviyobulkupload.go

@@ -24,10 +24,11 @@
 )
 
 const (
-	KlaviyoAPIURL       = "https://a.klaviyo.com/api/profile-bulk-import-jobs/"
-	BATCHSIZE           = 10000
-	MAXPAYLOADSIZE      = 4900000
-	IMPORT_ID_SEPARATOR = ":"
+	KlaviyoAPIURL         = "https://a.klaviyo.com/api/profile-bulk-import-jobs/"
+	BATCHSIZE             = 10000
+	MAXALLOWEDPROFILESIZE = 512000
+	MAXPAYLOADSIZE        = 4900000
+	IMPORT_ID_SEPARATOR   = ":"
 )
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
@@ -80,7 +81,7 @@
 			return nil, fmt.Errorf("failed to marshal profile: %w", err)
 		}
 
-		profileSize := len(profileJSON)
+		profileSize := len(profileJSON) + 1 // +1 for comma character
 
 		if (chunkSize+profileSize >= maxBytes || len(chunk) == maxElements) && len(chunk) > 0 {
 			chunks = append(chunks, chunk)
@@ -308,6 +309,8 @@
 	startTime := time.Now()
 	destination := asyncDestStruct.Destination
 	var failedJobs []int64
+	var abortedJobs []int64
+	var abortReason string
 	var successJobs []int64
 	filePath := asyncDestStruct.FileName
 	importingJobIDs := asyncDestStruct.ImportingJobIDs
@@ -317,6 +320,7 @@
 	statLabels := stats.Tags{
 		"module":   "batch_router",
 		"destType": destType,
+		"destID":   destinationID,
 	}
 	file, err := os.Open(filePath)
 	if err != nil {
@@ -325,6 +329,7 @@
 	defer file.Close()
 	var combinedProfiles []Profile
 	scanner := bufio.NewScanner(file)
+	profileSizeStat := kbu.statsFactory.NewTaggedStat("profile_size", stats.HistogramType, statLabels)
 	for scanner.Scan() {
 		var input Input
 		line := scanner.Text()
@@ -333,6 +338,15 @@
 			return kbu.generateKlaviyoErrorOutput("Error while parsing JSON.", err, importingJobIDs, destinationID)
 		}
 		profileStructure := kbu.ExtractProfile(input)
+		// if profileStructure length is more than 5 mb, throw an error
+		profileStructureJSON, _ := json.Marshal(profileStructure)
+		profileSize := float64(len(profileStructureJSON))
+		profileSizeStat.Observe(float64(profileSize)) // Record the size in the histogram
+		if float64(len(profileStructureJSON)) >= MAXALLOWEDPROFILESIZE {
+			abortReason = "Error while marshaling profiles. The profile size exceeds Klaviyo's limit of 500 kB for a single profile."
+			abortedJobs = append(abortedJobs, int64(input.Metadata.JobID))
+			continue
+		}
 		combinedProfiles = append(combinedProfiles, profileStructure)
 	}
 
@@ -379,12 +393,14 @@
 		if resp.StatusCode != 202 {
 			failedJobs = append(failedJobs, importingJobIDs[idx])
 			kbu.logger.Error("Got non 202 as statusCode.", errors.New(string(bodyBytes)))
+			continue
 		}
 		var uploadresp UploadResp
 		uploadRespErr := json.Unmarshal((bodyBytes), &uploadresp)
 		if uploadRespErr != nil {
 			failedJobs = append(failedJobs, importingJobIDs[idx])
 			kbu.logger.Error("Error while unmarshaling response.", uploadRespErr)
+			continue
 		}
 		importIds = append(importIds, uploadresp.Data.Id)
 	}
@@ -400,6 +416,8 @@
 	return common.AsyncUploadOutput{
 		ImportingParameters: importParameters,
 		FailedJobIDs:        failedJobs,
+		AbortJobIDs:         abortedJobs,
+		AbortReason:         abortReason,
 		FailedCount:         len(failedJobs),
 		ImportingJobIDs:     successJobs,
 		DestinationID:       destination.ID,