增量抽取数据失败，使用change_stream模式增量抽取只有插入操作的时序表，一直是 write_success=0 。如果incr_sync.mongo_fetch_method转成用oplog来做增量抽取又会报错误。

MongoShake的版本： mongo-shake-v2.8.8

源端MongoDB 版本： 7.0.28
目标端MongoDB版本：阿里云MongoDB 8.0.18-20260812202609 （看文档对应基准版本是7.0.39）

时序表：operation.gpsevents_20260X

**change_stream 模式下日志一直是 write_success=0 ，但是check point又一直往前推进 **
```
[2026/09/08 13:35:30 CST] [INFO] [name=ciotrep, stage=incr, get=754333, filter=754314, write_success=0, tps=0, ckpt_times=5, lsn_ckpt={7683033143435067406[1788845552, 14], 2026-09-08 13:32:32}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] try to update checkpoint mandatory from 7683033143435067406[1788845552, 14] to {1788845733 35}
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] waiting last checkpoint[7683033920824148003[1788845733, 35]] updated
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] last checkpoint[7683033920824148003[1788845733, 35]] updated [false]
[2026/09/08 13:35:33 CST] [INFO] checkpoint update sync.ckptTime to:2026-09-08 13:35:33.305656873 +0800 CST m=+1094.444924388
[2026/09/08 13:35:33 CST] [INFO] checkpoint func lowest:7683033920824148003[1788845733, 35] inMemoryTs:7683033143435067406[1788845552, 14] flush:false inputTs:7683033920824148003
[2026/09/08 13:35:33 CST] [INFO] ciotrep Record new checkpoint in MongoDB success [1788845733]
[2026/09/08 13:35:33 CST] [INFO] CheckpointOperation write success. updated from 7683033143435067406[1788845552, 14] to 7683033920824148003[1788845733, 35]
[2026/09/08 13:35:35 CST] [INFO] [name=ciotrep, stage=incr, get=755946, filter=755914, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:40 CST] [INFO] [name=ciotrep, stage=incr, get=757245, filter=757194, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:45 CST] [INFO] [name=ciotrep, stage=incr, get=758413, filter=758402, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:50 CST] [INFO] [name=ciotrep, stage=incr, get=760053, filter=760017, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
```

**oplog 模式下报错日志如下**
```
[2026/09/08 13:37:54 CST] [INFO] Close client with mongodb://root:XXX@XXX:3717,XXX:3717
[2026/09/08 13:37:54 CST] [INFO] update or.query to map[ts:map[$gt:{1788845746 252}]]
[2026/09/08 13:37:54 CST] [INFO] New session to mongodb://root:XXX@XXX:3717,XXX:3717 successfully
[2026/09/08 13:37:54 CST] [WARNING] insert docs with length[1] into ns[operation.system.buckets.gpsevents_202608] of dest mongo failed[bulk write exception: write errors: [bypassDocumentValidation is not permitted on timeseries collections]] res[&{0 0 0 0 0 map[]}]
[2026/09/08 13:37:54 CST] [CRITICAL] Replay-3, executor-3, oplog for namespace[operation.system.buckets.gpsevents_202608] op[i] failed. error type[mongo.BulkWriteException] error[bulk write exception: write errors: [bypassDocumentValidation is not permitted on timeseries collections]], logs number[1], firstLog: {"ts":{"T":1788845738,"I":116},"t":52,"h":null,"v":2,"op":"i","ns":"operation.system.buckets.gpsevents_202608","o":[{"Key":"_id","Value":"6a91be908f5ed5c1c8e58a80"},{"Key":"control","Value":[{"Key":"version","Value":1},{"Key":"min","Value":[{"Key":"_id","Value":"6a9f9eaaf8b2b632bad49f7f"},{"Key":"seqno","Value":"3540"},{"Key":"revTime","Value":"2026-09-08T13:35:36+08:00"},{"Key":"cltTime","Value":"2026-08-29T01:02:39+08:00"},{"Key":"cltDateTime","Value":"2026-08-28T17:00:00Z"}......
```

**使用配置如下( 改变部分只有 incr_sync.mongo_fetch_method = change_stream or incr_sync.mongo_fetch_method = oplog )**

```
# if you have any problem, please visit https://github.com/alibaba/MongoShake/wiki/FAQ
# for the detail explanation, please visit https://github.com/alibaba/MongoShake/wiki/%E9%85%8D%E7%BD%AE%E5%8F%82%E6%95%B0%E8%AF%B4%E6%98%8E
# 如果有问题，请先查看FAQ文档以及wiki上的说明。
# 关于各个参数的详细说明，请参考：wiki-配置参数说明

# current configuration version, do not modify.
# 当前配置文件的版本号，请不要修改该值。
conf.version = 14

# --------------------------- global configuration ---------------------------
# collector name
# id用于输出pid文件等信息。
id = mongoshake_timeseries

# high availability option.
# enable master election if set true. only one mongoshake can become master
# and do sync, the others will wait and at most one of them become master once
# previous master die. The master information stores in the `mongoshake` db in the source
# database by default.
# This option is useless when there is only one mongoshake running.
# 如果开启主备mongoshake拉取同一个源端，此参数需要开启。
master_quorum = false
# master_quorum only supports checkpoint.storage=database. When master_quorum=true,
# master_quorum.election_id is required and should be unique per independent sync job or HA group.
# master_quorum 仅支持 checkpoint.storage=database；开启后必须显式配置 master_quorum.election_id，
# 且每个独立同步任务或 HA 组应使用不同的 election id。
# master_quorum.election_id = 66e3ffdd9df1af8e2308fb53

# http api interface. Users can use this api to monitor mongoshake.
# `curl 127.0.0.1:9100`.
# We also provide a restful tool named "mongoshake-stat" to
# print ack, lsn, checkpoint and qps information based on this api.
# usage: `./mongoshake-stat --port=9100`
# Set the port to a value <= 0 to disable the corresponding listener.
# 全量和增量的restful监控端口，可以用curl查看内部监控metric统计情况；端口<=0表示关闭对应监听。
full_sync.http_port = 9101
incr_sync.http_port = 9100
# Prometheus metrics endpoint. Set the port to a value <= 0 to disable it.
# Prometheus 监控端口，访问 /metrics 获取指标；端口<=0表示关闭对应监听。
prom.http_port = 9102
# profiling on net/http/profile
# profiling端口，用于查看内部go堆栈。
system_profile_port = 9200

# global log level: debug, info, warning, error. lower level message will be filter
log.level = info
# log directory. log and pid file will be stored into this file.
# if not set, default is "./logs/"
# log和pid文件的目录，如果不设置默认打到当前路径的logs目录。
log.dir = logs
# log file name.
# log文件名。
log.file = collector_27017.log
# log flush enable. If set false, logs may not be print when exit. If
# set true, performance will be decreased extremely
# 设置log刷新，false表示包含缓存，如果true那么每条log都会直接刷屏，但对性能有影响；
# 反之，退出不一定能打印所有的log，调试时建议配置true。
log.flush = false
# log rotate when file size reaches this value in megabytes.
# 日志文件达到指定大小（MB）时轮转。
log.max_size_mb = 20
# keep rotated log files for this many days.
# 历史日志文件按天数保留。
log.max_age = 7

# sync mode: all/full/incr. default is incr.
# all means full synchronization + incremental synchronization.
# full means full synchronization only.
# incr means incremental synchronization only.
# 同步模式，all表示全量+增量同步，full表示全量同步，incr表示增量同步。
#sync_mode = all
sync_mode = incr

# connect source mongodb, set username and password if enable authority. Please note: password shouldn't contain '@'.
# split by comma(,) if use multiple instance in one replica-set. E.g., mongodb://username1:password1@primaryA,secondaryB,secondaryC
# split by semicolon(;) if sharding enable. E.g., mongodb://username1:password1@primaryA,secondaryB,secondaryC;mongodb://username2:password2@primaryX,secondaryY,secondaryZ
# 源MongoDB连接串信息，逗号分隔同一个副本集内的结点，分号分隔分片sharding实例，免密模式可以忽略“username:password@”
# 注意，密码里面如果含有'@'等特殊字符的话，shake会按照如下的URL编码方式进行处理。
# https://help.aliyun.com/zh/mongodb/support/connection-access-and-network
# 举例：
# 副本集：mongodb://username1:password1@primaryA,secondaryB,secondaryC
# 分片集：mongodb://username1:password1@primaryA,secondaryB,secondaryC;mongodb://username2:password2@primaryX,secondaryY,secondaryZ
mongo_urls = mongodb://XXXX:XXXX@XXXX:27017,XXXX:27017,XXXX:27017
# please fill the source config server url if source mongodb is sharding.
mongo_cs_url =
# please give at least one mongos address if source is sharding.
# 如果源端采用change stream拉取，这里还需要配置至少一个mongos的地址，多个mongos地址以逗号（,）分割
mongo_s_url =
# enable source ssl
mongo_ssl_root_ca_file =

# tunnel pipeline type. now we support rpc,file,kafka,mock,direct
# 通道模式。
tunnel = direct
# tunnel target resource url
# for rpc. this is remote receiver socket address
# for tcp. this is remote receiver socket address
# for file. this is the file path, for instance "data"
# for kafka. this is the topic and brokers address which split by comma, for
# instance: topic@brokers1,brokers2, default topic is "mongoshake"
# for mock. this is useless
# for direct. this is target mongodb address which format is the same as `mongo_urls`. If
# the target is sharding, this should be the mongos address.
# direct模式用于直接写入MongoDB，其余模式用于一些分析，或者远距离传输场景，
# 注意，如果是非direct模式，需要通过receiver进行解析，具体参考FAQ文档。
# 此处配置通道的地址，格式与mongo_urls对齐。
tunnel.address = mongodb://root:XXX@XXX:3717,XXX:3717
# the message format in the tunnel, used when tunnel is kafka.
# "raw": batched raw data format which has good performance but encoded so that users
# should parse it by receiver.
# "json": single oplog format by json.
# "bson": single oplog format by bson.
# 通道数据的类型，只用于kafka和file通道类型。
# raw是默认的类型，其采用聚合的模式进行写入和
# 读取，但是由于携带了一些控制信息，所以需要专门用receiver进行解析。
# json以json的格式写入kafka，便于用户直接读取。
# bson以bson二进制的格式写入kafka。
tunnel.message = raw
# how many partitions will be written, use some hash function in "incr_sync.shard_key".
# 如果目的端是kafka，最多启用多少个partition，最大不超过"incr_sync.worker"。默认1
tunnel.kafka.partition_number = 1
# Whether or not to use SASL authentication when connecting to the broker
# (defaults to false).
tunnel.kafka.sasl.enable = false
# for kafka. this is the username and password auth which split by comma, for instance: username@password
# UserName is the authentication identity (authcid) to present for
# SASL/PLAIN or SASL/SCRAM authentication
# Password for SASL/PLAIN authentication
tunnel.kafka.sasl.auth =
# SASLMechanism is the name of the enabled SASL mechanism.
# Currently only supported values: PLAIN、SCRAM-SHA-256和SCRAM-SHA-512 (defaults to PLAIN).
tunnel.kafka.sasl.mechanism = PLAIN
#represents the various compression codecs recognized by Kafka in messages.
#supported values:none,gzip,snappy,lz4,zstd
tunnel.kafka.compression = none
#The maximum permitted size of a message (defaults to 18*1024*1024). Should be
#set equal to or smaller than the broker's `message.max.bytes`.
#tunnel.kafka.producer.max_message_bytes = 18*1024*1024
tunnel.kafka.producer.max_message_bytes = 18874368

# tunnel json format, it'll only take effect in the case of tunnel.message = json
# and tunnel == kafka. Set canonical_extended_json if you want to use "Canonical
# Extended JSON Format", #559.
# 写入异构通道的json格式。如果希望使用Canonical Extended Json Format，则设置为
# canonical_extended_json
tunnel.json.format =
# if tunnel == direct or kafka and enable ssl
tunnel.mongo_ssl_root_ca_file =

# connect mode:
# primary: fetch data from primary.
# secondaryPreferred: fetch data from secondary if has, otherwise primary.(default)
# standalone: fetch data from given 1 node, no matter primary, secondary or hidden. This is only
# support when tunnel type is direct.
# 连接模式，primary表示从主上拉取，secondaryPreferred表示优先从secondary拉取（默认建议值），
# standalone表示从任意单个结点拉取。
mongo_connect_mode = secondaryPreferred

# filter db or collection namespace. at most one of these two parameters can be given.
# if the filter.namespace.black is not empty, the given namespace will be
# filtered while others namespace passed.
# if the filter.namespace.white is not empty, the given namespace will be
# passed while others filtered.
# all the namespace will be passed if no condition given.
# db and collection connected by the dot(.).
# different namespaces are split by the semicolon(;).
# filter: filterDbName1.filterCollectionName1;filterDbName2
# 黑白名单过滤，目前不支持正则，白名单表示通过的namespace，黑名单表示过滤的namespace，
# 不能同时指定。分号分割不同namespace，每个namespace可以是db，也可以是db.collection。
filter.namespace.black =
filter.namespace.white = operation.gpsevents_202606;operation.gpsevents_202607;operation.gpsevents_202608;operation.gpsevents_202609;dry_operation.gpsevents_202306;dry_operation.gpsevents_202307;dry_operation.gpsevents_202308;dry_operation.gpsevents_202309;dry_operation.gpsevents_202310;dry_operation.gpsevents_202311;dry_operation.gpsevents_202312;dry_operation.gpsevents_202401;dry_operation.gpsevents_202402;dry_operation.gpsevents_202403;dry_operation.gpsevents_202404;dry_operation.gpsevents_202405;dry_operation.gpsevents_202406;dry_operation.gpsevents_202407;dry_operation.gpsevents_202408;dry_operation.gpsevents_202409;dry_operation.gpsevents_202410;dry_operation.gpsevents_202411;dry_operation.gpsevents_202412;dry_operation.gpsevents_202501;dry_operation.gpsevents_202502;dry_operation.gpsevents_202503;dry_operation.gpsevents_202504;dry_operation.gpsevents_202505;dry_operation.gpsevents_202506;dry_operation.gpsevents_202507;dry_operation.gpsevents_202508;dry_operation.gpsevents_202509;dry_operation.gpsevents_202510;dry_operation.gpsevents_202511;dry_operation.gpsevents_202512;dry_operation.gpsevents_202601;dry_operation.gpsevents_202602;dry_operation.gpsevents_202603;dry_operation.gpsevents_202604;dry_operation.gpsevents_202605;dry_operation.gpsevents_202606;dry_operation.gpsevents_202607;dry_operation.gpsevents_202608;dry_operation.gpsevents_202609;dry_operation.gpsevents_202610;dry_operation.gpsevents_202611;dry_operation.gpsevents_202612
# filter oplog by oplog op type. different ops are split by semicolon(;).
# valid values are "i", "u", "d", "c". keep empty to pass all oplogs.
# this setting matches oplog op types instead of only MongoDB command oplogs.
# for "i", "u", and "d", matching inner ops inside applyOps are also filtered.
# e.g. set "d" to drop delete oplogs such as TTL delete or batched delete from incremental sync.
# e.g. set "i;u" to drop matching insert/update ops even when they are nested inside applyOps.
# 按 oplog op 类型过滤。多个 op 以分号(;)分隔。合法值包括 "i"、"u"、"d"、"c"。
# 该配置匹配的是 oplog op 类型，不是只匹配 MongoDB command oplog。保持为空表示不过滤；
# 对于 "i"、"u"、"d"，会同时过滤顶层 oplog 和 applyOps 内层命中的 DML 操作；
# 例如配置为 d 可过滤增量同步中的 delete oplog（如 TTL 删除、batched delete）；
# 配置为 i;u 时，也会过滤 applyOps 内层命中的 insert/update。
filter.op_types =

# some databases like "admin", "local", "mongoshake", "config" are
# filtered, users can enable these database based on some special needs.
# different database are split by the semicolon(;).
# e.g., admin;mongoshake
# pay attention: collection isn't support like "admin.xxx"
# 正常情况下，不建议配置该参数，但对于有些非常特殊的场景，用户可以启用admin，mongoshake等库的同步，
# 以分号分割，例如：admin;mongoshake
filter.pass.special.db =
# only transfer oplog commands for syncing. represent
# by oplog.op are "i","d","u".
# DDL will be transferred if disable like create index, drop database,
# transaction in mongodb 4.0.
# 是否需要开启DDL同步，true表示开启，源是sharding暂时不支持开启。
# 如果目的端是sharding，暂时不支持applyOps命令，包括事务。
filter.ddl_enable = true
# filter oplog gid if enabled.
# 如果MongoDB启用了gid，但是目的端MongoDB不支持gid导致同步会失败，可以启用gid过滤，将会去掉gid字段。
# 谨慎建议开启，shake本身性能受损很大。
filter.oplog.gids = false

# checkpoint info, used in resuming from break point.
# checkpoint存储信息，用于支持断点续传。
# context.storage.url is used to mark the checkpoint store database. E.g., mongodb://127.0.0.1:20070
# if not set, checkpoint will be written into source mongodb(db=mongoshake)
# checkpoint的具体写入的MongoDB地址，如果不配置，对于副本集和分片集群都将写入源库(db=mongoshake)
# 2.4版本以后不需要配置为源端cs的地址。
checkpoint.storage.url = mongodb://root:XXX@XXX:3717,XXX:3717
# checkpoint db's name.
# checkpoint存储的db的名字
checkpoint.storage.db = mongoshake_timeseries
# checkpoint collection's name.
# checkpoint存储的表的名字，如果启动多个mongoshake拉取同一个源可以修改这个表名以防止冲突。
checkpoint.storage.collection = ckpt_default
# set if enable ssl
checkpoint.storage.url.mongo_ssl_root_ca_file =
# real checkpoint: the fetching oplog position.
# pay attention: this is UTC time which is 8 hours latter than CST time. this
# variable will only be used when checkpoint is not exist.
# 本次开始拉取的位置，如果checkpoint已经存在（位于上述存储位置）则该参数无效，
# 如果需要强制该位置开始拉取，需要先删除原来的checkpoint，详见FAQ。
# 若checkpoint不存在，且该值为1970-01-01T00:00:00Z，则会拉取源端现有的所有oplog。
# 若checkpoint不存在，且该值不为1970-01-01T00:00:00Z，则会先检查源端oplog最老的时间是否
# 大于给定的时间，如果是则会直接报错退出。
#checkpoint.start_position = 1970-01-01T00:00:00Z
checkpoint.start_position = 2026-09-07T00:00:00Z

# transform from source db or collection namespace to dest db or collection namespace.
# at most one of these two parameters can be given.
# transform: fromDbName1.fromCollectionName1:toDbName1.toCollectionName1;fromDbName2:toDbName2
# 转换命名空间，比如a.b同步后变成c.d，谨慎建议开启，比较耗性能。
transform.namespace =

# --------------------------- full sync configuration ---------------------------
# the number of collection concurrence
# 并发最大拉取的表个数，例如，6表示同一时刻shake最多拉取6个表。
full_sync.reader.collection_parallel = 6
# the number of document writer thread in each collection.
# 同一个表内并发写的线程数，例如，8表示对于同一个表，将会有8个写线程进行并发写入。
full_sync.reader.write_document_parallel = 12
# number of documents in a batch insert in a document concurrence
# 目的端写入的batch大小，例如，128表示一个线程将会一次聚合128个文档然后再写入。
full_sync.reader.document_batch_size = 128
# number of documents in a batch in fetch from source db
# 源端拉取batch中最大条数
full_sync.reader.fetch_batch_size = 1024
# max number of fetching thread per table. default is 1
# 单个表最大拉取的线程数，默认是单线程拉取。需要具备splitVector权限。
# 注意：对单个表来说，仅支持索引对应的value是同种类型，如果有不同类型请勿启用该配置项！
full_sync.reader.parallel_thread = 1
# the parallel query index if set full_sync.reader.parallel_thread. index should only has
# 1 field.
# 如果设置了full_sync.reader.parallel_thread，还需要设置该参数，并行拉取所扫描的index，value
# 必须是同种类型。对于副本集，建议设置_id；对于集群版，建议设置shard_key。key只能有1个field。
full_sync.reader.parallel_index = _id
#full_sync.reader.parallel_index = cltDateTime
# maxChunkSize for splitVector, default is 1024MB, this value should be [1,1024]MB since below versions:
# 4.2.24, 4.4.19, 5.0.13, 6.0.2, see more details in: https://github.com/alibaba/MongoShake/issues/923
# 要使用单表并发拉取，splitVector命令的maxChunkSize从某些小版本开始就不能超过1024MB，否则会报错。(4.2.24, 4.4.19, 5.0.13, 6.0.2)
# 详情可参考如下issue中的讨论：https://github.com/alibaba/MongoShake/issues/923
full_sync.reader.split_max_chunk_size = 1024

# drop the same name of collection in dest mongodb in full synchronization
# 同步时如果目的库存在，是否先删除目的库再进行同步，true表示先删除再同步，false表示不删除。
full_sync.collection_exist_drop = false

# create index option.
# none: do not create indexes.
# foreground: create indexes when data sync finish in full sync stage.
# background: create indexes when starting.
# 全量期间数据同步完毕后，是否需要创建索引。none表示不创建，foreground表示创建前台索引，background表示创建后台索引。
# 从4.2开始就是强制后台建，background option已经失去了其原本的语义，会被忽略。而且4.4+开始用两阶段索引创建流程，所有承载数据的节点都会
# 一起开始后台建。详情可参考：
# https://www.mongodb.com/docs/manual/core/index-creation/#comparison-to-foreground-and-background-builds
full_sync.create_index = background

# convert insert to update when duplicate key found
# 如果_id存在在目的库，是否将insert语句修改为update语句。
full_sync.executor.insert_on_dup_update = true
# filter orphan document for source type is sharding.
# 源端是sharding，是否需要过滤orphan文档
full_sync.executor.filter.orphan_document = false
# enable majority write in full sync.
# the performance will degrade if enable.
# 全量阶段写入端是否启用majority write
full_sync.executor.majority_enable = false

# do not make destination collection sharding if source collection is sharding
# 不要在目标端执行shardCollection的操作，仅在从分片到副本集的同步场景才需要设置为true
full_sync.do_not_shard_destination = false

# --------------------------- incremental sync configuration ---------------------------
# fetch method:
# oplog: fetch oplog from source mongodb (default)
# change_stream: use change to receive change event from source mongodb, support MongoDB >= 4.0.
# we recommend to use change_stream if possible.
incr_sync.mongo_fetch_method = change_stream
#incr_sync.mongo_fetch_method = oplog

# After the document is updated, the fields that only need to be updated are set to false,
# and the contents of all documents are set to true
# 更新文档后,只需要更新的字段则设为false,需要全部文档内容则设为true
# 只在mongo_fetch_method = change_stream 模式下生效，且性能有所下降
incr_sync.change_stream.watch_full_document = true

# global id. used in active-active replication.
# this parameter is not supported on current open-source version.
# gid用于双活防止环形复制，目前只用于阿里云云上MongoDB，如果是阿里云云上实例互相同步
# 希望开启gid，请联系阿里云售后，sharding的有多个gid请以分号(;)分隔。
incr_sync.oplog.gids =

# distribute data to different worker by hash key to run in parallel.
# [auto] decide by if there has unique index in collections.
# use `collection` if has unique index otherwise use `id`.
# [id] shard by ObjectId. handle oplogs in sequence by unique _id
# [collection] shard by ns. handle oplogs in sequence by unique ns
# hash的方式，id表示按文档hash，collection表示按表hash，auto表示自动选择hash类型。
# 如果没有索引建议选择id达到非常高的同步性能，反之请选择collection。
incr_sync.shard_key = collection
# if shard_key is collection, and users want to improve performance when some collections
# do not have unique key.
# 对于按collection哈希，如果某些表不具有唯一索引，则可以设置按_id哈希以提高并发度。
# 用户需要确认该表不会创建唯一索引，一旦检测发现存在唯一索引，则会立刻crash退出。
# 例如，db1.collection1;db2.collection2，不支持仅指定db
incr_sync.shard_by_object_id_whitelist =

# oplog transmit worker concurrent
# if the source is sharding, worker number must equal to shard numbers.
# 内部发送（写目的DB）的worker数目，如果机器性能足够，可以提高worker个数。
incr_sync.worker = 32

# how many writing threads will be used in one worker.
# 对于目的端是kafka等非direct tunnel，启用多少个序列化线程，必须为"incr_sync.worker"的倍数。
# 默认为"incr_sync.worker"的值。
incr_sync.tunnel.write_thread = 32

# set the sync delay just like mongodb secondary slaveDelay parameter. unit second.
# 设置目的端的延迟，比如延迟源端20分钟，类似MongoDB本身主从同步slaveDelay参数，单位：秒
# 0表示不启用
incr_sync.target_delay = 0

# memory queue configuration, plz visit FAQ document to see more details.
# do not modify these variables if the performance and resource usage can
# meet your needs.
# 内部队列的配置参数，如果目前性能足够不建议修改，详细信息参考FAQ。
# batch_queue_size：每个worker线程的队列长度，worker线程从此队列取任务
# batching_max_size：一次分发给worker的任务最多包含多少个文档
# buffer_capacity：PendingQueue队列中一个buffer至少包含的文档个数，进行序列化
# buffer_size_threshold_in_kb：PendingQueue队列中一个buffer最多包含多少个字节
# fetch_batch_size：源端拉取batch的最大条数
incr_sync.worker.batch_queue_size = 64
incr_sync.adaptive.batching_max_size = 1024
incr_sync.fetcher.buffer_capacity = 64
incr_sync.fetcher.buffer_size_threshold_in_kb = 512
incr_sync.reader.fetch_batch_size = 8192

# --- direct tunnel only begin ---
# if tunnel type is direct, all the below variable should be set
# 下列参数仅用于tunnel为direct的情况。

# oplog changes to Insert while Update found non-exist (_id or unique-index)
# 如果_id不存在在目的库，是否将update语句修改为insert语句。
incr_sync.executor.upsert = false
#incr_sync.executor.upsert = true
# oplog changes to Update while Insert found duplicated key (_id or unique-index)
# 如果_id存在在目的库，是否将insert语句修改为update语句。
incr_sync.executor.insert_on_dup_update = false
#incr_sync.executor.insert_on_dup_update = true
# options:db,none
# db. write duplicated logs to mongoshake_conflict
# 如果写入存在冲突，记录冲突的文档。选项：db, none
# db：冲突写到目的DB的mongoshake_conflict库中
incr_sync.conflict_write_to = db

# enable majority write in incremental sync.
# the performance will degrade if enable.
# 增量阶段写入端是否启用majority write
incr_sync.executor.majority_enable = false
# whether executor should bypass document validation or not.
# 增量阶段写入是否临时绕开文档验证机制
incr_sync.executor.bypass_document_validation = true
#incr_sync.executor.bypass_document_validation = false

#incr_sync.executor.dup_key_strategy = skip

# --- direct tunnel only end ---

# 特殊字段，标识源端类型，默认为空。阿里云MongoDB serverless集群请配置aliyun_serverless
special.source.db.flag =

# 特殊字段，当源库和目标库都是分片集群时，且源表和目标表片键不一致，是否跳过检验
#例如，db1.collection1;db2.collection2
skip.nsshardkey.verify =

```