e2e: test resync doesn't clear custom configurations (#3538)

tests fix in #3435

Author: serprex

flow/e2e/api_test.go

@@ -109,23 +109,32 @@ func (s APITestSuite) waitForActiveSlotForPostgresMirror(env WorkflowRun, conn *
 	})
 }
 
-// checkCatalogTableMapping checks the table mappings in the catalog for a given flow
-func (s APITestSuite) checkCatalogTableMapping(
+func (s APITestSuite) loadConfigFromCatalog(
 	ctx context.Context,
 	conn *pgx.Conn,
 	flowName string,
-	expectedSourceTableNames []string,
-) (bool, error) {
+) (*protos.FlowConnectionConfigs, error) {
 	var configBytes sql.RawBytes
 	if err := conn.QueryRow(ctx,
 		"SELECT config_proto FROM flows WHERE name = $1", flowName,
 	).Scan(&configBytes); err != nil {
-		return false, err
+		return nil, err
 	}
 
 	var config protos.FlowConnectionConfigs
-	if err := proto.Unmarshal(configBytes, &config); err != nil {
-		return false, err
+	return &config, proto.Unmarshal(configBytes, &config)
+}
+
+// checkCatalogTableMapping checks the table mappings in the catalog for a given flow
+func (s APITestSuite) checkCatalogTableMapping(
+	ctx context.Context,
+	conn *pgx.Conn,
+	flowName string,
+	expectedSourceTableNames []string,
+) (bool, error) {
+	config, err := s.loadConfigFromCatalog(ctx, conn, flowName)
+	if err != nil {
+		return false, fmt.Errorf("failed to load config from catalog: %w", err)
 	}
 
 	if len(config.TableMappings) != len(expectedSourceTableNames) {
@@ -587,6 +596,12 @@ func (s APITestSuite) TestResyncCompleted() {
 	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
 	flowConnConfig.DoInitialSnapshot = true
 	flowConnConfig.InitialSnapshotOnly = true
+	flowConnConfig.SnapshotNumRowsPerPartition = 3
+	flowConnConfig.SnapshotMaxParallelWorkers = 7
+	flowConnConfig.SnapshotNumTablesInParallel = 13
+	flowConnConfig.IdleTimeoutSeconds = 9
+	flowConnConfig.MaxBatchSize = 5040
+	// if true, then the flow will be resynced
 	response, err := s.CreateCDCFlow(s.t.Context(), &protos.CreateCDCFlowRequest{ConnectionConfigs: flowConnConfig})
 	require.NoError(s.t, err)
 	require.NotNil(s.t, response)
@@ -633,6 +648,13 @@ func (s APITestSuite) TestResyncCompleted() {
 	env, err = GetPeerflow(s.t.Context(), s.pg.PostgresConnector.Conn(), tc, flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
 	EnvWaitForFinished(s.t, env, time.Minute)
+
+	// check that custom config options persist across resync
+	config, err := s.loadConfigFromCatalog(s.t.Context(), s.pg.PostgresConnector.Conn(), flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	flowConnConfig.Resync = true // this gets left true after resync
+	config.Env = nil             // env is modified by API
+	require.EqualExportedValues(s.t, flowConnConfig, config)
 }
 
 func (s APITestSuite) TestDropCompleted() {