[Source-MySQL]CDC: Enable frequent state emission during incremental runs #29308

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 · 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 · 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 · 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 · 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

rodireich · 2023-08-10T20:16:36Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

+      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?

rodireich · 2023-08-10T20:24:23Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

+    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

akashkulk · 2023-08-15T21:29:49Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

+
+  @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

akashkulk · 2023-08-15T21:34:25Z

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

+   */
+  @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. @subodh1810 thoughts?

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

akashkulk · 2023-08-15T21:55:26Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

+      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

akashkulk

Some minor comments + questions

subodh1810 · 2023-08-16T05:40:49Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

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


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

subodh1810 · 2023-08-16T05:41:38Z

...m/src/main/java/io/airbyte/integrations/debezium/internals/mysql/MySqlCdcTargetPosition.java

+    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

subodh1810 · 2023-08-16T05:43:22Z

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

+   */
+  @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)

akashkulk · 2023-08-16T22:47:52Z

...rations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/CdcTargetPosition.java

@@ -61,7 +61,7 @@ default boolean isHeartbeatSupported() {
   * @param event Event from the CDC load
   * @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 · 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

nguyenaiden · 2023-08-21T17:52:28Z

/legacy-test connector=connectors/source-mysql

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

pedroslopez · 2023-08-21T20:52:12Z

/legacy-test connector=connectors/source-mysql

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

nguyenaiden · 2023-08-22T01:12:48Z

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

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 added area/connectors Connector related issues connectors/source/mysql labels Aug 9, 2023

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

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

nguyenaiden requested review from rodireich, subodh1810, akashkulk and prateekmukhedkar August 9, 2023 20:51

rodireich reviewed Aug 10, 2023

View reviewed changes

akashkulk reviewed Aug 15, 2023

View reviewed changes

akashkulk suggested changes Aug 15, 2023

View reviewed changes

subodh1810 approved these changes Aug 16, 2023

View reviewed changes

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

akashkulk reviewed Aug 16, 2023

View reviewed changes

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

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

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-approvington approved these changes Aug 22, 2023

View reviewed changes

octavia-approvington merged commit b7dc462 into master Aug 22, 2023

octavia-approvington deleted the mysql-cdc-checkpointing branch August 22, 2023 01:13

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)

[Source-MySQL]CDC: Enable frequent state emission during incremental runs #29308

[Source-MySQL]CDC: Enable frequent state emission during incremental runs #29308

Conversation

nguyenaiden commented Aug 9, 2023 • edited Loading

What

How

Recommended reading order

🚨 User Impact 🚨

Testing plan

github-actions bot commented Aug 9, 2023 • edited by nguyenaiden Loading

Before Merging a Connector Pull Request

octavia-squidington-iii commented Aug 9, 2023

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

octavia-squidington-iii commented Aug 9, 2023

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

octavia-squidington-iii commented Aug 10, 2023

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

octavia-squidington-iii commented Aug 10, 2023

source-mysql test report (commit a009e1015b) - ❌

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akashkulk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

octavia-squidington-iii commented Aug 16, 2023

source-mysql test report (commit c291768937) - ❌

nguyenaiden commented Aug 21, 2023

nguyenaiden commented Aug 21, 2023

github-actions bot commented Aug 21, 2023

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

github-actions bot commented Aug 21, 2023

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

pedroslopez commented Aug 21, 2023

pedroslopez commented Aug 21, 2023

github-actions bot commented Aug 21, 2023

source-mysql test report (commit d4f68b0def) - ❌

github-actions bot commented Aug 21, 2023

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

nguyenaiden commented Aug 22, 2023

octavia-approvington commented Aug 22, 2023

github-actions bot commented Aug 22, 2023

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

github-actions bot commented Aug 22, 2023

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

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`) - ❌