runs walheartbeat send in parallel

flow/activities/flowable.go

@@ -673,12 +673,20 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context, config *protos.
 	}
 	defer connectors.CloseConnector(srcConn)
 
-	err = srcConn.SendWALHeartbeat()
-	if err != nil {
-		return fmt.Errorf("failed to send WAL heartbeat: %w", err)
+	ticker := time.NewTicker(10 * time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("context is done, exiting wal heartbeat send loop")
+			return nil
+		case <-ticker.C:
+			err = srcConn.SendWALHeartbeat()
+			if err != nil {
+				return fmt.Errorf("failed to send WAL heartbeat: %w", err)
+			}
+			log.Info("sent wal heartbeat")
+		}
 	}
-
-	return nil
 }
 
 func (a *FlowableActivity) QRepWaitUntilNewRows(ctx context.Context,

flow/workflows/cdc_flow.go

@@ -100,18 +100,6 @@ func (s *CDCFlowWorkflowState) TruncateProgress() {
 	}
 }
 
-func (s *CDCFlowWorkflowState) SendWALHeartbeat(ctx workflow.Context, cfg *protos.FlowConnectionConfigs) error {
-	walHeartbeatCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-		StartToCloseTimeout: 5 * time.Minute,
-	})
-
-	if err := workflow.ExecuteActivity(walHeartbeatCtx, flowable.SendWALHeartbeat, cfg).Get(ctx, nil); err != nil {
-		return fmt.Errorf("failed to send WAL heartbeat: %w", err)
-	}
-
-	return nil
-}
-
 // CDCFlowWorkflowExecution represents the state for execution of a peer flow.
 type CDCFlowWorkflowExecution struct {
 	flowExecutionID string
@@ -268,6 +256,13 @@ func CDCFlowWorkflowWithConfig(
 		state.Progress = append(state.Progress, "executed setup flow and snapshot flow")
 	}
 
+	heartbeatCancelCtx, cancelHeartbeat := workflow.WithCancel(ctx)
+	walHeartbeatCtx := workflow.WithActivityOptions(heartbeatCancelCtx, workflow.ActivityOptions{
+		StartToCloseTimeout: 5 * time.Minute,
+	})
+	workflow.ExecuteActivity(walHeartbeatCtx, flowable.SendWALHeartbeat, cfg)
+	//walHeartbeatFuture.Get(ctx, nil)
+
 	syncFlowOptions := &protos.SyncFlowOptions{
 		BatchSize: int32(limits.MaxBatchSize),
 	}
@@ -410,10 +405,8 @@ func CDCFlowWorkflowWithConfig(
 		selector.Select(ctx)
 	}
 
-	// send WAL heartbeat
-	if err := state.SendWALHeartbeat(ctx, cfg); err != nil {
-		return state, err
-	}
+	// cancel the SendWalHeartbeat activity
+	cancelHeartbeat()
 
 	state.TruncateProgress()
 	return nil, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflowWithConfig, cfg, limits, state)