Cloud Kafka-Couchbase N1QL Sink Connector 静默数据丢失问题求助

我们基于Couchbase实现了使用N1qlSinkHandler的Kafka Sink Connector，配置如下：

{    
    "name": "test_couchbasesink",    
    "config": {        
        "connector.class": "com.couchbase.connect.kafka.CouchbaseSinkConnector",        
        "tasks.max": "3",        
        "topics": "path.lite.billing.ack",        
        "couchbase.seed.nodes": "*******",        
        "couchbase.bootstrap.timeout": "60s",        
        "couchbase.bucket": "******",        
        "couchbase.default.collection": "******",        
        "couchbase.username": "******",        
        "couchbase.password": "******",        
        "couchbase.ssl.enabled": true,        
        "couchbase.enable.tls": true,        
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",        
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",        
        "key.converter.schemas.enable": "false",        
        "value.converter.schemas.enable": "false",        
        "transforms": "convertBillingAck",        
        "couchbase.document.id": "${/id}",        
        "couchbase.sink.handler": "com.couchbase.connect.kafka.handler.sink.N1qlSinkHandler",        
        "couchbase.n1ql.operation": "UPDATE",        
        "couchbase.n1ql.where.fields": "",        
        "transforms.convertBillingAck.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",        
        "transforms.convertBillingAck.field": "billingAck",        
        "transforms.convertBillingAck.format": "yyyy-MM-dd'T'HH:mm:ss.SSSXXX",        
        "transforms.convertBillingAck.target.type": "string",        
        "errors.tolerance": "all",        
        "errors.log.enable": "true",        
        "errors.log.include.messages": "true",        
        "errors.deadletterqueue.topic.name": "dlq.ack",        
        "errors.deadletterqueue.context.headers.enable": "true"    
    }
}

生产环境出现静默数据丢失：每接收1000条记录约丢失60-70条，无任何确认或报错信息；Cloud Kafka无异常日志，丢失的记录也未进入配置的DLQ主题。

我们推测原因是：Kafka读取记录后立即提交偏移量，未等待Couchbase的操作确认——N1QL本身不支持KV操作的持久性选项，而当前配置未强制Connector等待N1QL执行结果。

限制条件：

• 需要实现文档部分更新，无法切换到支持持久性的默认KV处理器
• SubdocumentKVUpdate处理器在Confluent Cloud环境中不被支持

解决方案及优化配置项 #

1. 启用N1QL一致性保障 #

添加couchbase.n1ql.consistency配置，设置为REQUEST_PLUS，确保Connector在提交Kafka偏移量前，等待Couchbase完成N1QL操作并返回确认，避免偏移量提前提交导致数据丢失：

"couchbase.n1ql.consistency": "REQUEST_PLUS"

2. 调整Kafka偏移量提交策略 #

修改偏移量提交相关配置，强制Connector仅在Couchbase操作确认后再提交偏移量：

"offset.flush.interval.ms": "0",  // 关闭定时自动提交，仅手动提交
"offset.flush.timeout.ms": "30000"  // 给操作确认预留足够超时时间

3. 修正N1QL WHERE条件配置 #

当前couchbase.n1ql.where.fields为空，可能导致生成的UPDATE语句未精准匹配目标文档，部分记录因条件不匹配被静默跳过。需明确指定匹配字段：

"couchbase.n1ql.where.fields": "id"

此配置会生成UPDATE bucket.collection SET ... WHERE id = $id的N1QL语句，确保仅更新指定ID的文档，同时让Connector能验证操作是否命中目标文档。

4. 优化错误处理逻辑 #

当前errors.tolerance: "all"会忽略所有错误（包括N1QL执行失败），导致失败记录既不进入DLQ也不触发重试。建议调整为：

"errors.tolerance": "none",  // 遇到错误时停止任务，避免静默丢失
"errors.retry.timeout.ms": "60000",  // 重试超时时间
"errors.retry.delay.max.ms": "5000"  // 最大重试间隔

若需保留容错能力，可将errors.tolerance设为all但添加重试配置，确保临时失败的记录有机会被重新处理。

5. 自定义N1QL语句（可选） #

如果需要更精准的操作验证，可自定义N1QL语句并添加RETURNING子句，让Connector获取操作结果，确认更新是否成功：

"couchbase.n1ql.statement": "UPDATE `bucket_name`.`collection_name` SET billingAck = $billingAck WHERE id = $id RETURNING *"

替换bucket_name和collection_name为实际值，Connector会根据返回结果判断操作是否生效，未生效的记录会触发错误处理流程。

内容的提问来源于stack exchange，提问作者Krishnamraju B