DBZ-7698 Add ordered transaction metadata for Vitess connector #188

twthorn · 2024-03-28T20:49:25Z

Changes

Implements transaction metadata order feature using new features added in debezium/debezium#5437 for customizing transaction metadata.

Adds the transaction epoch & rank fields. Tracks epoch in state for correct epoch recording. Computes rank for each GTID received.

How to Use

We will add this section/similar one to docs once we finalize things:

When the following two configs are set to the listed values, the data message Envelope is also enriched with a new transaction field.

transaction.contex: VitessOrderedTransactionContext
transaction.struct.maker: VitessOrderedTransactionStructMaker

This field provides information about every event in the form of a composite of fields. With ordered transaction metadata enabled, there are two additional fields included:

transaction_epoch - non-decreasing value representing the "epoch" that its transaction rank belongs to

transaction_rank - non-decreasing value (within an epoch) that represents the order of the transaction

Additionally, a third field (already included in standard transaction metadata) is also relevant for ordering:

total_order absolute position of the event among all events generated by the transaction

For how these fields can be used to establish an ordering of events. Consider the following example.
There are two data change events that occurred in the same shard and for the same primary key. However, kafka experienced a re-partition event so the consumer order of the two events cannot be trusted. A downstream application can determine which event to apply (the newer event) and which to discard with the following logic:

If transaction_epoch is not equal, return the event with the grater epoch. Otherwise, continue

If transaction_rank is not equal, return the event with the greater rank. Otherwise, continue:

Return the event with a greater total_order

If we reach step 3, the two events are in the same transaction. The total_order field represents the order of events within a transaction, so a greater value will be the newer event.

Following is an example of a data change event with ordered transaction metadata:

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": 1637988245467,
  "ts_us": 1637988245467841,
  "ts_ns": 1637988245467841698,
  "transaction": {
    "id": "[{\"keyspace\":\"test_unsharded_keyspace\",\"shard\":\"0\",\"gtid\":\"MySQL56/e03ece6c-4c04-11ec-8e20-0242ac110004:1-68\"}]",
    "total_order": 1,
    "data_collection_order": 1,
    "transaction_rank": 68,
    "transaction_epoch": 0
  }
}

Additional notes

Tests will not pass in Github until changes in this PR debezium/debezium#5437 are merged. However, I ran this locally with the snapshot version built on that PR and all itests passed.

HenryCaiHaiying · 2024-04-02T19:47:58Z

I think it would be good to have an example of what's stored in offsetMap which gets persisted into Kafka offset topic. Especially on whether the transaction_id and rank field in offset map represents one shard or multiple shards.

Also it would be a good idea to have an example of the message content (for the new transaction_id/rank fields) to the Kafka data topic.

twthorn · 2024-04-02T21:46:15Z

Thanks for the review @HenryCaiHaiying I added some examples in java doc comments.

HenryCaiHaiying · 2024-04-02T19:44:56Z

...n/java/io/debezium/connector/vitess/pipeline/txmetadata/VitessOrderedTransactionContext.java

+public class VitessOrderedTransactionContext extends TransactionContext {
+    public static final String OFFSET_TRANSACTION_EPOCH = "transaction_epoch";
+    public static final String OFFSET_TRANSACTION_RANK = "transaction_rank";
+    protected String previousTransactionId = null;


Is the previousTransactionId here the vGtid which presents multiple shards?

Yes, I have updated the name, it is the full vgtid

HenryCaiHaiying · 2024-04-02T19:45:34Z

src/main/java/io/debezium/connector/vitess/pipeline/txmetadata/VitessEpochProvider.java

+    private Map<String, Long> shardToEpoch = new HashMap<>();
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    public Long getEpoch(Long previousEpoch, String previousTransactionId, String transactionId) {


Is the previousTransactionId and transactionId here represents one shard or multiple shards?

If id represents one shard, maybe the variable name can just be called previousGtidString to avoid confusion.

Also this method seems can be a static method, it doesn't use any instance variables.

Is the previousTransactionId and transactionId here represents one shard or multiple shards?
Updated variable name, it represents single shard gtid

Also this method seems can be a static method, it doesn't use any instance variables.

It uses the shardToEpoch instance variable, so it cannot be static. However, the rank provider method could be static so I made that one static.

HenryCaiHaiying · 2024-04-03T02:08:53Z

src/main/java/io/debezium/connector/vitess/pipeline/txmetadata/VitessEpochProvider.java

+    private Map<String, Long> shardToEpoch = new HashMap<>();
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    public Long getEpochForGtid(Long previousEpoch, String previousGtidString, String gtidString) {


This method doesn't use shardToEpoch instance variables, can be a static method

Good catch, I'll update that one.

HenryCaiHaiying

lgtm

jpechane

Would it be possible to override io.debezium.connector.AbstractSourceInfo.sequence() to align the connector with others?

src/main/java/io/debezium/connector/vitess/Vgtid.java

twthorn · 2024-04-11T18:55:13Z

Would it be possible to override io.debezium.connector.AbstractSourceInfo.sequence() to align the connector with others?

My thinking is that we don't want this epoch/rank tracking on by default so it should not be in the source info. More reasoning mentioned here.

jpechane · 2024-04-11T19:06:41Z

Would it be possible to override io.debezium.connector.AbstractSourceInfo.sequence() to align the connector with others?

My thinking is that we don't want this epoch/rank tracking on by default so it should not be in the source info. More reasoning mentioned here.

Yes, I understand. My reasoning is that it would be present only if transactions are enabled.

twthorn · 2024-04-11T19:36:45Z

Would it be possible to override io.debezium.connector.AbstractSourceInfo.sequence() to align the connector with others?

My thinking is that we don't want this epoch/rank tracking on by default so it should not be in the source info. More reasoning mentioned here.

Yes, I understand. My reasoning is that it would be present only if transactions are enabled.

Are we sure we want the content of SourceInfo to be dependent on that transaction metadata setting? Seems not as modular. That setting only affects the transaction block or transaction messages so it keeps the modifications isolated. I'm not sure about having SourceInfo changing based on that setting, seems like an unintuitive side effect.

For other connectors I see that use sequence, it doesn't depend on the setting. It contains data that is intrinsic to the transaction, not some derived stateful calculation.

Additionally, if epoch/rank is present in the transaction block and in the sequence value that does mean duplication.

jpechane · 2024-04-18T08:38:19Z

@twthorn I am still not fully convinced on not using sequence in source info due to break of consistency with other connectors and I'd be willing to accept dependency on transaction metadata setting as this can be properly documented. Also I'd have no issue in duplication of the values in this case. Still let's keep it as is and we'll intorduce it later if there will be real need.

jpechane

@twthorn Thanks a lot. This PR is ready to be merged after the core PR is refactored on the top of the latest main.

…ard in shard to epoch map

jpechane · 2024-04-19T08:34:20Z

Forse-pushed rebase on the latest main

twthorn mentioned this pull request Mar 28, 2024

DBZ-7698 Add custom transaction metadata debezium/debezium#5437

Merged

HenryCaiHaiying reviewed Apr 2, 2024

View reviewed changes

HenryCaiHaiying reviewed Apr 3, 2024

View reviewed changes

HenryCaiHaiying approved these changes Apr 3, 2024

View reviewed changes

jpechane reviewed Apr 11, 2024

View reviewed changes

src/main/java/io/debezium/connector/vitess/Vgtid.java Outdated Show resolved Hide resolved

twthorn requested a review from jpechane April 12, 2024 14:00

jpechane approved these changes Apr 18, 2024

View reviewed changes

twthorn added 15 commits April 19, 2024 10:31

DBZ-7698 Add ordered transaction metadata for Vitess connector

2fae5cd

DBZ-7698 Refactor to generic transaction block customization

7f6d04e

DBZ-7698 Improve test naming

68eaf6c

DBZ-7698 Fix naming, implement getTransactionInfo

9d5ebf3

DBZ-7698 fix itests by Removing default for TRANSACTION_STRUCT_MAKER

d9f7ab2

DBZ-7698 Remove import

5f6e9e2

DBZ-7698 Add more unit tests

2416e72

DBZ-7698 Switch to Decimal type, add examples

967be82

DBZ-7698 Improve naming, minor refactors

084e805

DBZ-7698 Add another test, refactor method to static

c4e0eef

DBZ-7698 Override and use load instance method

e973a49

DBZ-7698 Handle current/empty gtid for epoch provider, and missing sh…

34541c8

…ard in shard to epoch map

DBZ-7698 Add integration test for epoch change

80b45a7

DBZ-7698 Use DebeziumException, refactor things to match debezium-core

ccfc7b9

DBZ-7698 Refactor to single factory interface/class/config

947f303

jpechane force-pushed the DBZ-7698 branch from 2c548c5 to 947f303 Compare April 19, 2024 08:33

jpechane merged commit 31fbfef into debezium:main Apr 19, 2024
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

DBZ-7698 Add ordered transaction metadata for Vitess connector #188

DBZ-7698 Add ordered transaction metadata for Vitess connector #188

twthorn commented Mar 28, 2024 •

edited

Loading

HenryCaiHaiying commented Apr 2, 2024

twthorn commented Apr 2, 2024

HenryCaiHaiying Apr 2, 2024

twthorn Apr 3, 2024

HenryCaiHaiying Apr 2, 2024

HenryCaiHaiying Apr 2, 2024 •

edited

Loading

twthorn Apr 3, 2024

HenryCaiHaiying Apr 3, 2024

twthorn Apr 3, 2024

HenryCaiHaiying left a comment

jpechane left a comment

twthorn commented Apr 11, 2024

jpechane commented Apr 11, 2024

twthorn commented Apr 11, 2024

jpechane commented Apr 18, 2024

jpechane left a comment

jpechane commented Apr 19, 2024

DBZ-7698 Add ordered transaction metadata for Vitess connector #188

DBZ-7698 Add ordered transaction metadata for Vitess connector #188

Conversation

twthorn commented Mar 28, 2024 • edited Loading

Changes

How to Use

Additional notes

HenryCaiHaiying commented Apr 2, 2024

twthorn commented Apr 2, 2024

HenryCaiHaiying Apr 2, 2024

Choose a reason for hiding this comment

twthorn Apr 3, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Apr 2, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Apr 2, 2024 • edited Loading

Choose a reason for hiding this comment

twthorn Apr 3, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Apr 3, 2024

Choose a reason for hiding this comment

twthorn Apr 3, 2024

Choose a reason for hiding this comment

HenryCaiHaiying left a comment

Choose a reason for hiding this comment

jpechane left a comment

Choose a reason for hiding this comment

twthorn commented Apr 11, 2024

jpechane commented Apr 11, 2024

twthorn commented Apr 11, 2024

jpechane commented Apr 18, 2024

jpechane left a comment

Choose a reason for hiding this comment

jpechane commented Apr 19, 2024

twthorn commented Mar 28, 2024 •

edited

Loading

HenryCaiHaiying Apr 2, 2024 •

edited

Loading