Data unavailable in second server after successful insert with Ignite 3

[malte-f19]

We have an issue using Ignite 3 from multiple services. Often data is not available although it should be.

TL;DR

• A fresh cluster with three nodes running
• A table with a combined primary key and some data column
• Two applications: One (the sender) is writing to Ignite, the second one (the receiver) is reading from Ignite.
• After the sender stored multiple rows with the same id but with different sequence numbers, it sends a REST request with the id to the receiver.
• The receiver then tries to select data by the id and doesn't get any records back.

Setup

This is the SQL we've run in Ignite:

CREATE ZONE my_zone (PARTITIONS 25, REPLICAS 1, AUTO SCALE UP 10, AUTO SCALE DOWN 600) STORAGE PROFILES ['memStorageProfile'];

CREATE TABLE IF NOT EXISTS data_table(
    id                VARCHAR(50) NOT NULL,
    sequence          INT NOT NULL,
    data              VARCHAR(99999) NOT NULL,
    PRIMARY KEY (id, sequence)
    ) ZONE my_zone STORAGE PROFILE 'memStorageProfile';

The corresponding entity class is defined like this:

@Table("data_table")
public class DataTable {
    @Id
    @Column(value = "id", length = 50, nullable = false)
    private String id;

    @Id
    @Column(value = "sequence", nullable = false)
    private int sequence;

    @Column(value = "data", nullable = false)
    private String data;

    // Constructors, getters and setters omitted.
}

The sender uses the following code to store data into Ignite:

try (IgniteClient igniteClient = IgniteClient
        .builder()
        .addresses("127.0.0.1:10800", "127.0.0.1:10801", "127.0.0.1:10802")
        .build()) {
    Table dataTable = igniteClient.tables().table("data_table");
    RecordView<DataTable> recordView = dataTable.recordView(DataTable.class);
    String id = UUID.randomUUID().toString();
    List<DataTable> data = new ArrayList<>();
    for (int i = 0; i < 5; i++) {
        data.add(new DataTable(id, i, "data-" + i));
    }
    recordView.upsertAll(null, data);

    // Send rest call to receiver
}

The receiver is using the following code to get the data:

IgniteClient igniteClient = IgniteClient
        .builder()
        .addresses("127.0.0.1:10800", "127.0.0.1:10801", "127.0.0.1:10802")
        .build();
Table dataTable = igniteClient.tables().table("data_table");
RecordView<DataTable> recordView = dataTable.recordView(DataTable.class);

// Wait to receive the REST call with the id...

Cursor<DataTable> cursor = recordView.query(null, columnValue("id", equalTo(id)))) {
while (cursor.hasNext()) {
    // Do something
}

Actual results

When receiving the REST call (which is sent after the upsertAll call in our sender returned) we don't receive any data when running query, ie the cursor is empty.

I added some retry logic to figure out if data will be available after some time. Here is an example output of that:

Data not yet available, waiting 100 ms...
Data not yet available, waiting 200 ms...
Data retrieved successfully after 2 cycles and 300 ms

So, it took between 100 and 300ms in this example after the REST request was received on the receiver side until the data was actually readable. Note it sometimes takes longer.

Expected results

After the upsertAll returned, other services can read the data.

Things we tried

We tried different things to get around this, with no success:

• Wrap the upsertAll in a transaction.
• Use a read-only transaction when reading the data.
• Immediately after writing the data, we read them in the sender using the exact same code as the receiver and successfully get the newly inserted data back.
• Instead of a cluster with three nodes, we created a cluster with a single node.
• Replace aimem with aipersist.

Why is that happening and how can we get around this? Any hint is much appreciated. I can also provide the minimal example in case that is helpful. Thanks in advance!

[alievmirza]

Hello! Thank you @malte-f19 for the detailed description!

I would be glad to see server logs. The full reproducer would be also very helpful.

[alievmirza]

You can send all related info here or create ticket in https://issues.apache.org/jira/projects/IGNITE/summary with ignite-3 label

[ptupitsyn]

@alievmirza JIRA is for contributors only, it is is not open for registrations anymore. Let's continue the discussion here and then we (devs) file a ticket if necessary.

[ptupitsyn]

@malte-f19 This behavior is expected.

Separate IgniteClient instances are not guaranteed to see the same data at the same time.

Each client has its own "observable timestamp", a hybrid clock value that it "looks at". This guarantees correct causality within a single client instance and speeds up reads because we can skip waiting for certain things to happen, but leads to the delays you observe.

I can suggest workarounds if you describe the use case in more detail.

[slukyano]

I'd say this is roughly the same as any other inter-process or inter-thread synchronization. If you have two workers (threads, processes, clients, etc.) that both read the same concurrency-enabled data structure, you must establish a happens-before relationship between the workers for worker2 to be guaranteed to see the write made by worker1.

Take two Java threads t1 and t2 and a volatile field f. If t1 writes to f, then sends "some" signal to t2, and t2 reads f, t2 is not guaranteed to see the write. For that to be guaranteed, the signal from t1 to t2 needs to create a happens-before relationship in terms of Java concurrency model.

Similarly, with Ignite. When two clients access the same table and signal each other, the signal needs to create a happens-before relationship in Ignite's concurrency model.

Ignite doesn't provide a straightforward API to implement this exact use case now. It could, and it isn't even hard, but right now the use case feels to me like it might have a better way to solve it.

[ptupitsyn]

A workaround is to use a normal (non-readonly) transaction to read the data. It will make things slower and might introduce contention, though.

[malte-f19]

Thank you all for your input, much appreciated!

The workaround you suggested, @ptupitsyn, is something we tried as well. But that introduced new issues when reading the data. We got exceptions like this:

org.apache.ignite.sql.SqlException: Failed to acquire the intention table lock due to a conflict [locker=019ba41c-adae-0000-a5b0-32ae00000001, holder=019ba41c-ada0-0000-a5b0-32ad00000001, abandoned=false]

or

org.apache.ignite.sql.SqlException: Failed to acquire a lock due to a possible deadlock [locker=019bb69e-a1c1-0000-a5b0-32ae00000001, holder=019bb69e-a150-0000-a5b0-32ad00000001]

Let me try to describe our use-case in more detail. We have several services running in a Kubernetes cluster, executing calculation jobs. There's one service responsible for orchestrating the job, generating unique job-ids, triggering the other services as needed and taking care of persistent data storage. Other services either collect data from external sources or run calculations on the data itself. The data is collected into a context we used to transfer between the services using JSON via REST calls. Since the contexts can get huge and not every service needs all data to work, we wanted to move the data into some kind of cache. Ignite seems to be the perfect system for this use-case. We divided the context into several pieces by type (like input data, calculated data, etc.) and even smaller chunks using additional data points. So we came up with several tables using the job-id and one or two other columns as a primary key.

A typical workflow is as follows:

• A job gets triggered, so the orchestrator-service generates a job-id and sends this it to the first data collection service.
• The collection service gets necessary data and stores it into the appropriate Ignite tables using the job-id.
• After the storage is done, it sends a REST request to the orchestrator containing the job-id.
• The orchestrator-service will read the data and persist it as needed and call the next service (again using REST with the job-id).
• These steps are repeated until the calculation is done.

Since we really want to use Ignite as an in-memory cache, and we don't care about persisting the data, we're using aimem as a storage engine.

I hope this makes it clear what we want to achieve. If not or you need additional information, feel free to ask. I will provide what I can.

[ptupitsyn]

For what it's worth, below is a workaround using the internal API to pass around the "causality token" or "observable timestamp":

// Sender service gets timestamp after doing updates
TcpIgniteClient senderTcpClient = (TcpIgniteClient) igniteClient;
long ts = senderTcpClient.channel().observableTimestamp().get().longValue(); // Include ts in the REST call
        
// Receiver service gets sender timestamp from REST call and applies:
long senderTs = ...;
TcpIgniteClient receiverTcpClient = (TcpIgniteClient) igniteClient;
receiverTcpClient.channel().observableTimestamp().update(ts);

// Now read the data, transaction is not necessary

(add a dependency to org.apache.ignite:ignite-core to make it work)

@malte-f19 thank you for raising this issue, it triggered an important discussion with the team, hopefully we will make this better in the next release.

[malte-f19]

Thank you for this workaround @ptupitsyn. Maybe it's worth making the timestamp available in the client API? At least I was searching for something like that when I first stumbled over the issue.
I've integrated that into our application, and it seems to work, but further tests are needed. I just ran a few simple tests and nothing under load.
Since you mentioned this is an internal API. Is that a stable API or is it likely to change with the next version?

[ptupitsyn]

• This API is internal but not likely to change. It is a fundamental part of the client.
• We are working on a public API for that