refactor: process upstream messages individually if push fails

events/upstream_push.go

@@ -150,12 +150,25 @@ func (t *pushToUpstreamEventHandler) Run(ctx *api.Context, events []api.Event) [
 	}
 
 	upstreamMsg.ApplyLabels(t.conf.LabelsMap())
-	if err := upstream.Push(ctx, t.conf, upstreamMsg); err != nil {
+	err := upstream.Push(ctx, t.conf, upstreamMsg)
+	if err == nil {
+		return failedEvents
+	}
+
+	if len(events) == 1 {
 		errMsg := fmt.Errorf("failed to push to upstream: %w", err)
 		failedEvents = append(failedEvents, addErrorToFailedEvents(events, errMsg)...)
+		return failedEvents
+	} else {
+		// Error encountered while pushing could be an SQL or Application error
+		// Since we do not know which event in the bulk is failing
+		// Process each event individually since upsteam.Push is idempotent
+		var failedIndividualEvents []api.Event
+		for _, e := range events {
+			failedIndividualEvents = append(failedIndividualEvents, t.Run(ctx, []api.Event{e})...)
+		}
+		return failedIndividualEvents
 	}
-
-	return failedEvents
 }
 
 type GroupedPushEvents struct {