DBZ iterator migration to use SourceStateIterator#36333
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
Octavia Squidington III (octavia-squidington-iii)
added
the
CDK
Octavia Squidington III (octavia-squidington-iii)
added
the
area/connectors
| cdkVersionRequired = '0.23.19' | ||
| features = ['db-sources', 'datastore-postgres'] | ||
| useLocalCdk = false | ||
| useLocalCdk = true |
There was a problem hiding this comment.
In order to properly validate this change, it needs to be applied to at least one connector.
There was a problem hiding this comment.
yep - this change exactly serves this before releasing cdk - so all tests on postgres would pass with this change. once approved i'll need to release cdk and bump up the required version here too.
| } | ||
| if (offsetManager == null) { | ||
| throw new RuntimeException("Offset can not be null"); | ||
| } |
There was a problem hiding this comment.
Move this check to the constructor.
| // include all necessary data such as snapshot completion. | ||
| // This is the case for MS SQL Server, at least. | ||
| return createStateMessage(initialOffset); | ||
| } |
There was a problem hiding this comment.
We might be able to get rid of this check entirely at this point. Try it out and see if source-mssql tests pass
There was a problem hiding this comment.
it will break some test - it's minor though, turns out state in global state will have some minor changes:
before:
"mssql_cdc_offset": {
"[\"db_ziwfjgytty\",{\"server\":\"db_ziwfjgytty\",\"database\":\"db_ziwfjgytty\"}]": "{\"commit_lsn\":\"00000029:00000298:0003\",\"snapshot\":true,\"snapshot_completed\":true}"
},
after:
"mssql_cdc_offset": {
"[\"db_ziwfjgytty\",{\"server\":\"db_ziwfjgytty\",\"database\":\"db_ziwfjgytty\"}]": "{\"transaction_id\":null,\"event_serial_no\":0,\"commit_lsn\":\"00000029:00000298:0003\",\"change_lsn\":\"NULL\"}"
},
I'll revert this change for now
| previousCheckpointOffset.clear(); | ||
| previousCheckpointOffset.putAll(checkpointOffsetToSend); | ||
| shouldEmitStateMessage = false; | ||
| return stateMessage; |
There was a problem hiding this comment.
should we call resetCheckpointValues() i.e. call checkpointOffsetToSend.clear(); before returning a state message?
Also we can probably replace resetCheckpointValues() with checkpointOffsetToSend.clear()`
There was a problem hiding this comment.
that's a great catch, thanks! that actually solved test issue I've had since I turned up the mssql. Surprisingly tests on postgres did not capture that
| this.schemaHistoryManager = schemaHistoryManager; | ||
| this.previousCheckpointOffset = (HashMap<String, String>) offsetManager.read(); | ||
| this.initialOffset = new HashMap<>(this.previousCheckpointOffset); | ||
| resetCheckpointValues(); |
There was a problem hiding this comment.
nit : not sure if there is a need for this? I understand you're porting over code from before
There was a problem hiding this comment.
removed. yeah initial value has been empty. Previously it also reset other stuff but I've deleted most of them so it's not applicable here.
Akash Kulkarni (akashkulk)
left a comment
Akash Kulkarni (akashkulk)
left a comment
There was a problem hiding this comment.
some small nits
|
/publish-java-cdk
|
|
also manually tested on postgres and mssql on cdc incremental sync. |
Octavia Squidington III (octavia-squidington-iii)
added
the
area/documentation
4 participants
Move DBZ iterator to use SourceStateIterator
Apply this on postgres