resolevd comments

Author: masterashu

flow/cmd/handler.go

@@ -658,24 +658,6 @@ func (h *FlowRequestHandler) getWorkflowID(ctx context.Context, flowJobName stri
 	return workflowID, nil
 }
 
-func applySnapshotConfigOverrides(config *protos.FlowConnectionConfigs, update *protos.CDCFlowConfigUpdate) {
-	if update == nil {
-		return
-	}
-	if update.SnapshotNumRowsPerPartition > 0 {
-		config.SnapshotNumRowsPerPartition = update.SnapshotNumRowsPerPartition
-	}
-	if update.SnapshotNumPartitionsOverride > 0 {
-		config.SnapshotNumPartitionsOverride = update.SnapshotNumPartitionsOverride
-	}
-	if update.SnapshotMaxParallelWorkers > 0 {
-		config.SnapshotMaxParallelWorkers = update.SnapshotMaxParallelWorkers
-	}
-	if update.SnapshotNumTablesInParallel > 0 {
-		config.SnapshotNumTablesInParallel = update.SnapshotNumTablesInParallel
-	}
-}
-
 func (h *FlowRequestHandler) resyncByRecreatingFlow(
 	ctx context.Context,
 	flowName string,
@@ -708,11 +690,16 @@ func (h *FlowRequestHandler) resyncByRecreatingFlow(
 
 	config.Resync = true
 	config.DoInitialSnapshot = true
-	applySnapshotConfigOverrides(config, cdcConfigUpdate)
 	// validate mirror first because once the mirror is dropped, there's no going back
-	if _, err := h.ValidateCDCMirror(ctx, &protos.CreateCDCFlowRequest{
-		ConnectionConfigs: config,
-	}); err != nil {
+	if internalVersion, err := internal.PeerDBForceInternalVersion(ctx, config.Env); err != nil {
+		return NewInternalApiError(err)
+	} else {
+		config.Version = internalVersion
+	}
+	configCore := pconv.FlowConnectionConfigsToCore(config, 0)
+	internal.ApplySnapshotConfigOverrides(configCore, cdcConfigUpdate)
+
+	if _, err := h.validateCDCMirrorImpl(ctx, configCore, false); err != nil {
 		return err
 	}
 
@@ -721,7 +708,6 @@ func (h *FlowRequestHandler) resyncByRecreatingFlow(
 	}
 
 	workflowID := getWorkflowID(config.FlowJobName)
-	configCore := pconv.FlowConnectionConfigsToCore(config, 0)
 	if _, err := h.createCDCFlow(ctx, configCore, workflowID); err != nil {
 		return err
 	}

flow/internal/flow_configuration_helpers.go

@@ -26,3 +26,21 @@ func FetchConfigFromDB(ctx context.Context, catalogPool shared.CatalogPool, flow
 
 	return &cfgFromDB, nil
 }
+
+func ApplySnapshotConfigOverrides(config *protos.FlowConnectionConfigsCore, update *protos.CDCFlowConfigUpdate) {
+	if update == nil {
+		return
+	}
+	if update.SnapshotNumRowsPerPartition > 0 {
+		config.SnapshotNumRowsPerPartition = update.SnapshotNumRowsPerPartition
+	}
+	if update.SnapshotNumPartitionsOverride > 0 {
+		config.SnapshotNumPartitionsOverride = update.SnapshotNumPartitionsOverride
+	}
+	if update.SnapshotMaxParallelWorkers > 0 {
+		config.SnapshotMaxParallelWorkers = update.SnapshotMaxParallelWorkers
+	}
+	if update.SnapshotNumTablesInParallel > 0 {
+		config.SnapshotNumTablesInParallel = update.SnapshotNumTablesInParallel
+	}
+}

flow/shared/telemetry/activity_logging.go

@@ -93,25 +93,6 @@ func LogActivityStartFlowConfigUpdate(ctx context.Context, flowName string, upda
 		changes = append(changes, fmt.Sprintf("tables removed: %v", removedTables))
 	}
 
-	if update.BatchSize > 0 {
-		changes = append(changes, fmt.Sprintf("batchSize: %v", update.BatchSize))
-	}
-	if update.IdleTimeout > 0 {
-		changes = append(changes, fmt.Sprintf("idleTimeout: %v", update.IdleTimeout))
-	}
-	if update.SnapshotNumPartitionsOverride > 0 {
-		changes = append(changes, fmt.Sprintf("snapshotNumPartitionsOverride: %v", update.SnapshotNumPartitionsOverride))
-	}
-	if update.SnapshotMaxParallelWorkers > 0 {
-		changes = append(changes, fmt.Sprintf("snapshotMaxParallelWorkers: %v", update.SnapshotMaxParallelWorkers))
-	}
-	if update.SnapshotNumTablesInParallel > 0 {
-		changes = append(changes, fmt.Sprintf("snapshotNumTablesInParallel: %v", update.SnapshotNumTablesInParallel))
-	}
-	if update.SnapshotNumRowsPerPartition > 0 {
-		changes = append(changes, fmt.Sprintf("snapshotNumRowsPerPartition: %v", update.SnapshotNumRowsPerPartition))
-	}
-
 	logActivity(ctx, ActionStartFlowConfigUpdate,
 		slog.String("flowName", flowName),
 		slog.String("activityDetails", strings.Join(changes, ", ")))

flow/workflows/cdc_flow.go

@@ -653,11 +653,12 @@ func CDCFlowWorkflow(
 				// so we need to NOT sync the tableMappings to catalog to preserve original names
 
 				// We still override the snapshot parameters (when resync with updated values)
-				overrideSnapshotParametersInState(val, state)
-				resyncCfg := syncStateToConfigProtoInCatalog(ctx, cfg, state)
+				internal.ApplySnapshotConfigOverrides(cfg, val.GetFlowConfigUpdate().GetCdcFlowConfigUpdate())
+				uploadConfigToCatalog(ctx, cfg)
+
 				state.DropFlowInput = &protos.DropFlowInput{
-					FlowJobName:           resyncCfg.FlowJobName,
-					FlowConnectionConfigs: resyncCfg,
+					FlowJobName:           cfg.FlowJobName,
+					FlowConnectionConfigs: cfg,
 					DropFlowStats:         val.DropMirrorStats,
 					SkipDestinationDrop:   val.SkipDestinationDrop,
 					Resync:                true,