DBZ iterator migration to use SourceStateIterator #36333

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Merged

xiaohansong merged 11 commits into master from xiaohan/dbz-iterator

Mar 26, 2024

Contributor

xiaohansong commented Mar 20, 2024 •

edited

Loading

Move DBZ iterator to use SourceStateIterator

Apply this on postgres

xiaohansong added 3 commits

March 12, 2024 16:17


          dbz iterator refactor

c240454


          Mt``erge remote-tracking branch 'origin/master' into xiaohan/dbz-iter…

cd2b912

…ator


          dbz iterator

6de8269

vercel bot commented Mar 20, 2024 •

edited

Loading

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Mar 26, 2024 4:29am

octavia-squidington-iii added the CDK label


          add postgres to use new cdk

4cb382c

octavia-squidington-iii added the connectors/source/postgres label

xiaohansong requested a review from akashkulk

March 20, 2024 22:36

xiaohansong marked this pull request as ready for review

March 20, 2024 23:55

xiaohansong requested a review from a team as a code owner

March 20, 2024 23:55


          Merge branch 'master' into xiaohan/dbz-iterator

d15f205

octavia-squidington-iii added the area/connectors label

vercel bot deployed to Preview

March 20, 2024 23:59

View deployment

postamar reviewed

View reviewed changes

...dk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/AirbyteDebeziumHandler.java Outdated Show resolved Hide resolved

airbyte-integrations/connectors/source-postgres/build.gradle Outdated

@@ @@ -14,7 +14,7 @@ java { @@
               airbyteJavaConnector {
                   cdkVersionRequired = '0.23.19'
                   features = ['db-sources', 'datastore-postgres']
-                  useLocalCdk = false
+                  useLocalCdk = true

Contributor

postamar Mar 25, 2024

In order to properly validate this change, it needs to be applied to at least one connector.

Contributor Author

xiaohansong Mar 25, 2024

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.

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated

+                  }
+                  if (offsetManager == null) {
+                    throw new RuntimeException("Offset can not be null");
+                  }

Contributor

postamar Mar 25, 2024

Move this check to the constructor.

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java

+                    // include all necessary data such as snapshot completion.
+                    // This is the case for MS SQL Server, at least.
+                    return createStateMessage(initialOffset);
+                  }

Contributor

postamar Mar 25, 2024

We might be able to get rid of this check entirely at this point. Try it out and see if source-mssql tests pass

Contributor Author

xiaohansong Mar 25, 2024

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

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated Show resolved Hide resolved


          comment fix

f481150

octavia-squidington-iii added the connectors/source/mssql label

akashkulk reviewed

View reviewed changes

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java

+                  previousCheckpointOffset.clear();
+                  previousCheckpointOffset.putAll(checkpointOffsetToSend);
+                  shouldEmitStateMessage = false;
+                  return stateMessage;

Contributor

akashkulk Mar 25, 2024

should we call resetCheckpointValues() i.e. call checkpointOffsetToSend.clear(); before returning a state message?

Also we can probably replace resetCheckpointValues() with checkpointOffsetToSend.clear()`

Contributor Author

xiaohansong Mar 25, 2024

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

akashkulk reviewed

View reviewed changes

...es/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumMessageProducer.java Outdated

+                  this.schemaHistoryManager = schemaHistoryManager;
+                  this.previousCheckpointOffset = (HashMap<String, String>) offsetManager.read();
+                  this.initialOffset = new HashMap<>(this.previousCheckpointOffset);
+                  resetCheckpointValues();

Contributor

akashkulk Mar 25, 2024

nit : not sure if there is a need for this? I understand you're porting over code from before

Contributor Author

xiaohansong Mar 25, 2024

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.

akashkulk suggested changes

View reviewed changes

Contributor

akashkulk left a comment

some small nits


          reset values

17b1bd1

akashkulk approved these changes

View reviewed changes

xiaohansong added 2 commits

March 25, 2024 16:57


          Merge remote-tracking branch 'origin/master' into xiaohan/dbz-iterator

fee495e


          publish cdk

e2d1ed4

xiaohansong requested a review from a team as a code owner

March 25, 2024 23:59

Contributor Author

xiaohansong commented Mar 26, 2024 •

edited by github-actions bot

Loading

/publish-java-cdk

🕑 https://github.com/airbytehq/airbyte/actions/runs/8430583675
✅ Successfully published Java CDK version=0.27.4!


          bump up version for postgres and mssql

979336b

Contributor Author

xiaohansong commented Mar 26, 2024

also manually tested on postgres and mssql on cdc incremental sync.

octavia-squidington-iii added the area/documentation label

vercel bot deployed to Preview

March 26, 2024 04:11

View deployment


          psotgres doc bump

3b38c44

vercel bot deployed to Preview

March 26, 2024 04:29

View deployment

xiaohansong merged commit 1b99f1a into master

30 checks passed

xiaohansong deleted the xiaohan/dbz-iterator branch

March 26, 2024 04:50

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

236960f

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

4f61b78

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

b2af2eb

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

fd38ace

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

75c80ce

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

a6a86b6

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

0a16314

stephane-airbyte added a commit that referenced this pull request


          convert #36333 to kotlin

873ca2b

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area/connectors area/documentation CDK connectors/source/mssql connectors/source/postgres