Understanding Apache Paimon Concurrency Control

Posted on In Views:

Designing and testing Apache Paimon concurrency control to reveal common conflict scenarios

Previously, we tried Apache Paimon at the playground. In the conclusion I mentioned that I would like to know what scenarios concurrency control is designed to handle and what happens when there is a conflict.

This article will design an experiment environment that actually shows what snapshot conflict and files conflict are.

First, use this more complicated playground.

If you already have a main branch, you can just checkout to the jdbc branch.

Experiment Design

We will start two Flink tasks and keep receiving data from Kafka and writing to the same Paimon table.

The two Flink tasks use different consumer groups, so the amount of data written should be equal.

Therefore, we run the following script in Flink SQL.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
CREATE TABLE orders (  
    order_number BIGINT,  
    price        DECIMAL(32,2),  
    buyer        STRING  
);  
  
CREATE TEMPORARY TABLE kafka_source1 (  
    order_number BIGINT,  
    price        DECIMAL(32,2)  
) WITH (  
    'connector' = 'kafka',  
    'topic' = 'test_topic',  
    'properties.bootstrap.servers' = 'broker:29092',  
    'properties.group.id' = 'session1',  
    'scan.startup.mode' = 'earliest-offset',  
    'format' = 'json'  
);  
INSERT INTO orders  
SELECT  
    *,  
    'session1' AS buyer  
FROM kafka_source1;  
CREATE TEMPORARY TABLE kafka_source2 (  
    order_number BIGINT,  
    price        DECIMAL(32,2)  
) WITH (  
    'connector' = 'kafka',  
    'topic' = 'test_topic',  
    'properties.bootstrap.servers' = 'broker:29092',  
    'properties.group.id' = 'session2',  
    'scan.startup.mode' = 'earliest-offset',  
    'format' = 'json'  
);  
INSERT INTO orders  
SELECT  
    *,  
    'session2' AS buyer  
FROM kafka_source2;

As you can see from the script, I used two kafka sources and wrote to the same orders table, only the buyer field is different. In the end, all we need to do is GROUP BY buyer to know if the two tasks are the same.

Here are some ingest samples.

1
2
3
4
{"order_number": 1, "price": 100}  
{"order_number": 2, "price": 200}  
...  
{"order_number": 100, "price": 10000}

Experiment Result

In the scene where file is used as a metastore, it is easy to see that Flink’s task is constantly restarting.

There are two types of error cases mentioned in the document.

The first one is snapshot conflict.

This will result in a direct data loss, which will be reflected in the following error on Flink.

1
2
3
4
5
6
java.lang.RuntimeException: Exception occurs when committing snapshot #19 (path s3://warehouse/flink/default.db/Orders/snapshot/snapshot-19) by user 2439e7bb-506d-4ff3-a204-ab66610dd75d with identifier 3265 and kind APPEND. Cannot clean up because we can't determine the success.  
	at org.apache.paimon.operation.FileStoreCommitImpl.tryCommitOnce(FileStoreCommitImpl.java:986)  
	at org.apache.paimon.operation.FileStoreCommitImpl.tryCommit(FileStoreCommitImpl.java:715)  
	at org.apache.paimon.operation.FileStoreCommitImpl.commit(FileStoreCommitImpl.java:301)  
  
the rest omitted

Another scenario is files conflict.

This will also cause the Flink task to restart with the following error.

1
2
3
4
5
6
java.lang.RuntimeException: This exception is intentionally thrown after committing the restored checkpoints. By restarting the job we hope that writers can start writing based on these new commits.  
	at org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager.recover(RestoreAndFailCommittableStateManager.java:84)  
	at org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager.initializeState(RestoreAndFailCommittableStateManager.java:77)  
	at org.apache.paimon.flink.sink.CommitterOperator.initializeState(CommitterOperator.java:153)  
      
the rest omitted

Either type of conflict will affect the real time ability to write data, or worse, data will be lost in a snapshot conflict.

Solution

Of course, Paimon also provides a solution to this problem, which is to use a metastore that can be locked, such as MySQL.

So in our experiment, we need to change the practice of CREATE CATALOG. The correct CATALOG is shown in README.md, and the key point is to add the following line.

1
'lock.enabled' = 'true'

When we re-run the experiment using MySQL Catalog, we find that both Flink tasks run stably, no longer restarting, and the data in paimon is no longer lost.

Conclusion

In this experiment, we can see that when using two writers to write to the same table, it is very easy to generate conflict even if the amount of data is not large and the write speed is not fast.

If this is the case, it is important to use a metastore that can be locked in any case.

Nevertheless, as of today Trino still does not support jdbc metastore, which will greatly affect the popularity of Paimon, after all, not everyone has Hive in this era.

Originally published on Medium