introduce skip destination drop

PeerDB-io · Jan 18, 2025 · d755c96 · d755c96
1 parent 9e85860
commit d755c96
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 17 deletions.
diff --git a/flow/workflows/drop_flow.go b/flow/workflows/drop_flow.go
@@ -42,30 +42,32 @@ func executeCDCDropActivities(ctx workflow.Context, input *protos.DropFlowInput)
 			_ = workflow.Sleep(ctx, time.Second)
 		}
 	}
-	dropDestination = func(f workflow.Future) {
-		destinationError = f.Get(ctx, nil)
-		destinationOk = destinationError == nil
-		if !destinationOk {
-			dropDestinationFuture := workflow.ExecuteActivity(ctx, flowable.DropFlowDestination, &protos.DropFlowActivityInput{
-				FlowJobName: input.FlowJobName,
-				PeerName:    input.FlowConnectionConfigs.DestinationName,
-			})
-			selector.AddFuture(dropDestinationFuture, dropDestination)
-			_ = workflow.Sleep(ctx, time.Second)
-		}
-	}
 
 	dropSourceFuture := workflow.ExecuteActivity(ctx, flowable.DropFlowSource, &protos.DropFlowActivityInput{
 		FlowJobName: input.FlowJobName,
 		PeerName:    input.FlowConnectionConfigs.SourceName,
 	})
 	selector.AddFuture(dropSourceFuture, dropSource)
-	dropDestinationFuture := workflow.ExecuteActivity(ctx, flowable.DropFlowDestination, &protos.DropFlowActivityInput{
-		FlowJobName: input.FlowJobName,
-		PeerName:    input.FlowConnectionConfigs.DestinationName,
-	})
 
-	selector.AddFuture(dropDestinationFuture, dropDestination)
+	if !input.SkipDestinationDrop {
+		dropDestination = func(f workflow.Future) {
+			destinationError = f.Get(ctx, nil)
+			destinationOk = destinationError == nil
+			if !destinationOk {
+				dropDestinationFuture := workflow.ExecuteActivity(ctx, flowable.DropFlowDestination, &protos.DropFlowActivityInput{
+					FlowJobName: input.FlowJobName,
+					PeerName:    input.FlowConnectionConfigs.DestinationName,
+				})
+				selector.AddFuture(dropDestinationFuture, dropDestination)
+				_ = workflow.Sleep(ctx, time.Second)
+			}
+		}
+		dropDestinationFuture := workflow.ExecuteActivity(ctx, flowable.DropFlowDestination, &protos.DropFlowActivityInput{
+			FlowJobName: input.FlowJobName,
+			PeerName:    input.FlowConnectionConfigs.DestinationName,
+		})
+		selector.AddFuture(dropDestinationFuture, dropDestination)
+	}
 
 	for {
 		selector.Select(ctx)

diff --git a/protos/flow.proto b/protos/flow.proto
@@ -332,6 +332,7 @@ message DropFlowInput {
   string flow_job_name = 1;
   bool drop_flow_stats = 4;
   FlowConnectionConfigs flow_connection_configs = 5;
+  bool skip_destination_drop = 6;
 }
 
 message TableSchemaDelta {