Merge branch 'main' into fix-create-peer-if-not-exists

PeerDB-io · May 20, 2024 · 2a4df81 · 2a4df81
2 parents 783d50f + 1ece82e
commit 2a4df81
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/flow/connectors/kafka/qrep.go b/flow/connectors/kafka/qrep.go
@@ -9,6 +9,7 @@ import (
 	"github.com/twmb/franz-go/pkg/kgo"
 	lua "github.com/yuin/gopher-lua"
 
+	"github.com/PeerDB-io/peer-flow/connectors/utils"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/model"
 	"github.com/PeerDB-io/peer-flow/pua"
@@ -28,6 +29,11 @@ func (c *KafkaConnector) SyncQRepRecords(
 	numRecords := atomic.Int64{}
 	schema := stream.Schema()
 
+	shutdown := utils.HeartbeatRoutine(ctx, func() string {
+		return fmt.Sprintf("sent %d records to %s", numRecords.Load(), config.DestinationTableIdentifier)
+	})
+	defer shutdown()
+
 	queueCtx, queueErr := context.WithCancelCause(ctx)
 	pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, nil, queueErr)
 	if err != nil {