Authors: Henry Cai, Thomas Thornton

Status

Current state: "Under Discussion"

...

Once the consumer is aware of the location of the remote log segment file, it can initiate the read from remote tiered storage and parses the content in log segment file and converts the message into ConsumerRecord objects;  We will address them in the follow-up KIP since this work is dependent on the programming language choice on the consumer side.

Handle Transaction Messages

We plan to support transaction messages in the phase II of this project so we didn’t include transactional message support structure as mandatory fields in this KIP.  Nonetheless here is the idea on how to support transaction messages:

If the consumer is reading with ReadUncommitted isolation level, no special filter is needed since it should be able to read all messages from server log.

If the consumer is reading with ReadCommitted isolation level, for messages involved in transactions, two special logic are needed on the consumer side to process based on transaction boundary:

1. LSO (Last Stable Offset) marks the end of the last committed transaction.  The client should not see messages beyond LSO.  When a leader broker generates FetchResponse normally it will make sure it will not send the messages beyond LSO.  When the consumer is reading directly from server logs from remote storage, it needs to stop at the LSO boundary. Fortunately LSO field is present in FetchResponseData.PartitionData so the consumer can use this information to stop the message consumption;
2. Aborted Transactions contains the message offsets for all the messages in an aborted transaction, the consumer client will need to filter out those messages before it can be seen by the user;  There are two possible implementation ideas (and we favor option b: broker retrieve the aborted transaction content):
1. AbortedTransactions are stored in TransactionIndex file and working together with ProducerSnapshot index file, both indexes are stored on broker side as well as on remote tiered storage side.  The leader broker can return the location of those two index files back to the consumer client and consumer will download both of them from remote storage to local and then do the filtering accordingly;
2. Broker can retrieve the aborted transaction from remote tiered storage and then set it in FetchResponse.PartitionData structure.  Although this is an extra downloading on broker side but since this is a small index file normally (since transaction abort should be a rare case), downloading that index should not cause too much performance issue on the broker side and this can greatly reduce the complexity of consumer code;

Handle Queue

We are not sure whether queue (shared group) is already supported on tiered storage, but if we need to support it here are some thoughts:

The exchange of messages is now between ShareConsumer and broker through ShareFetchRequest/ShareFetchResponse.  We can replicate the design from Consumer/Broker through FetchRequest/FetchResponse into the ShareFetch message flow.

However, the broker is also acting as share partition leader and needs to manage and distribute the inflight messages (the messages between SPSO and SPEO) to share consumers.  If we need to maintain this logic, we would need to nominate one of the consumers to act as share partition leader.

Consumer read a subset of messages

Sometimes the consumer might only want to read a subset of messages within an offset range.  Kafka log segment file can be very big (on the order of GB), having the consumer read the whole GB long segment and then find the offset range might not be performant.  We would need the OffsetIndex file to help us to quickly narrow down the search.  There are also two options to expose the offset index to the consumer with the second option as the preference:

1. When the leader broker respond FetchResponse back to the consumer client, it can also add the location of OffsetIndex file (which is stored on remote tiered storage) and have client download the index file and do the necessary index search;
2. The broker will download and read the OffsetIndex file and sets the corresponding begin/end byte position in the file back to the consumer client;  This way the client only needs to download the data log segment within a small range marked by begin/end file position;  Since the OffsetIndex file is small comparing to the log segment file and this approach can simplify the consumer side logic, we prefer this option;

Version compatibility check with broker

Over the time he broker might evolve the storage format of the log files which makes the earlier version of the client not able to read the log file anymore.  To protect this kind of version incompatibility issue, we can introduce another field in FetchRequest: SupportedStorageFormatVersions, the consumer will fill in this field with the server side storage format versions it understands (we can mark the current storage format as ApachKafkaV1), the server can fallback to the old behavior if it thinks the client version is too low.  And we mark this field as list type to contain a list of versions the client supports, so in the case when the client is newer while the server is older (forward-compatibility case), the server can still respond with the new behavior if server’s old version is still in the list of versions client understands.

This new field in FetchRequest can also handle the case that some Kafka-API-compabile vendors (e.g. Apache Pulsar, WarpStreams, AutoMQ) use their own proprietary storage format.

Public Interfaces

Protocol Changes

...