DBZ-7698 Add transaction order metadata #187

twthorn · 2024-03-21T21:11:36Z

Motivation

Provide transaction order metadata to establish ordering between any two operations for the same primary key, regardless of order consumed by the downstream application.

See Zulip & Jira for more info on motivation.

Implementation

The key new class introduced is OrderedTransactionContext. It is similar to TransactionContext but also adds two new fields, epoch & rank. Rank is a monotonically increasing numerical value that is used for comparing which transaction is more recent. Each database can have its own function for this, we given an example for Vitess. Epoch is a value that is incremented whenever this monotonically increasing property of rank would be violated (e.g., in Vitess, this would be shrinking the host set of the GTID, or a MySQL version upgrade). In order to prevent wrongfully incrementing the epoch, it must be stored in offsets state. We store the transaction rank alongside since they are semantically related (although not strictly necessary for transaction rank to be in offsets, similar to transaction order key).

Additional Details

This demonstrates how this transaction order metadata is possible. Note: since this is to be a self-contained PR in the vitess connector repo, there are some workarounds that we do, that would be different if we want to more completely adopt this in the main Debezium repo. These are as follows:

OrderedTransactionContext subclasses TransactionContext - we do this just to plug in nicely with Debezium Core. What we should really do is either change TransactionContext to handle ordered metadata, or define an interface and have two classes one basic and one with ordered metadata.
EpochProvider/RankProvider interfaces - these should be in debezium-core's transaction package, and could be implemented by all connectors
The struct of a data change event is missing the epoch & rank information. This is needed in order for downstream applications to automatically handle repartition events. In order to add this, we must modify this method (or create a way to modify the behavior when an ordered transaction context is presented) to include the new fields (we omit this for now since we are only making changes in the Vitess Connector repo).
Vgtid to per-shard gtid - Previously our transaction ID for metadata and other purposes was the full Vgtid (all shards & gtid sets). Now, with this change, in order to plug into the TransactionMonitor class & TransactionContext, we need to be able to deduce the rank & epoch solely from the transaction ID value. This means we cannot pass in Vgtid, because we would not know which shard to look at for the gtid set. Therefore we must pass in just the GTID for the desired shard. Note: if we change debezium-core we could avoid this (add some more functionalities for receiving shard info), but as of right now it works, and may be the preferred method (since we want debezium-core to remain generic (rather than customize to handle this gtid filtering for shard case) and the offsets we store still have vgtid which is needed to resuming a VStream).
More tests - for now there are some tests/assertions we can't do yet. Specifically for (3) once that is fixed we'd want to add assertions on the source info. Additionally, we should also have an integration test on the epoch increase case (we can manipulate the stored offsets to trigger epoch increase event like shrinking host set).

Naros · 2024-03-22T19:29:17Z

src/main/java/io/debezium/connector/vitess/transaction/RankProvider.java

+interface RankProvider {
+    BigInteger getRank(String transactionId);
+}


Just thinking aloud; what about expanding this to allow arbitrary attributes somehow? I'm just not sure that only a transaction Id would be sufficient. For example, in Informix, if you don't have concurrent transactions, the underlying LSN doesn't change. Just curious if passing a parameter object here would be more beneficial for more corner cases where a single String may not be sufficient.

Responded in Zulip overall we could use the source info struct as the single parameter for this interface, which would include all needed info to calculate rank.

HenryCaiHaiying

What is currently stored in the Kafka offset topic? The VGTID or GTID?, where do you store the epoch information in Kafka offset topic?

HenryCaiHaiying · 2024-03-23T17:33:27Z

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

+            .withWidth(Width.SHORT)
+            .withImportance(ConfigDef.Importance.HIGH)
+            .withDescription(
+                    "Whether to provided ordered metadata on transactions");


provided --> provide

HenryCaiHaiying · 2024-03-23T17:42:16Z

src/main/java/io/debezium/connector/vitess/transaction/Gtid.java

+
+    private Set<String> hosts = new HashSet();
+
+    public List<String> getSequenceValues() {


It would be good to give an example of transactionId and the expected value of version, sequenceValues

HenryCaiHaiying · 2024-03-23T17:57:14Z

src/main/java/io/debezium/connector/vitess/transaction/OrderedTransactionContext.java

+    public static OrderedTransactionContext load(Map<String, ?> offsets, EpochProvider epochProvider, RankProvider rankProvider) {
+        OrderedTransactionContext context = new OrderedTransactionContext(epochProvider, rankProvider);
+        context.transactionId = (String) offsets.get(OFFSET_TRANSACTION_ID);
+        context.previousTransactionId = (String) offsets.get(OFFSET_TRANSACTION_ID);


transactionId and previousTransactionId is using the same key in the offsets map?

The offset_transaction_id stored in offset map is for a specific shard. A VGTID in the same offset map represents multiple shards. If the offset_transaction_id in the offset map represents shard1, but the new events coming from VSTREAM is for shard2, how do you do the comparison?

Ah yes forgot to add this part, we do need mapping of shard to epoch.

jpechane · 2024-03-26T15:09:06Z

Two initial comments

Should not io.debezium.connector.vitess.SourceInfo implement io.debezium.connector.AbstractSourceInfo.sequence() in this PR?
I wonder if we need rank and epoch provider. If we are oing to have this really generic then maybe just a tuple of values should be used. And it is up to each connector to implement 1, 2, 3 or arbitrary values needed. WDYT?

twthorn · 2024-03-26T15:54:25Z

Should not io.debezium.connector.vitess.SourceInfo implement io.debezium.connector.AbstractSourceInfo.sequence() in this PR?

Transaction rank could be added here. Epoch couldn't because it needs state. However, I don't think we want to provide transaction rank for all events going forward (regardless of ordered transaction metadata setting). It isn't useful without epoch data.

For connectors that do have this, e.g., postgres, it is metdata that is tied to each event, and not calcualted/derived, e.g., lsn.

I wonder if we need rank and epoch provider. If we are oing to have this really generic then maybe just a tuple of values should be used. And it is up to each connector to implement 1, 2, 3 or arbitrary values needed. WDYT?

I agree with this, we want as general a solution as possible. I am working on implementation in debezium core that will allow for any values to be augmented to the transaction context. It could be tuple/array/etc., entirely up to the connector that implements it. The core code won't require implementing connectors to use rank/epoch.

twthorn force-pushed the DBZ-7698 branch from 25d1409 to d825d41 Compare March 21, 2024 21:23

twthorn marked this pull request as draft March 21, 2024 21:27

twthorn force-pushed the DBZ-7698 branch from d825d41 to 8b39a0a Compare March 21, 2024 23:57

DBZ-7698 Add transaction order metadata

71e4c64

twthorn force-pushed the DBZ-7698 branch from 8b39a0a to 71e4c64 Compare March 22, 2024 01:03

Naros reviewed Mar 22, 2024

View reviewed changes

HenryCaiHaiying reviewed Mar 23, 2024

View reviewed changes

twthorn closed this Mar 28, 2024

twthorn deleted the DBZ-7698 branch March 28, 2024 20:46

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

DBZ-7698 Add transaction order metadata #187

DBZ-7698 Add transaction order metadata #187

twthorn commented Mar 21, 2024 •

edited

Loading

Naros Mar 22, 2024

twthorn Mar 22, 2024 •

edited

Loading

HenryCaiHaiying left a comment

HenryCaiHaiying Mar 23, 2024

HenryCaiHaiying Mar 23, 2024

HenryCaiHaiying Mar 23, 2024

HenryCaiHaiying Mar 23, 2024

twthorn Mar 25, 2024

jpechane commented Mar 26, 2024

twthorn commented Mar 26, 2024 •

edited

Loading


		private Set<String> hosts = new HashSet();

		public List<String> getSequenceValues() {

DBZ-7698 Add transaction order metadata #187

DBZ-7698 Add transaction order metadata #187

Conversation

twthorn commented Mar 21, 2024 • edited Loading

Motivation

Implementation

Additional Details

Naros Mar 22, 2024

Choose a reason for hiding this comment

twthorn Mar 22, 2024 • edited Loading

Choose a reason for hiding this comment

HenryCaiHaiying left a comment

Choose a reason for hiding this comment

HenryCaiHaiying Mar 23, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Mar 23, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Mar 23, 2024

Choose a reason for hiding this comment

HenryCaiHaiying Mar 23, 2024

Choose a reason for hiding this comment

twthorn Mar 25, 2024

Choose a reason for hiding this comment

jpechane commented Mar 26, 2024

twthorn commented Mar 26, 2024 • edited Loading

twthorn commented Mar 21, 2024 •

edited

Loading

twthorn Mar 22, 2024 •

edited

Loading

twthorn commented Mar 26, 2024 •

edited

Loading