From c535df9dc2e53ced498f290861ef67d971d181ca Mon Sep 17 00:00:00 2001
From: Kunal Gupta <39487888+iamKunalGupta@users.noreply.github.com>
Date: Mon, 20 May 2024 09:40:36 +0530
Subject: [PATCH] fix: close and recreate destination when needed

(cherry picked from commit b30da813e9c7a81db043e510cb5f23f685172eae)
---
 flow/activities/flowable_core.go | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go
index ad3dca378a..7d48d9834f 100644
--- a/flow/activities/flowable_core.go
+++ b/flow/activities/flowable_core.go
@@ -113,6 +113,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	if err != nil {
 		return nil, err
 	}
+	connectors.CloseConnector(ctx, dstConn)
 	logger.Info("pulling records...", slog.Int64("LastOffset", lastOffset))
 	consumedOffset := atomic.Int64{}
 	consumedOffset.Store(lastOffset)
@@ -142,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()