cdc

Go MongoDB CDC

Go MongoDB CDC captures and processes real-time changes from MongoDB using Change Streams.

Features

• Real-time Data Capture - Instantly captures INSERT, UPDATE, DELETE and REPLACE operations using MongoDB Change Streams API
• Scalable Partition System - Intelligent partition management that distributes workload across multiple workers
• Automatic Failover - Automatic partition transfer and load balancing between workers
• Oplog Rollover Protection - Automatic re-snapshot mechanism when oplog history is lost, ensuring zero data loss (similar to Debezium's snapshot.mode = "when_needed")
• Smart Resume Token Recovery - Automatic detection and recovery from expired/invalid resume tokens with graceful fallback
• Universal Hash Distribution - Advanced multi-hash algorithm ensuring optimal distribution for any ID pattern
• Bootstrap Mode - Full collection scanning feature for processing existing data on first run
• Advanced Checkpoint System - Comprehensive checkpoint mechanism to prevent data loss
• Sharded Cluster Support - Works with both replica sets and sharded clusters
• Comprehensive Metrics - Detailed performance and system metrics with Prometheus
• Easily manageable configurations - Simple configuration with YAML and JSON files

Example

package main

import (
	"context"

	"github.com/Trendyol/go-mongo-cdc/logger"

	cdc "github.com/Trendyol/go-mongo-cdc"

	"log"
	"time"

	"github.com/Trendyol/go-mongo-cdc/config"
	"github.com/Trendyol/go-mongo-cdc/mongo/message"
	"github.com/Trendyol/go-mongo-cdc/stream"
)

func main() {
	cfg := config.Config{
		MongoDB: config.MongoDB{
			Connection: config.Connection{
				URI:        "localhost:27017",
				Database:   "exampleDB",
				Collection: "exampleCollection",
			},
			ConnectionPool: config.ConnectionPool{
				MaxPoolSize:   100,
				MinPoolSize:   5,
				MaxIdleTimeMS: 300000,
			},
			Timeouts: config.Timeouts{
				ConnectTimeoutMS:         30000,
				ServerSelectionTimeoutMS: 60000,
				SocketTimeoutMS:          120000,
			},
		},
		Metric: config.MetricConfig{
			Port: 8080,
		},
	Checkpoint: config.CheckpointConfig{
		TokenSaveInterval:       10 * time.Second,
		ChangeStreamSaveCount:   500,
		BootstrapSaveCount:      5000,
		BootstrapSaveInterval:   10 * time.Second,
		BootstrapQueryBatchSize: 5000,
	},
		Partition: config.PartitionConfig{
			ConsumerGroup:          "myapp",
			HeartbeatInterval:      10 * time.Second,
			WorkerTimeout:          90 * time.Second,
			RebalanceCheckInterval: 10 * time.Second,
			TotalPartition:         30,
		},
		Logger: config.LoggerConfig{LogLevel: "debug"},
	}

	connector, err := cdc.NewConnector(cfg, ProcessChangeEvent)
	if err != nil {
		log.Fatal("failed to create connector:", err)
	}

	defer connector.Close()

	ctx := context.Background()
	connector.Start(ctx)
}

func ProcessChangeEvent(lc *stream.ListenerContext) error {
	select {
	case <-lc.Context.Done():
		logger.Log.Info("Shutdown signal received, stopping event processing")
		return lc.Context.Err()
	default:
	}

	switch lc.Message.OperationType {
	case message.OperationInsert, message.OperationUpdate, message.OperationReplace:
		if lc.Message.FullDocument != nil {
			logger.Log.Info("Document changed - operation: %s, document: %v, partitionId: %d", string(lc.Message.OperationType), lc.Message.DocumentID, lc.PartitionID)
		}
	case message.OperationDelete:
		logger.Log.Info("Document deleted - documentId: %v, partitionId: %d", lc.Message.DocumentID, lc.PartitionID)
	}

	if err := lc.Ack(); err != nil {
		logger.Log.Error("Failed to acknowledge message: %v", err)
		return err
	}
	return nil
}

Usage

$ go get github.com/Trendyol/go-mongo-cdc

Configuration

MongoDB Configuration

Connection Settings (mongodb.connection)

Connection Pool Settings (mongodb.connectionPool)

Timeout Settings (mongodb.timeouts)

Metric Configuration

Note: Monitoring oplog is critical for Change Streams. If the oplog window is too small or oplog history is lost, Change Streams may fail. We recommend using MongoDB's native metrics exposed through MongoDB Exporter or MongoDB's built-in monitoring tools to track oplog health (size, used percentage, window).

Checkpoint Configuration

Partition Configuration

Logger Configuration

Graceful Shutdown Configuration

Configuration Example

cdcconfig:
  mongodb:
    connection:
      uri: "localhost:27017"
      database: exampleDB
      collection: exampleCollection
    connectionPool:
      maxPoolSize: 100
      minPoolSize: 3
      maxIdleTimeMS: 300000

  metric:
    port: 8085

  checkpoint:
    bootstrapSaveCount: 5000
    bootstrapQueryBatchSize: 5000
    bootstrapSaveInterval: 10s
    tokenSaveInterval: 10s
    changeStreamSaveCount: 500

  partition:
    heartbeatInterval: 10s
    workerTimeout: 90s
    rebalanceCheckInterval: 10s
    totalPartition: 15
    workersCollection: "workers"
    partitionsCollection: "partition_assignments"
    consumerGroup: "myapp"

  logger:
    logLevel: "debug"

gracefulShutdownTimeout: 5s

appPort: :8080

Exposed Metrics

Examples

• Simple Example
• Grafana Dashboard Example

Contributing

Go MongoDB CDC is always open for direct contributions. For more information please check our Contribution Guideline document.

License

Released under the MIT License.