Fetch configuration from DB #3407
Conversation
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
2 times, most recently
from
7f140e4 to
00660b8
Compare
| if flowConfigUpdate.SnapshotNumTablesInParallel > 0 { | ||
| state.SnapshotNumTablesInParallel = flowConfigUpdate.SnapshotNumTablesInParallel | ||
| } | ||
| cfg = syncStateToConfigProtoInCatalog(ctx, cfg, state.FlowConfigUpdate) |
There was a problem hiding this comment.
this > 0 behavior was added recently in #3435, this probably regresses
There was a problem hiding this comment.
given that we no longer have or rely on state, this should be a non-issue.
flow/workflows/cdc_flow.go
Outdated
| ctx workflow.Context, | ||
| cfg *protos.FlowConnectionConfigs, | ||
| flowJobName string, | ||
| // cfg *protos.FlowConnectionConfigs, |
There was a problem hiding this comment.
this breaks backwards compatibility. Should still take config & pull name from that to maintain function signature
There was a problem hiding this comment.
The problem with keeing the protos.FlowConnectionConfig here, is that in cases where this is too large - it will still fail.
Is the suggestion to pass both and then remove the cfg from here?
There was a problem hiding this comment.
no, only pass cfg. new code should pass an empty cfg with only FlowJobName present
There was a problem hiding this comment.
OK understood, I'll go in this direction then :)
| } | ||
| // because we have renamed the tables. | ||
| cfg.TableMappings = state.SyncFlowOptions.TableMappings | ||
| return nil, errors.New("cannot start CDCFlow with Resync enabled, please drop the flow and start again") |
No longer passing the entire configuration object to the Temporal job as we can hit the 2MB hard limit imposed by Temporal. Instead, pass flowJobName and fetch the config from the DB in all the places required. There are some additional changes that were required to be made, as some of the jobs are used from multiple contexts - and I had to adapt some of the information passed to ensure that they continue to work as expected.
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
from
2341981 to
190bc92
Compare
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
from
20f7fd4 to
0b42cca
Compare
| func FetchConfigFromDB(flowName string) (*protos.FlowConnectionConfigs, error) { | ||
| var configBytes sql.RawBytes | ||
| dbCtx := context.Background() | ||
| pool, _ := GetCatalogConnectionPoolFromEnv(dbCtx) |
| ) (*protos.SetupFlowOutput, error) { | ||
| s.Info("executing setup flow") | ||
| // gotta fetch the config from the catalog. | ||
| config, err := internal.FetchConfigFromDB(flowJobName) |
There was a problem hiding this comment.
can't call this in workflows
could maybe run as side effect, but then it ends up in history
There was a problem hiding this comment.
Why can this not be called in workflows? 🤔
No longer passing the entire configuration object to the Temporal job as we can hit the 2MB hard limit imposed by Temporal. Instead, pass flowJobName and fetch the config from the DB in all the places required.
There are some additional changes that were required to be made, as some of the jobs are used from multiple contexts - and I had to adapt some of the information passed to ensure that they continue to work as expected.