feat: Add GaussDB CDC connector with critical bug fixes #4195
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
## Summary Add complete GaussDB CDC connector implementation with support for snapshot and streaming modes using GaussDB's mppdb_decoding logical replication plugin. ## Key Features - Full snapshot support for initial data capture - Streaming CDC using GaussDB logical replication - Support for all common GaussDB data types - Configurable connection pooling and retry mechanisms - Comprehensive test suite ## Critical Bug Fixes ### 1. Default Value Converter Issue - **Problem**: GaussDB returns function calls (e.g., `pg_systimestamp()`, `CURRENT_TIMESTAMP`) as default values, causing Debezium to fail when trying to use them as actual values - **Solution**: Created `GaussDBDefaultValueConverter` that properly handles function-based defaults by returning `Optional.empty()` for them - **Files**: - `io/debezium/connector/gaussdb/GaussDBDefaultValueConverter.java` (new) - `io/debezium/connector/gaussdb/GaussDBSchema.java` (modified) ### 2. Missing Source Info Fields - **Problem**: Debezium envelope requires multiple fields (version, connector, name, snapshot) in the source struct, but they were not being set, causing validation errors - **Solution**: Added all required source info fields to both snapshot and streaming source struct builders - **Files**: - `GaussDBScanFetchTask.java:294-339` (snapshot) - `GaussDBStreamFetchTask.java:495-551` (streaming) ## Implementation Details ### Core Components - **GaussDBSource**: Main source implementation - **GaussDBDialect**: Dialect for GaussDB-specific SQL and behavior - **GaussDBConnection**: Connection management with retry logic - **GaussDBReplicationConnection**: Logical replication connection handling - **GaussDBScanFetchTask**: Snapshot data reading - **GaussDBStreamFetchTask**: Streaming CDC data reading ### Configuration - Hostname, port, database, username, password - Plugin name (mppdb_decoding) - Slot name for logical replication - Connection timeout and retry settings - Snapshot fetch size ### Testing - Unit tests for all major components - Integration tests for snapshot and streaming modes - Data type compatibility tests - Boundary condition tests ## Verified Configuration - ✅ GaussDB `wal_level = logical` (required for CDC) - ✅ `mppdb_decoding` plugin available and functional - ✅ Replication slot creation and management working ## Known Issues - Integration test timeout issue under investigation (fetch task execution) - Requires further debugging of Flink job initialization flow ## Dependencies - GaussDB JDBC driver (included in lib/) - Debezium PostgreSQL connector (for replication protocol) - Flink CDC base framework 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <[email protected]>
- Add SERIAL/BIGSERIAL support in GaussDBTypeUtils - Bundle PostgreSQL driver in flink-sql-connector-gaussdb-cdc to avoid NoClassDefFoundError - Implement GaussDbE2eITCase with robust replication slot cleanup - Improve replication connection stability with KeepAlive and socket timeout settings - Add diagnostic logging for GaussDB replication stream
…eption - Add getSlotNameForBackfillTask() to GaussDBSourceConfig returning slotName_backfill - Add backfillReplicationConnection field to GaussDBSourceFetchTaskContext - Add isBoundedRead() method to detect bounded reads (SnapshotSplit or StreamSplit with endOffset) - Modify createReplicationStream() to use backfill connection for bounded reads - Add endingPos field and reachEnd() check for bounded stream reads - Add 10ms stabilization delay after stream.start() This fixes the replication slot contention issue where GaussDB would fail with EOFException when trying to restart the replication stream during backfill phase. Tested E2E: - Snapshot: 2 records captured successfully - INSERT: id=103 captured and synced to MySQL - UPDATE: id=101 value change captured and synced
… support - Resolved TableId catalog mismatch in GaussDBSourceFetchTaskContext - Implemented robust polling mechanism in E2E tests for reliable verification - Added native UTF-8 encoding support for Chinese character synchronization - Enhanced documentation including READMEs and setup guides - Cleaned up duplicate JAR conflicts in deployment process
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
This PR adds a complete GaussDB CDC connector implementation with support for snapshot and streaming modes using GaussDB's mppdb_decoding logical replication plugin.
Key Features
Critical Bug Fixes
1. Default Value Converter Issue
pg_systimestamp(),CURRENT_TIMESTAMP) as default values, causing Debezium to fail when trying to use them as actual valuesGaussDBDefaultValueConverterthat properly handles function-based defaults by returningOptional.empty()for them2. Missing Source Info Fields
Implementation Details
Core Components
Configuration Options
Testing
Verified Configuration
wal_level = logical(required for CDC)mppdb_decodingplugin available and functionalKnown Issues
testReadSingleTableAllRecordstimes out (fetch task execution issue under investigation)Test Plan
Dependencies
Documentation
Checklist
🤖 Generated with Claude Code
Co-Authored-By: Claude Sonnet 4.5 [email protected]