Skip to content

Commit 3f9d2c9

Browse files
committed
Fix snapshot override parameter not picked up during existing resync
1 parent c68bd69 commit 3f9d2c9

1 file changed

Lines changed: 16 additions & 1 deletion

File tree

flow/workflows/cdc_flow.go

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -224,6 +224,21 @@ func overrideSnapshotParametersInState(req *protos.FlowStateChangeRequest, state
224224
}
225225
}
226226

227+
func overrideSnapshotParametersInConfig(req *protos.FlowStateChangeRequest, cfg *protos.FlowConnectionConfigsCore) {
228+
if req.FlowConfigUpdate != nil && req.FlowConfigUpdate.GetCdcFlowConfigUpdate() != nil {
229+
cdcConfigUpdate := req.FlowConfigUpdate.GetCdcFlowConfigUpdate()
230+
if cdcConfigUpdate.SnapshotMaxParallelWorkers > 0 {
231+
cfg.SnapshotMaxParallelWorkers = cdcConfigUpdate.SnapshotMaxParallelWorkers
232+
}
233+
if cdcConfigUpdate.SnapshotNumTablesInParallel > 0 {
234+
cfg.SnapshotNumTablesInParallel = cdcConfigUpdate.SnapshotNumTablesInParallel
235+
}
236+
if cdcConfigUpdate.SnapshotNumRowsPerPartition > 0 {
237+
cfg.SnapshotNumRowsPerPartition = cdcConfigUpdate.SnapshotNumRowsPerPartition
238+
}
239+
}
240+
}
241+
227242
func processTableAdditions(
228243
ctx workflow.Context,
229244
logger log.Logger,
@@ -649,7 +664,7 @@ func CDCFlowWorkflow(
649664
// so we need to NOT sync the tableMappings to catalog to preserve original names
650665

651666
// We still override the snapshot parameters (when resync with updated values)
652-
overrideSnapshotParametersInState(val, state)
667+
overrideSnapshotParametersInConfig(val, cfg)
653668
uploadConfigToCatalog(ctx, cfg)
654669
state.DropFlowInput = &protos.DropFlowInput{
655670
FlowJobName: cfg.FlowJobName,

0 commit comments

Comments
 (0)