MongoDB-Kafka连接器数据同步异常：如何重置Kafka偏移量？

Absolutely—resetting Kafka consumer offsets (what you're describing as "clearing Kafka cache/offsets") is the exact recommended solution for this specific issue, and it should get your connector capturing MongoDB events again. Let me break down why this works and how to do it properly:

Why This Happens #

The warning you're seeing (WARN Failed to resume change stream: Resume of change stream was not possible, as the resume point may no longer be in the oplog. 286) tells the full story:

• The MongoDB source connector stores its change stream resume token linked directly to Kafka consumer offsets.
• When MongoDB purges old oplog entries (to free up disk space), the resume token tied to your connector's stored Kafka offset becomes invalid.
• The connector gets stuck trying to resume from a point that no longer exists, so it stops processing new events—even though it still shows a RUNNING status.

Step-by-Step Fix: Reset Kafka Consumer Offsets #

Here's how to resolve this:

1. (Optional but Recommended) Pause the Connector
   Before resetting offsets, pause your MongoDB source connector to avoid partial data writes during the process. Use the Kafka Connect REST API:

   curl -X PUT http://<your-connect-host>:8083/connectors/<your-mongo-connector-name>/pause
   

2. Verify Current Offset Status
   First, confirm your connector's consumer group and target topic:

   kafka-consumer-groups.sh --bootstrap-server <your-kafka-broker>:9092 --describe --group <your-connector-consumer-group-id>
   

   You’ll likely see that the current offset is far behind the latest available offset for your topic.

3. Reset the Offsets
   Choose an option based on your data needs:

   • Reset to latest offset (no duplicate data)
     This makes the connector start capturing events from right now, skipping any data generated while it was stuck:

     kafka-consumer-groups.sh --bootstrap-server <your-kafka-broker>:9092 --reset-offsets --to-latest --topic <your-target-kafka-topic> --group <your-connector-consumer-group-id> --execute
     

   • Reset to earliest offset (full data sync, may duplicate existing data)
     If you need to reprocess all historical data (and your downstream systems can handle duplicates), use this along with setting copy.existing=true in your connector config:

     kafka-consumer-groups.sh --bootstrap-server <your-kafka-broker>:9092 --reset-offsets --to-earliest --topic <your-target-kafka-topic> --group <your-connector-consumer-group-id> --execute
     

4. Resume the Connector
   Restart the connector to apply the new offsets:

   curl -X PUT http://<your-connect-host>:8083/connectors/<your-mongo-connector-name>/resume
   

   After this, the connector will establish a new valid change stream resume token and start processing new MongoDB events again.

Preventative Measures #

To avoid this issue recurring:

• Increase MongoDB Oplog Size: Adjust your cluster's oplog retention window to be longer than the maximum expected downtime or lag for your connector. This ensures the resume point stays in the oplog.
• Monitor Offset Lag: Set up alerts for when your connector's consumer offset lags too far behind the latest topic offset—this lets you address issues before the oplog expires the resume point.
• Use Idempotent Sinks: If you rely on copy.existing=true, ensure your sink connectors can handle duplicate records without unintended side effects.

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