...
Code Block |
---|
case class MessageAndMetadata[T](message: T, topic: String = "", offset: Long = -1L) |
The {{ Wiki Markup KafkaStream
\[T
\]
}}'s iterator is a {{ConsumerIterator
\[T
\]
}} which is an iterator over {{MessageAndMetadata
\[T
\]
}} objects.
Can we eliminate the need for two methods in the API? Also, providing a topic-count-map in the createMessageStreams
API is burdensome. Can we get rid of that?
...
Code Block |
---|
def createMessageStreams[T](topic: String, numStreams: Int, decoder: Decoder[T] = new DefaultDecoder) : Seq[KafkaStream[T]] |
The above also reveals an advantage of creating multiple streams for multiple topics at one-shot (by providing the topic-count-map). If you want to create multiple streams for multiple topics, then you would need to make a createMessageStreams
call for each topic, which would trigger one rebalance for each topic. With the one-shot call, only one rebalance (for all topics) will be required.
...
Advantages
- Simpler, more intuitive API which is consistent with the
createMessageStreamsByFilter
API as well. - It may be possible to combine the
createMessageStreams
andcreateMessageStreamsByFilter
calls into one API. This would need to be fleshed out in some detail, but we could have a higher-levelTopic
class that can either be a static topic, or aTopicFilter
. Another advantage of this is that the high-levelTopic
class it could do things like validate topic names. However, the consumer code would need to explicitly call (new Topic(new Whitelist("white.*")
) or (new Topic("topicname")
) but that does not seem so bad.
Disadvantages
- The above also reveals an advantage of creating multiple streams for multiple topics at one-shot . If you want to create multiple streams for multiple topics, then you would need to make a
createMessageStreams
call for each topic, which would trigger one rebalance for each topic. With the one-shot call (which receives atopic-count-map), only one rebalance (for all topics) will be required. - Not a disadvantage, but additional work: the consumer connector code is currently broken with respect to supporting multiple calls to
createMessageStreams
on the same connector object. For example, theconsumerIdString
is per connector object, and not per call. There are a bunch of other global variables that may need to become per-call instances. Anyway, the point is that we would need to completely fix that if we want to deprecate the option to provide a topic-count-map.
Impact to clients
Client code will need to change, since the current pattern of:
...
Code Block |
---|
/** * Create a list of MessageStreams for each topic. * * @param topicCountMap a map of (topic, #streams) pair * @return a map of (topic, list of KafkaMessageStream) pair. The number of items in the * list is #streams. Each KafkaMessageStream supports an iterator of messages. */ def createMessageStreams[T](topicCountMap: Map[String,Int], decoder: Decoder[T] = new DefaultDecoder) : Map[String,List[KafkaMessageStream[T]]] |