increment rows

PeerDB-io · May 22, 2024 · b6403eb · b6403eb
1 parent 371cffb
commit b6403eb
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 0 deletions.
diff --git a/flow/connectors/kafka/kafka.go b/flow/connectors/kafka/kafka.go
@@ -203,6 +203,7 @@ func (c *KafkaConnector) createPool(
 			for _, kr := range result.records {
 				c.client.Produce(ctx, kr, func(_ *kgo.Record, err error) {
 					if err != nil {
+						c.logger.Error("[kafka] produce error", slog.Any("error", err))
 						queueErr(err)
 					} else if recordCounter.Add(-1) == 0 && lastSeenLSN != nil {
 						shared.AtomicInt64Max(lastSeenLSN, result.lsn)

diff --git a/flow/connectors/kafka/qrep.go b/flow/connectors/kafka/qrep.go
@@ -3,6 +3,7 @@ package connkafka
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"sync/atomic"
 	"time"
 
@@ -39,6 +40,7 @@ func (c *KafkaConnector) SyncQRepRecords(
 	queueCtx, queueErr := context.WithCancelCause(ctx)
 	pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, nil, queueErr)
 	if err != nil {
+		c.logger.Error("failed to create pool", slog.Any("error", err))
 		return 0, err
 	}
 	defer pool.Close()

diff --git a/flow/connectors/postgres/qrep_query_executor.go b/flow/connectors/postgres/qrep_query_executor.go
@@ -161,6 +161,7 @@ func (qe *QRepQueryExecutor) processRowsStream(
 		}
 
 		stream.Records <- record
+		numRows.Add(1)
 	}
 
 	qe.logger.Info("processed row stream", slog.String("cursor", cursorName), slog.Int64("records", numRows.Load()))
-Original file line number
+Diff line change
@@ Expand Up / @@ -161,6 +161,7 @@ func (qe *QRepQueryExecutor) processRowsStream( @@
     		}
     		stream.Records <- record
+    		numRows.Add(1)
     	}
     	qe.logger.Info("processed row stream", slog.String("cursor", cursorName), slog.Int64("records", numRows.Load()))
@@ Expand Down @@