Status

Current state: Under Vote

Discussion thread:

https://lists.apache.org/list.html?dev@kafka.apache.org:lte=1M:kip-349

Must use Advanced search options to go back in history. 

JIRA: KAFKA-6690

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

To support use-cases where there’s a need to prioritize source topics. If a consumer is subscribed to more than one topic, i.e., a high priority topic High and a low priority topic Low.  The consumer should consume events from topic Low only when all the events from topic High are consumed.

From the reporter of KAFKA-6690:

We use Kafka to process the asynchronous events of our Document Management System such as preview generation, indexing for search etc.
The traffic gets generated via Web and Desktop Sync application. In such cases, we had to prioritize the traffic from web and consume them first.  But this might lead to the starvation of events from sync if the consumer speed is slow and the event rate is high from web.  A solution to handle the starvation with a timeout after which the events are consumed normally for a specified period of time would be great and help us use our resources effectively.

Public Interfaces

The addition of new subscribe API that allows caller to prioritize topics.  New class TopicPriority constructor

public TopicPriority(java.lang.String topic, int priority);

where priority is a positive integer.  

This subscribe method takes a list of TopicPriority as a parameter.  

public void subscribe(java.util.List<TopicPriority> topicPriorities);

This subscribe method parameter has an additional parameter maxDelta that's used to prevent starvation of lower priority topics. 

public void subscribe(java.util.List<TopicPriority> topicPriorities, Integer maxDelta);

If the delta (in minutes) between the oldest message in the highest and lowest priority topics exceeds

maxDelta, then topic prioritization is disabled.  When the delta falls below maxDelta then prioritization then can be enabled. 

Proposed Changes

The new behavior is only in effect when the user specifies priorities for topics using the above new API.  In this case topics are checked starting from highest priority first and then in descending order based on priority.  All events must be consumed from a higher priority topic before consumption is performed on a lower priority topic. 

Note that the issue of starvation of lower priority topics has been discussed in the most recent thread.  The above API will be extended to handle starvation.  One idea is to check the delta between the oldest messages in the different topics.  The consumer would then revert to using no priorities when the delta reaches a max threshold for the delta, and then re-enable prioritization after achieving a min threshold. 

Compatibility, Deprecation, and Migration Plan

This feature preserves backward compatibility by treating topics without priorities as having the same priority (i.e. current consumer semantics does not change). 

Rejected Alternatives

None