[Source-MySQL]CDC: Enable frequent state emission during incremental runs by nguyenaiden · Pull Request #29308 · airbytehq/airbyte

Duy Nguyen (nguyenaiden) · 2023-08-09T20:19:34Z

What

Enable checkpointing for MySQL CDC mode during incremental sync runs

How

Overrode 2 methods in MysqlCdcTargetPosition - isRecordBehindOffset and isSameOffset so that in the computeNext method of DebeziumStateDecoratingIterator to verify that the airbyte record processing events are not ahead of what Debezium has already processed and emit Airbyte state messages accordingly.
Sync log line indicating that checkpointing state has been emitted:

🚨 User Impact 🚨

Users should not see any functional impact to their syncs. An extra log line and checkpointing states will be emitted on a cadence.

Testing plan

Doc

github-actions · 2023-08-09T20:20:07Z

Before Merging a Connector Pull Request

Wow! What a great pull request you have here! 🎉

To merge this PR, ensure the following has been done/considered for each connector added or updated:

PR name follows PR naming conventions
Breaking changes are considered. If a Breaking Change is being introduced, ensure an Airbyte engineer has created a Breaking Change Plan.
Connector version has been incremented in the Dockerfile and metadata.yaml according to our Semantic Versioning for Connectors guidelines
You've updated the connector's metadata.yaml file any other relevant changes, including a breakingChanges entry for major version bumps. See metadata.yaml docs
Secrets in the connector's spec are annotated with airbyte_secret
All documentation files are up to date. (README.md, bootstrap.md, docs.md, etc...)
Changelog updated in docs/integrations/<source or destination>/<name>.md with an entry for the new version. See changelog example
Migration guide updated in docs/integrations/<source or destination>/<name>-migrations.md with an entry for the new version, if the version is a breaking change. See migration guide example
If set, you've ensured the icon is present in the platform-internal repo. (Docs)

If the checklist is complete, but the CI check is failing,

Check for hidden checklists in your PR description
Toggle the github label checklist-action-run on/off to re-run the checklist CI.

Octavia Squidington III (octavia-squidington-iii) · 2023-08-09T21:12:46Z

source-mysql test report (commit `10fc23225c`) - ❌

⏲️ Total pipeline duration: 20mn21s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-09T21:34:25Z

source-mysql test report (commit `10fc23225c`) - ❌

⏲️ Total pipeline duration: 19mn32s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-10T17:30:41Z

source-mysql test report (commit `3b43f276d7`) - ❌

⏲️ Total pipeline duration: 21mn30s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-10T19:52:02Z

source-mysql test report (commit `a009e1015b`) - ❌

⏲️ Total pipeline duration: 20mn12s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

Rodi Reich Zilberman (rodireich) · 2023-08-10T20:16:36Z

+      return false;
+    }
+
+    final String eventFileName = event.eventValueAsJson().get("source").get("file").asText();


Is there a case node would not exist?
need to check if there?

Rodi Reich Zilberman (rodireich) · 2023-08-10T20:24:23Z

+    final String offsetBFileName = offsetJsonB.get("file").asText();
+    final long offsetBPosition = offsetJsonB.get("pos").asLong();
+
+    if (offsetAFileName.compareTo(offsetBFileName) != 0) {


isn't this redundant?

Also just can we just simplify this to :
if ( (!offSetAFileName.equals(offsetBFileName)) {

Actually just
return offSetAFileName.equals(offsetBFileName) && offsetAPosition == offsetBPosition

I prefer return offSetAFileName.equals(offsetBFileName) && offsetAPosition == offsetBPosition

Akash Kulkarni (akashkulk) · 2023-08-15T21:29:49Z

+
+  @Override
+  public boolean isSameOffset(final Map<String, String> offsetA, final Map<String, String> offsetB) {
+    if (offsetA == null || offsetA.size() != 1) {


nit : can make checks for both offsets in one conditional

Akash Kulkarni (akashkulk) · 2023-08-15T21:34:25Z

+   */
+  @Test
+  protected void verifyCheckpointStatesByRecords() throws Exception {
+    // We require a huge amount of records, otherwise Debezium will notify directly the last offset.


IIRC there is a config option to control the number of state messages? so that it can be set to something smaller : example :

airbyte/airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java

Line 106 in a9dacb9

}

Can we use the same config to override the default state emission frequency for cdc incremental runs?

If you set that in the config, it will be respected :

airbyte/airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java

Line 117 in a9dacb9

config.get("sync_checkpoint_seconds") != null ? Duration.ofSeconds(config.get("sync_checkpoint_seconds").asLong())

By default for the tests we are already setting the sync_checkpoint_records to 1
.put("sync_checkpoint_records", 1) (here)

We already overriding it to 1 like Subodh said.

I see. I guess my question is why do we need a large number of records to create to test this? Is this because the offset (bin log file/pos) doesn't advance unless we are inserting such a large amount of records?

That is a good question. I don't think we need that many, but 20k records isn't a lot of records anyway. We can reduce this if we need to. Subodh Chaturvedi (@subodh1810) thoughts?

We need so many records because the offset file might not always be available to read (being edited by Debezium in parallel)

Akash Kulkarni (akashkulk) · 2023-08-15T21:55:26Z

+      return eventFileName.compareTo(offsetFileName) > 0;
+    }
+
+    return eventPosition > offsetPosition;


Judging by the code here we're evaluating whether the offset associated with the event is past the offset (which is inline with the logic of sending state messages). Do you think it makes sense to rename this interface method?

Yup! Even I agree with Akash, lets change the method name to isEventAheadOffset

Akash Kulkarni (akashkulk)

Some minor comments + questions

Subodh Chaturvedi (subodh1810) · 2023-08-16T05:40:49Z

+      return eventFileName.compareTo(offsetFileName) > 0;
+    }
+
+    return eventPosition > offsetPosition;


Yup! Even I agree with Akash, lets change the method name to isEventAheadOffset

Subodh Chaturvedi (subodh1810) · 2023-08-16T05:41:38Z

+    final String offsetBFileName = offsetJsonB.get("file").asText();
+    final long offsetBPosition = offsetJsonB.get("pos").asLong();
+
+    if (offsetAFileName.compareTo(offsetBFileName) != 0) {


I prefer return offSetAFileName.equals(offsetBFileName) && offsetAPosition == offsetBPosition

Subodh Chaturvedi (subodh1810) · 2023-08-16T05:43:22Z

+   */
+  @Test
+  protected void verifyCheckpointStatesByRecords() throws Exception {
+    // We require a huge amount of records, otherwise Debezium will notify directly the last offset.


By default for the tests we are already setting the sync_checkpoint_records to 1
.put("sync_checkpoint_records", 1) (here)

Akash Kulkarni (akashkulk) · 2023-08-16T22:47:52Z

   * @return Returns `true` when the record is behind the offset. Otherwise, it returns `false`
   */
-  default boolean isRecordBehindOffset(final Map<String, String> offset, final ChangeEventWithMetadata event) {
+  default boolean isEventAheadOffset(final Map<String, String> offset, final ChangeEventWithMetadata event) {


nit : update the comment on the interface method

Octavia Squidington III (octavia-squidington-iii) · 2023-08-16T23:08:08Z

source-mysql test report (commit `c291768937`) - ❌

⏲️ Total pipeline duration: 20mn12s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

combined conditionals for cleanliness

Bumped connector versions

Duy Nguyen (nguyenaiden) · 2023-08-21T17:52:28Z

/legacy-test connector=connectors/source-mysql

Duy Nguyen (nguyenaiden) · 2023-08-21T17:52:33Z

/legacy-test connector=connectors/source-mysql-strict-encrypt

github-actions · 2023-08-21T18:14:24Z

source-mysql test report (commit `57abadf4da`) - ❌

⏲️ Total pipeline duration: 20mn51s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	❌
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

github-actions · 2023-08-21T18:27:34Z

source-mysql-strict-encrypt test report (commit `57abadf4da`) - ❌

⏲️ Total pipeline duration: 12mn59s

Step	Result
Validate airbyte-integrations/connectors/source-mysql-strict-encrypt/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	❌
Build connector tar	✅
Build source-mysql-strict-encrypt docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql-strict-encrypt:integrationTest	❌
Acceptance tests	❌

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql-strict-encrypt test

Pedro S. Lopez (pedroslopez) · 2023-08-21T20:52:12Z

/legacy-test connector=connectors/source-mysql

Pedro S. Lopez (pedroslopez) · 2023-08-21T20:52:26Z

/legacy-test connector=connectors/source-mysql-strict-encrypt

github-actions · 2023-08-21T21:28:10Z

source-mysql test report (commit `d4f68b0def`) - ❌

⏲️ Total pipeline duration: 19mn43s

Step	Result
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql:integrationTest	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

github-actions · 2023-08-21T21:40:04Z

source-mysql-strict-encrypt test report (commit `d4f68b0def`) - ❌

⏲️ Total pipeline duration: 11mn42s

Step	Result
Validate airbyte-integrations/connectors/source-mysql-strict-encrypt/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	❌
Build connector tar	✅
Build source-mysql-strict-encrypt docker image for platform linux/x86_64	✅
./gradlew :airbyte-integrations:connectors:source-mysql-strict-encrypt:integrationTest	❌
Acceptance tests	❌

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql-strict-encrypt test

Duy Nguyen (nguyenaiden) · 2023-08-22T01:12:48Z

/approve-and-merge reason="All tests passed locally. Merging as CI is flaky"

Octavia The Approver (octavia-approvington) · 2023-08-22T01:13:29Z

It's time

github-actions · 2023-08-22T01:50:14Z

source-mysql test report (commit `54bba8bdc4`) - ❌

⏲️ Total pipeline duration: 36mn34s

Step	Result
Java Connector Unit Tests	✅
Build connector tar	✅
Build source-mysql docker image for platform linux/x86_64	✅
Java Connector Integration Tests	❌
Acceptance tests	✅
Validate airbyte-integrations/connectors/source-mysql/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql test

github-actions · 2023-08-22T02:30:17Z

source-mysql-strict-encrypt test report (commit `54bba8bdc4`) - ❌

⏲️ Total pipeline duration: 39mn53s

Step	Result
Java Connector Unit Tests	❌
Build connector tar	✅
Build source-mysql-strict-encrypt docker image for platform linux/x86_64	✅
Java Connector Integration Tests	❌
Acceptance tests	❌
Validate airbyte-integrations/connectors/source-mysql-strict-encrypt/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	❌

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-mysql-strict-encrypt test

…runs (airbytehq#29308)

Octavia Squidington III (octavia-squidington-iii) added area/connectors Connector related issues connectors/source/mysql labels Aug 9, 2023

Duy Nguyen (nguyenaiden) marked this pull request as ready for review August 9, 2023 20:50

Duy Nguyen (nguyenaiden) requested a review from a team as a code owner August 9, 2023 20:50

Duy Nguyen (nguyenaiden) requested review from Akash Kulkarni (akashkulk), Prateek Mukhedkar (prateekmukhedkar), Rodi Reich Zilberman (rodireich) and Subodh Chaturvedi (subodh1810) August 9, 2023 20:51

Rodi Reich Zilberman (rodireich) reviewed Aug 10, 2023

View reviewed changes

Akash Kulkarni (akashkulk) reviewed Aug 15, 2023

View reviewed changes

Akash Kulkarni (akashkulk) suggested changes Aug 15, 2023

View reviewed changes

Subodh Chaturvedi (subodh1810) approved these changes Aug 16, 2023

View reviewed changes

Duy Nguyen (nguyenaiden) force-pushed the mysql-cdc-checkpointing branch from a009e10 to c291768 Compare August 16, 2023 22:45

Akash Kulkarni (akashkulk) reviewed Aug 16, 2023

View reviewed changes

Akash Kulkarni (akashkulk) approved these changes Aug 16, 2023

View reviewed changes

nguyenaiden added 4 commits August 21, 2023 10:43

Enabled logging for MySQL CDC Incremental sync

82756c9

Add checkpointing to incremental runs for mysql CDC

143a24a

Renamed Target position methods

a275611

combined conditionals for cleanliness

Update comments to match with isEventAheadOffset

57abadf

Bumped connector versions

Duy Nguyen (nguyenaiden) force-pushed the mysql-cdc-checkpointing branch from c291768 to 57abadf Compare August 21, 2023 17:51

Octavia Squidington III (octavia-squidington-iii) added area/documentation Improvements or additions to documentation connectors/source/mysql-strict-encrypt labels Aug 21, 2023

Updated migration guide

d4f68b0

Merge branch 'master' into mysql-cdc-checkpointing

54bba8b

Octavia The Approver (octavia-approvington) approved these changes Aug 22, 2023

View reviewed changes

Octavia The Approver (octavia-approvington) merged commit b7dc462 into master Aug 22, 2023

Octavia The Approver (octavia-approvington) deleted the mysql-cdc-checkpointing branch August 22, 2023 01:13

Harry Touloupas (harrytou) pushed a commit to KYVENetwork/airbyte that referenced this pull request Sep 1, 2023

[Source-MySQL]CDC: Enable frequent state emission during incremental …

195bef3

…runs (airbytehq#29308)

Conversation

Duy Nguyen (nguyenaiden) commented Aug 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

How

Recommended reading order

🚨 User Impact 🚨

Testing plan

Uh oh!

github-actions Bot commented Aug 9, 2023 • edited by nguyenaiden Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Before Merging a Connector Pull Request

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 9, 2023

source-mysql test report (commit 10fc23225c) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 9, 2023

source-mysql test report (commit 10fc23225c) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 10, 2023

source-mysql test report (commit 3b43f276d7) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 10, 2023

source-mysql test report (commit a009e1015b) - ❌

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Akash Kulkarni (akashkulk) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 16, 2023

source-mysql test report (commit c291768937) - ❌

Uh oh!

Duy Nguyen (nguyenaiden) commented Aug 21, 2023

Uh oh!

Duy Nguyen (nguyenaiden) commented Aug 21, 2023

Uh oh!

github-actions Bot commented Aug 21, 2023

source-mysql test report (commit 57abadf4da) - ❌

Uh oh!

github-actions Bot commented Aug 21, 2023

source-mysql-strict-encrypt test report (commit 57abadf4da) - ❌

Uh oh!

Pedro S. Lopez (pedroslopez) commented Aug 21, 2023

Duy Nguyen (nguyenaiden) commented Aug 9, 2023 •

edited

Loading

github-actions Bot commented Aug 9, 2023 •

edited by nguyenaiden

Loading

source-mysql test report (commit `10fc23225c`) - ❌

source-mysql test report (commit `10fc23225c`) - ❌

source-mysql test report (commit `3b43f276d7`) - ❌

source-mysql test report (commit `a009e1015b`) - ❌

source-mysql test report (commit `c291768937`) - ❌

source-mysql test report (commit `57abadf4da`) - ❌

source-mysql-strict-encrypt test report (commit `57abadf4da`) - ❌

source-mysql test report (commit `d4f68b0def`) - ❌

source-mysql-strict-encrypt test report (commit `d4f68b0def`) - ❌

source-mysql test report (commit `54bba8bdc4`) - ❌

source-mysql-strict-encrypt test report (commit `54bba8bdc4`) - ❌