Function processCDCFlowConfigUpdate

flow/workflows/cdc_flow.go:94–169

(
	ctx workflow.Context,
	logger log.Logger,
	cfg *protos.FlowConnectionConfigsCore,
	state *cdc_state.CDCFlowWorkflowState,
	mirrorNameSearch temporal.SearchAttributes,
)

}

func processCDCFlowConfigUpdate(
	ctx workflow.Context,
	logger log.Logger,
	cfg *protos.FlowConnectionConfigsCore,
	state *cdc_state.CDCFlowWorkflowState,
	mirrorNameSearch temporal.SearchAttributes,
) error {
	flowConfigUpdate := state.FlowConfigUpdate

	// Capture old values for logging before updates are applied
	oldValues := telemetry.OldCDCFlowValues{
		BatchSize:                     state.SyncFlowOptions.BatchSize,
		IdleTimeout:                   state.SyncFlowOptions.IdleTimeoutSeconds,
		SnapshotNumRowsPerPartition:   state.SnapshotNumRowsPerPartition,
		SnapshotNumPartitionsOverride: state.SnapshotNumPartitionsOverride,
		SnapshotMaxParallelWorkers:    state.SnapshotMaxParallelWorkers,
		SnapshotNumTablesInParallel:   state.SnapshotNumTablesInParallel,
	}
	if len(flowConfigUpdate.UpdatedEnv) > 0 {
		oldValues.Env = make(map[string]string, len(flowConfigUpdate.UpdatedEnv))
		if cfg.Env != nil {
			for key := range flowConfigUpdate.UpdatedEnv {
				oldValues.Env[key] = cfg.Env[key]
			}
		}
	}

	if flowConfigUpdate.BatchSize > 0 {
		state.SyncFlowOptions.BatchSize = flowConfigUpdate.BatchSize
	}
	if flowConfigUpdate.IdleTimeout > 0 {
		state.SyncFlowOptions.IdleTimeoutSeconds = flowConfigUpdate.IdleTimeout
	}
	if flowConfigUpdate.UpdatedEnv != nil {
		if cfg.Env == nil {
			cfg.Env = make(map[string]string, len(flowConfigUpdate.UpdatedEnv))
		}
		maps.Copy(cfg.Env, flowConfigUpdate.UpdatedEnv)
	}
	if flowConfigUpdate.SnapshotNumRowsPerPartition > 0 {
		state.SnapshotNumRowsPerPartition = flowConfigUpdate.SnapshotNumRowsPerPartition
	}
	if flowConfigUpdate.SnapshotNumPartitionsOverride > 0 {
		state.SnapshotNumPartitionsOverride = flowConfigUpdate.SnapshotNumPartitionsOverride
	}
	if flowConfigUpdate.SnapshotMaxParallelWorkers > 0 {
		state.SnapshotMaxParallelWorkers = flowConfigUpdate.SnapshotMaxParallelWorkers
	}
	if flowConfigUpdate.SnapshotNumTablesInParallel > 0 {
		state.SnapshotNumTablesInParallel = flowConfigUpdate.SnapshotNumTablesInParallel
	}

	tablesAreAdded := len(flowConfigUpdate.AdditionalTables) > 0
	tablesAreRemoved := len(flowConfigUpdate.RemovedTables) > 0
	if tablesAreAdded || tablesAreRemoved {
		logger.Info("processing CDCFlowConfigUpdate", slog.Any("updatedState", flowConfigUpdate))

		if tablesAreAdded {