Exactly-once delivery guarantee misunderstanding

[Fantom0x00F]

Hi!

I have not yet had practical experience using the connector and am currently in the process of studying its working principles. During this study, I encountered a theoretical question regarding the guarantee of delivery semantics. I don't fully understand how exactly-once delivery is achieved given that the connector relies on relatively weak consistency models.

For simplicity, let's consider the following two scenarios.

Scenario 1:

In my understanding, read sequential consistency is insufficient to guarantee exactly-once delivery since sequential consistency allows reading stale records. For instance, the connector might read the current value, write a BEFORE record, insert the data, write an AFTER record, and then crash without committing the offset. After restarting the task, the sequential consistency read model allows the new connector to read the old AFTER value instead of the latest one, which could lead to data duplication.

Scenario 2:

The connector relies on a single-object consistency model to ensure delivery guarantees. However, there can be scenarios where a multi-object consistency model might be needed. For example, the connector might read data from Kafka, read an AFTER record, and then hang for a while due to various reasons. A rebalance might occur, and a new instance of the connector might start, but the old instance wouldn't be aware of this until it finishes its processing logic. As a result, there could be two instances of the connector running simultaneously, leading to conflicting writes.

Could you please advise what details or information I might be overlooking and how KafkaConnector would behave in the described situations?

Thank you in advance.

[Paultagoras]

Hi @Fantom0x00F - let me think on this for a bit and get back to you. It's an interesting question 🙂

[mzitnik]

Regarding the first scenario, You are right. I will also try to explain differently.
If the commit fails after we set the state to AFTER -> Kafka will pull the same or larger batch of records again since we know that we managed to write into ClickHouse some range of the data. We are dropping the data that belong to the offset previous range and inserting the rest if there is (after the drop)

In the second scenario, you are describing a situation in which we have two workers assigned to the same partition that is writing into ClickHouse.

During rebalancing, Kafka ensures that each partition is consumed by only one consumer within a consumer group at any given time, and it tries to balance the partition assignments as evenly as possible among the active consumers.

See: KIP-429 there is a new one KIP-848

[Fantom0x00F]

Hello @mzitnik

Thank you for your answer. I think you may have misunderstood the first scenario, so let me clarify some details and temporarily unmark your comment as answer.

You mentioned:

We are dropping the data that belong to the offset previous range and inserting the rest if there is (after the drop).

For the connector to drop data, Consumer 2 would need to read the last successful record. However, the documentation for ClickHouse Keeper explicitly describes the following:

By default, ClickHouse Keeper provides the same guarantees as ZooKeeper: linearizable writes and non-linearizable reads.

Non-linearizable reads imply that Consumer 2 might read an outdated record. This is allowed in ClickHouse Keeper, and there is a separate parameter quorum_reads which guarantees linearizable reads, but it is disabled by default.

Of course, this situation is likely to have a very low probability in practice, but it is theoretically possible.

Thus, my question regarding Scenario 1 can be rephrased as follows: which part of the ClickHouse KafkaConnect connector's algorithm guarantees the linearizability of reading from the KeeperMap table and prevents reading stale records?

Regarding the Scenario 2, I will write a separate comment after we conclude the discussion on Scenario 1 to avoid any confusion.

[Paultagoras]

So to clarify based on some internal discussions, Keeper is guaranteed to read all writes that were committed (before the new session in question is created) - so long as you received back a confirmation. And we do wait for confirmation, before submitting the batch to ClickHouse - so even if the crash happens, and we can't confirm the "AFTER" message was received, we still have the "BEFORE" record we need to splice the batch.

[Fantom0x00F]

This is very interesting!

I can see why this guarantee holds for the connector if there are no restarts: the sequential consistency model (non-linearizable) includes the read-your-writes guarantee, meaning that if a read happens in the same session as the write, the data will be up-to-date. In other words, Consumer 1 in the example is guaranteed to see its own latest AFTER record.

However, by definition, the sequential consistency model does not provide this guarantee across different sessions. That's why I mentioned the scenario with a restart and the appearance of Consumer 2.

Can you clarify whether the statement Keeper is guaranteed to read all writes that were committed is ensured by Keeper itself or by the KeeperMap engine? If it’s Keeper, this seems to conflict with the documentation I referred to earlier. Why do we have the quorum_reads setting if, according to you, reads are linearizable by default?

[Paultagoras]

@Fantom0x00F So it's more that session creation is a write request, so because of linearizable writes the creation cannot be committed until Keeper (the instance you're writing to) gets all previous writes.

In the 1st scenario you mentioned, that would be the end of writes from Session 1. Otherwise, future writes wouldn't necessarily be reflected.