addressed comments

Author: mihir20

integration_test/docker_test/docker_test.go

@@ -556,10 +556,7 @@ func sendEventsToGateway(t *testing.T) {
 			}
 		]
 	}`)
-	sendEvent(t, payloadRetlWebhook, "retl", writeKey,
-		withHeader("X-Rudder-Source-Id", "xxxyyyzzEaEurW247ad9WYZLUyk"),
-		withHeader("X-Rudder-Destination-Id", "xxxyyyzzP9kQfzOoKd1tuxchYAG"),
-		withUrlPath("/internal/v1/retl"))
+	sendRETL(t, payloadRetlWebhook, "xxxyyyzzEaEurW247ad9WYZLUyk", "xxxyyyzzP9kQfzOoKd1tuxchYAG")
 
 	payloadRetlKafka := strings.NewReader(`{
 		"batch":
@@ -653,10 +650,7 @@ func sendEventsToGateway(t *testing.T) {
 			}
 		]
 	}`)
-	sendEvent(t, payloadRetlKafka, "retl", writeKey,
-		withHeader("X-Rudder-Source-Id", "xxxyyyzzEaEurW247ad9WYZLUyk"),
-		withHeader("X-Rudder-Destination-Id", "xxxyyyzzhyrw8v0CrTMrDZ4ovej"),
-		withUrlPath("/internal/v1/retl"))
+	sendRETL(t, payloadRetlKafka, "xxxyyyzzEaEurW247ad9WYZLUyk", "xxxyyyzzhyrw8v0CrTMrDZ4ovej")
 }
 
 func blockOnHold(t *testing.T) {
@@ -722,20 +716,46 @@ func sendPixelEvents(t *testing.T, writeKey string) {
 	}
 }
 
-func withHeader(key, value string) func(r *http.Request) {
-	return func(req *http.Request) {
-		req.Header.Add(key, value)
+func sendRETL(t *testing.T, payload *strings.Reader, sourceID, DestinationID string) {
+	t.Helper()
+	t.Logf("Sending rETL Event")
+
+	var (
+		httpClient = &http.Client{}
+		method     = "POST"
+		url        = fmt.Sprintf("http://localhost:%s/internal/v1/retl", httpPort)
+	)
+
+	req, err := http.NewRequest(method, url, payload)
+	if err != nil {
+		t.Logf("sendEvent error: %v", err)
+		return
 	}
-}
 
-// withUrlPath will override the path of url in request
-func withUrlPath(urlPath string) func(r *http.Request) {
-	return func(req *http.Request) {
-		req.URL.Path = urlPath
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("X-Rudder-Source-Id", sourceID)
+	req.Header.Add("X-Rudder-Destination-Id", DestinationID)
+
+	res, err := httpClient.Do(req)
+	if err != nil {
+		t.Logf("sendEvent error: %v", err)
+		return
+	}
+	defer func() { httputil.CloseResponse(res) }()
+
+	body, err := io.ReadAll(res.Body)
+	if err != nil {
+		t.Logf("sendEvent error: %v", err)
+		return
+	}
+	if res.Status != "200 OK" {
+		return
 	}
+
+	t.Logf("Event Sent Successfully: (%s)", body)
 }
 
-func sendEvent(t *testing.T, payload *strings.Reader, callType, writeKey string, reqOptions ...func(r *http.Request)) {
+func sendEvent(t *testing.T, payload *strings.Reader, callType, writeKey string) {
 	t.Helper()
 	t.Logf("Sending %s Event", callType)
 
@@ -756,10 +776,6 @@ func sendEvent(t *testing.T, payload *strings.Reader, callType, writeKey string,
 		[]byte(fmt.Sprintf("%s:", writeKey)),
 	)))
 
-	for _, reqOption := range reqOptions {
-		reqOption(req)
-	}
-
 	res, err := httpClient.Do(req)
 	if err != nil {
 		t.Logf("sendEvent error: %v", err)

processor/processor.go

@@ -1544,8 +1544,8 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 
 	outCountMap := make(map[string]int64) // destinations enabled
 	destFilterStatusDetailMap := make(map[string]map[string]*types.StatusDetail)
-	// map of messageID to destinationID: for messages that needs to be delivered to a specific destinations only
-	msgToDestMap := make(map[string]string)
+	// map of jobID to destinationID: for messages that needs to be delivered to a specific destinations only
+	jobIDToSpecificDestMapOnly := make(map[int64]string)
 
 	spans := make([]stats.TraceSpan, 0, len(jobList))
 	defer func() {
@@ -1621,8 +1621,8 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 		// Iterate through all the events in the batch
 		for _, singularEvent := range gatewayBatchEvent.Batch {
 			messageId := misc.GetStringifiedData(singularEvent["messageId"])
-			if len(destinationID) != 0 {
-				msgToDestMap[messageId] = destinationID
+			if destinationID != "" {
+				jobIDToSpecificDestMapOnly[batchEvent.JobID] = destinationID
 			}
 
 			payloadFunc := ro.Memoize(func() json.RawMessage {
@@ -1730,9 +1730,9 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 			}
 			// REPORTING - GATEWAY metrics - END
 
-			// Getting all the destinations which are enabled for this
-			// event will be dropped if no valid destination is present
-			// if empty string destinationID is all the destinations for the source is validated
+			// Getting all the destinations which are enabled for this event.
+			// Event will be dropped if no valid destination is present
+			// if empty destinationID is passed in this fn all the destinations for the source are validated
 			// else only passed destinationID will be validated
 			if !proc.isDestinationAvailable(singularEvent, sourceID, destinationID) {
 				continue
@@ -1903,11 +1903,11 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 				enabledDestinationsList := proc.getConsentFilteredDestinations(
 					singularEvent,
 					lo.Filter(proc.getEnabledDestinations(sourceId, *destType), func(item backendconfig.DestinationT, index int) bool {
-						destId := msgToDestMap[event.Metadata.MessageID]
-						if len(destId) != 0 {
+						destId := jobIDToSpecificDestMapOnly[event.Metadata.JobID]
+						if destId != "" {
 							return destId == item.ID
 						}
-						return len(destId) == 0
+						return destId == ""
 					}),
 				)