Status

Current state: Under Discussion

...

Authors: Henry Cai, Thomas Thornton

Motivation

A common queue scheduling feature is delayed messages where the message is not supposed to be delivered or consumed right away.  This is usually caused by a large influx of messages or other activities happening in the system at the moment, the message producer wants to make sure the messages are being consumed/processed in a little bit later time or the message consumption is being spread over a period of time.  Another common use is message retry handling (e.g. retries in the new Kafka Queue feature), when a message consumer/worker cannot process the message due to some transient failures in external systems, usually the worker wants to unacknowledge the message and retry it later.  The retry is usually more ideal to be scheduled at a later time usually with some exponential backoff time interval.

...

We are proposing to implement delayed messages for Kafka to fill in this feature gap. Similar to SQS, the proposal is also focusing on messages delayed delivery for a short time window (up to 15 minutes late).  In our use cases, most of the message late delivery is within 1 minute.

Public Interfaces

Delayed Message

We are proposing add a new top level field: deliveredAfter in ProducerRecord

...

When Kafka consumer issues a fetch request to the broker and broker will only return the records when the message's deliverAfter constraint is satisfied.  

Delayed Topic

In order not to affect the regular message processing and similar to SQS design, all the delayed messages need to be enqueued at a specialized topic: DelayedTopic.

...

The default value of the parameter is false.  Setting the parameter to true will turn the topic into a delayed topic.

Proposed Changes

For the regular kafka topics, the messages are enqueued and dequeued in a FIFO fashion.  When the broker serves the consumer's FetchRequest, the response records are returned/sorted based on the message's offset.  For delayed messages, the messages are supposed to be consumed/sorted based on the message's deliverAfter field. Internally the broker implementation will need to provide a deliverAfter timeIndex to order the messages based on deliverAfter field. For the messages having the same deliverAfter, the order is based on the message's original offset.

Consumer Fetching Behavior

Currently Kafka consumer usually bookkeep and store the offset it last fetched from the broker and when it needs to fetch from the broker again, the consumer will set the fetchOffset in the FetchRequest.PartitionData and the broker will return the records at and after that offset with offset increasing.

With delayed topic, Kafka consumer will bookkeep and store the fetchOffset of the last message from the last FetchResponse stream.  The consumer will set this fetchOffset in the next FetchRequest.PartitionData.  The broker will use the fetchOffset to find the corresponding message it last sent and retrieve the deliverAfter from the message.  Using the pair of deliverAfter/fetchOffset, the broker processes the FetchRequest to find all records whose deliverAfter is at or  after deliverAfter in FetchRequest.PartitionData.  For the messages having the same deliverAfter, the return records will be sorted based on the message's offset. The FetchResposne stream will terminate when the record's deliverAfter is later than the current system clock.

Broker Implementation

In Kafka broker for the regular topic we will append the records into the current log segment file based on the enqueue order and close the log segment file either on time or size trigger.  We will maintain this behavior for the delayed topic enqueue and add a new DeliverAfterIndex to help us to get messages dequeued based on deliverAfter/offset order.

...

In the current UnifiedLog.LogSegments, we kept a ConcurrentNavigableMap<Long, LogSegment> segments data structure to quickly find the log segments satisfying a requesting offset from FetchRequest. We will add a similar lookup data structure ConcurrentNavigableMap<DeliverAfterIndexKey, LogSegment> deliverAfterSegments which returns the segments at or after the requesting deliverAfter/offset pair.  From there LogSegment.slice() method will be refactored to return a batch iterator of records.  Note in the above DeliverAfterIndex design, we are storing the physical location of the message/message-batch in the log segment file so the new BatchIterator can quickly retrieve the underlying records from the log segment file.

Merging streams from multiple log segments

For the regular topics, the messages are naturally inserted/sorted on the message offset in one log segment file and all the messages in the later log segment file are going to be after the messages in the current log segment file.  For delayed messages, with the help from the DeliverAfterIndex file, we can get the messages in one log segment file ordered based on deliverAfter and message offset.  However the deliverAfter for the messages in the next log segment are not necessarily after the deliverAfter for the messages in the previous log segment file since the order of the log segment files is based on enqueue order.  For this reason, when the broker returns the stream of records to the consumer, we will need to do a merge of multiple record streams from multiple log segment files, the merging order will be based on deliverAfter/messageOffset.

Batching behavior

Although the new field deliverAfter is specified on the record/message level, Kafka systems mostly work with the messages in batches for performance reasons.  We will also propose setting a deliverAfter field in RecordBatch level.  The messages in the same batch will satisfy the deliverAfter constraint on the batch level.  For example, for message 1 with deliverAfter set to 10:42 and message 2 with deliverAfter set to 10:44, the batch level deliverAfter will be set to 10:44 so both messages in the batch will be satisfying the constraint.

Performance

The enqueue of delayed topics will be a bit slower than the regular topic since we need to build the extra DeliverAfterIndex file.

...

We are also separating out delayed topic away from the regular topic to not affect regular stream performance.

Tiered Storage

Delayed topics can have remote.storage.enable set to true to have them uploaded to tiered storage.  Tiered storage is already uploading multiple index files together with the log segment file, we will also upload the DeliverAfterIndex files.

...

For a catchup read consumer who needs to read from very old logs from tiered storage, we would need to download multiple adjacent log segments together with their deliverAfterIndex files to establish the dequeue order between the messages in those log segment files.

Queue

The delayed messages can naturally support the retry handling in Queues.  When the consumer worker failed to process the message and unack the message, the broker can set a deliverAfter field on the message so the retry of the message delivery can happen sometime later.  When the broker gets the number of messages from Replica class to distribute them to the consumers, it needs to filter out the messages which are not supposed to be delivered at the current time clock.  The refactoring we are going to do in LocalLog.read->LogSegment.read -> LogSegment.slice is going to handle that transparently.

Compatibility, Deprecation, and Migration Plan

This feature does not support:

...

• Standard topics: Messages consumed in offset order per-partition (n before n+1)
• Delayed topics: Messages consumed in deliverAfter timestamp order, with offset as tie breaker

Backward Compatibility

• Existing topics are unaffected (delayed.delivery.enable defaults to false)
• Brokers handle both delayed and standard topics simultaneously
• Old clients cannot consume from delayed topics (FetchRequest version check)
• __consumer_offset state remains the same only encoding offset. The broker retrieves the message for the offset and the corresponding deliverAfter.

Migration Strategy

No migration needed. This is a new topic type, not a modification of existing topics.

...

1. Upgrade brokers to support new FetchRequest version
2. Upgrade consumers needing deliverAfter support to support new FetchRequest version
3. Create or update topics with delayed.delivery.enable=true

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

• Kafka Streams with Punctuators: Requires custom application code and additional infrastructure. Users want native enqueue/dequeue API.
• External Systems (SQS, Redis, RabbitMQ): Forces users outside of Kafka ecosystem, goal is unified platform
• Application-level filtering: Each consumer must implement delay logic, messages still consume bandwidth and storage during waiting state