Skip to content

mysql: add syncer.EventCacheCount setting #4413

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dtunikov merged 4 commits into from
Jun 12, 2026
Merged

mysql: add syncer.EventCacheCount setting #4413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fb3f368
mysql: add syncer.EventCacheCount setting
dtunikov Jun 11, 2026
bea8ed6
rm mysql dyn config target
dtunikov Jun 12, 2026
9453337
Update flow/connectors/mysql/ssh_test.go
dtunikov Jun 12, 2026
f117d38
Update flow/connectors/mysql/ssh_keepalive_test.go
dtunikov Jun 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions flow/connectors/mysql/cdc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ func (c *MySqlConnector) SetupReplConn(context.Context, map[string]string) error
return nil
}

func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSyncer, error) {
func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string) (*replication.BinlogSyncer, error) {
var tlsConfig *tls.Config
if !c.config.DisableTls {
var err error
Expand Down Expand Up @@ -271,6 +271,10 @@ func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSy
config.Password = token
}

eventCacheCount, err := internal.PeerDBMySQLEventCacheCount(ctx, env)
if err != nil {
return nil, fmt.Errorf("failed to get event cache count: %w", err)
}
//nolint:gosec
return replication.NewBinlogSyncer(replication.BinlogSyncerConfig{
ServerID: rand.Uint32(),
Expand All @@ -286,12 +290,14 @@ func (c *MySqlConnector) startSyncer(ctx context.Context) (*replication.BinlogSy
ParseTime: true,
TLSConfig: tlsConfig,
HeartbeatPeriod: c.binlogHeartbeatPeriod,
EventCacheCount: eventCacheCount,
}), nil
}

func (c *MySqlConnector) startStreaming(
ctx context.Context,
pos string,
env map[string]string,
) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
parsedOffset, err := parseReplicationOffsetText(c.Flavor(), pos)
if err != nil {
Expand All @@ -300,9 +306,9 @@ func (c *MySqlConnector) startStreaming(

switch parsedOffset.mechanism {
case protos.MySqlReplicationMechanism_MYSQL_FILEPOS.String():
return c.startCdcStreamingFilePos(ctx, parsedOffset.pos)
return c.startCdcStreamingFilePos(ctx, parsedOffset.pos, env)
case protos.MySqlReplicationMechanism_MYSQL_GTID.String():
return c.startCdcStreamingGtid(ctx, parsedOffset.gset)
return c.startCdcStreamingGtid(ctx, parsedOffset.gset, env)
default:
return nil, nil, nil, mysql.Position{}, fmt.Errorf("empty mysql replication offset")
}
Expand All @@ -311,8 +317,9 @@ func (c *MySqlConnector) startStreaming(
func (c *MySqlConnector) startCdcStreamingFilePos(
ctx context.Context,
pos mysql.Position,
env map[string]string,
) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
syncer, err := c.startSyncer(ctx)
syncer, err := c.startSyncer(ctx, env)
if err != nil {
return nil, nil, nil, mysql.Position{}, err
}
Expand All @@ -327,8 +334,9 @@ func (c *MySqlConnector) startCdcStreamingFilePos(
func (c *MySqlConnector) startCdcStreamingGtid(
ctx context.Context,
gset mysql.GTIDSet,
env map[string]string,
) (*replication.BinlogSyncer, *replication.BinlogStreamer, mysql.GTIDSet, mysql.Position, error) {
syncer, err := c.startSyncer(ctx)
syncer, err := c.startSyncer(ctx, env)
if err != nil {
return nil, nil, nil, mysql.Position{}, err
}
Expand Down Expand Up @@ -389,7 +397,7 @@ func (c *MySqlConnector) PullRecords(
return fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
}

syncer, mystream, gset, pos, err := c.startStreaming(ctx, req.LastOffset.Text)
syncer, mystream, gset, pos, err := c.startStreaming(ctx, req.LastOffset.Text, req.Env)
if err != nil {
return err
}
Expand Down
2 changes: 1 addition & 1 deletion flow/connectors/mysql/ssh_keepalive_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,7 +316,7 @@ func TestMySQLCloseSyncerWithTimeout(t *testing.T) {

pos, err := connector.GetMasterPos(ctx)
require.NoError(t, err)
syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled
syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos, nil) //nolint:dogsled
require.NoError(t, err)

// Let CDC streaming establish before blocking the MySQL server.
Expand Down
2 changes: 1 addition & 1 deletion flow/connectors/mysql/ssh_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ func TestMySQLSyncerClose(t *testing.T) {

pos, err := connector.GetMasterPos(ctx)
require.NoError(t, err)
syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled
syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos, nil) //nolint:dogsled
require.NoError(t, err)

// Let CDC streaming establish before closing the syncer.
Expand Down
12 changes: 12 additions & 0 deletions flow/internal/dynamicconf.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -460,6 +460,14 @@ var DynamicSettings = [...]*protos.DynamicSetting{
ApplyMode: protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
TargetForSetting: protos.DynconfTarget_POSTGRES,
},
{
Name: "PEERDB_MYSQL_EVENT_CACHE_COUNT",
Description: "Maximum number of events the MySQL Go driver loads in memory at once, useful when pushing large rows",
DefaultValue: "10240",
ValueType: protos.DynconfValueType_INT,
ApplyMode: protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
TargetForSetting: protos.DynconfTarget_ALL,
},
}

var DynamicIndex = func() map[string]int {
Expand Down Expand Up @@ -613,6 +621,10 @@ func PeerDBCDCChannelBufferSize(ctx context.Context, env map[string]string) (int
return dynamicConfSigned[int](ctx, env, "PEERDB_CDC_CHANNEL_BUFFER_SIZE")
}

func PeerDBMySQLEventCacheCount(ctx context.Context, env map[string]string) (int, error) {
return dynamicConfSigned[int](ctx, env, "PEERDB_MYSQL_EVENT_CACHE_COUNT")
}

func PeerDBNormalizeBufferHours(ctx context.Context, env map[string]string) (int64, error) {
return dynamicConfSigned[int64](ctx, env, "PEERDB_NORMALIZE_BUFFER_HOURS")
}
Expand Down
Loading