From c23151e00752d1b524fae1354f1bc0c79e9b31f0 Mon Sep 17 00:00:00 2001
From: Amogh-Bharadwaj <amogh@peerdb.io>
Date: Sat, 25 May 2024 00:40:19 +0530
Subject: [PATCH] Revert "move the dst connector retreival further down"

This reverts commit b404aaaabbc8aa3ef1e212233aa1e2896410e812.
---
 flow/activities/flowable_core.go | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go
index 49059ac8e3..7d48d9834f 100644
--- a/flow/activities/flowable_core.go
+++ b/flow/activities/flowable_core.go
@@ -11,7 +11,7 @@ import (
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
-	lua "github.com/yuin/gopher-lua"
+	"github.com/yuin/gopher-lua"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/log"
 	"go.temporal.io/sdk/temporal"
@@ -143,6 +143,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	hasRecords := !recordBatch.WaitAndCheckEmpty()
 	logger.Info("current sync flow has records?", slog.Bool("hasRecords", hasRecords))
 
+	dstConn, err = connectors.GetAs[TSync](ctx, config.Destination)
+	if err != nil {
+		return nil, fmt.Errorf("failed to recreate destination connector: %w", err)
+	}
+
 	if !hasRecords {
 		// wait for the pull goroutine to finish
 		err = errGroup.Wait()
@@ -156,11 +161,6 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		}
 		logger.Info("no records to push")
 
-		dstConn, err = connectors.GetAs[TSync](ctx, config.Destination)
-		if err != nil {
-			return nil, fmt.Errorf("failed to recreate destination connector: %w", err)
-		}
-
 		err := dstConn.ReplayTableSchemaDeltas(ctx, flowName, recordBatch.SchemaDeltas)
 		if err != nil {
 			return nil, fmt.Errorf("failed to sync schema: %w", err)