fix: ignore gwJobs' destinationID in processor rsources.statCollector (…

…#4321)
rudderlabs · Jan 16, 2024 · 8d58bd5 · 8d58bd5
1 parent 1166d5c
commit 8d58bd5
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/processor/processor.go b/processor/processor.go
@@ -3000,7 +3000,7 @@ func (proc *Handle) handlePendingGatewayJobs(partition string) bool {
 		return false
 	}
 
-	rsourcesStats := rsources.NewStatsCollector(proc.rsourcesService)
+	rsourcesStats := rsources.NewStatsCollector(proc.rsourcesService, rsources.IgnoreDestinationID())
 	rsourcesStats.BeginProcessing(unprocessedList.Jobs)
 
 	proc.Store(partition,

diff --git a/processor/worker.go b/processor/worker.go
@@ -140,7 +140,7 @@ func (w *worker) Work() (worked bool) {
 
 	w.handle.stats().DBReadThroughput(w.partition).Count(throughputPerSecond(jobs.EventsCount, time.Since(start)))
 
-	rsourcesStats := rsources.NewStatsCollector(w.handle.rsourcesService())
+	rsourcesStats := rsources.NewStatsCollector(w.handle.rsourcesService(), rsources.IgnoreDestinationID())
 	rsourcesStats.BeginProcessing(jobs.Jobs)
 	subJobs := w.handle.jobSplitter(jobs.Jobs, rsourcesStats)
 	for _, subJob := range subJobs {

diff --git a/processor/worker_test.go b/processor/worker_test.go
@@ -198,7 +198,7 @@ func (m *mockWorkerHandle) handlePendingGatewayJobs(partition string) bool {
 	if len(jobs.Jobs) > 0 {
 		_ = m.markExecuting(partition, jobs.Jobs)
 	}
-	rsourcesStats := rsources.NewStatsCollector(m.rsourcesService())
+	rsourcesStats := rsources.NewStatsCollector(m.rsourcesService(), rsources.IgnoreDestinationID())
 	for _, subJob := range m.jobSplitter(jobs.Jobs, rsourcesStats) {
 		m.Store(partition,
 			m.transformations(partition,