徐启文
12/24/2025, 2:08 AMsource:
type: mysql
scan.startup.mode: earliest-offset # Start from earliest offset
scan.startup.mode: latest-offset # Start from latest offset
scan.startup.mode: specific-offset # Start from specific offset
scan.startup.mode: timestamp # Start from timestamp
scan.startup.mode: snapshot # Read snapshot only
scan.startup.specific-offset.file: 'mysql-bin.000003' # Binlog filename under specific offset startup mode
scan.startup.specific-offset.pos: 4 # Binlog position under specific offset mode
scan.startup.specific-offset.gtid-set: 24DA167-... # GTID set under specific offset startup mode
scan.startup.timestamp-millis: 1667232000000 # Timestamp under timestamp startup mode
# ...
val savepointPath = Option(
params.get("s",
params.get("fromSavepoint", null))
)
val isRestoring = savepointPath.isDefined
if (isRestoring) {
println(s"Restoring from: ${savepointPath.get}")
println("Using scan.startup.mode: latest-offset")
} else {
println("Starting fresh job")
println("Using scan.startup.mode: initial")
}
val scanMode = if (isRestoring) "latest-offset" else "initial"
Is it necessary for me to judge the parameter like this, and what should be the correct way to write it? Looking forward to your reply.