From f3d1d8e93de7143951d2ee94b46b69642f7048c6 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Wed, 27 Dec 2023 09:27:53 -0500 Subject: [PATCH] update max batch size on signal (#910) --- flow/workflows/cdc_flow.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go index 55751bd096..b73f6945fe 100644 --- a/flow/workflows/cdc_flow.go +++ b/flow/workflows/cdc_flow.go @@ -291,7 +291,10 @@ func CDCFlowWorkflowWithConfig( c.Receive(ctx, &batchSize) w.logger.Info("received batch size signal: ", batchSize) syncFlowOptions.BatchSize = batchSize + cfg.MaxBatchSize = uint32(batchSize) + limits.MaxBatchSize = int(batchSize) }) + batchSizeSelector.AddDefault(func() { w.logger.Info("no batch size signal received, batch size remains: ", syncFlowOptions.BatchSize)