TT-16809, fixed mcp records not being excluded in some non-mcp pumps

Author: andrei-tyk

pumps/mongo.go

@@ -470,6 +470,11 @@ func (m *MongoPump) shouldProcessItem(item interface{}, isForGraphRecords bool)
 		return &thisItem, true
 	}
 
+	// MCP records are handled by dedicated MCP pumps, skip them here.
+	if thisItem.IsMCPRecord() {
+		return &thisItem, true
+	}
+
 	isGraphRecord := thisItem.IsGraphRecord()
 	if isForGraphRecords && !isGraphRecord {
 		return &thisItem, true

pumps/mongo_selective.go

@@ -196,7 +196,12 @@ func (m *MongoSelectivePump) WriteData(ctx context.Context, data []interface{})
 	analyticsPerOrg := make(map[string][]interface{})
 
 	for _, v := range data {
-		orgID := v.(analytics.AnalyticsRecord).OrgID
+		rec := v.(analytics.AnalyticsRecord)
+		// MCP records are handled by dedicated MCP pumps, skip them here.
+		if rec.IsMCPRecord() {
+			continue
+		}
+		orgID := rec.OrgID
 		collectionName, collErr := m.GetCollectionName(orgID)
 		skip := false
 		if collErr != nil {

pumps/sql.go

@@ -209,6 +209,10 @@ func (c *SQLPump) WriteData(ctx context.Context, data []interface{}) error {
 	for _, r := range data {
 		if r != nil {
 			rec := r.(analytics.AnalyticsRecord)
+			// MCP records are handled by dedicated MCP pumps, skip them here.
+			if rec.IsMCPRecord() {
+				continue
+			}
 			typedData = append(typedData, &rec)
 		}
 	}