comment out kafka push

PeerDB-io · Apr 12, 2024 · ef229c6 · ef229c6
1 parent 2c05847
commit ef229c6
Showing 1 changed file with 9 additions and 9 deletions.
diff --git a/flow/connectors/kafka/kafka.go b/flow/connectors/kafka/kafka.go
@@ -168,13 +168,13 @@ func lvalueToKafkaRecord(ls *lua.LState, value lua.LValue) (*kgo.Record, error)
 
 func (c *KafkaConnector) SyncRecords(ctx context.Context, req *model.SyncRecordsRequest[model.RecordItems]) (*model.SyncResponse, error) {
 	var wg sync.WaitGroup
-	wgCtx, wgErr := context.WithCancelCause(ctx)
-	produceCb := func(r *kgo.Record, err error) {
-		if err != nil {
-			wgErr(err)
-		}
-		wg.Done()
-	}
+	wgCtx, _ := context.WithCancelCause(ctx)
+	// produceCb := func(r *kgo.Record, err error) {
+	// 	if err != nil {
+	// 		wgErr(err)
+	// 	}
+	// 	wg.Done()
+	// }
 
 	numRecords := int64(0)
 	tableNameRowsMapping := utils.InitialiseTableRowsMap(req.TableMappings)
@@ -259,8 +259,8 @@ Loop:
 						kr.Topic = record.GetDestinationTableName()
 					}
 
-					wg.Add(1)
-					c.client.Produce(wgCtx, kr, produceCb)
+					//wg.Add(1)
+					//c.client.Produce(wgCtx, kr, produceCb)
 					record.PopulateCountMap(tableNameRowsMapping)
 				}
 			}