Method in high level consumer to get all message streams

[sheelc]

The messages function in consumer/zk.clj seems to return a single stream of messages but also takes in an optional number of threads. Since the number of threads is passed to Kafka's .createMessageStreams, there are extra streams that are being dropped by messages (as the code always takes the first stream).

Would it be appropriate to add a method like message-streams to the high level consumer that can return all the message streams? Additionally, I'm not sure why it's useful to pass in the number of threads to messages if there's not a way to get the other message streams.

I'd be happy to try to come up with a PR adding message-streams if I'm on the right track, but please do let me know if I'm missing something!

[pingles]

You are correct. Returning a single stream (on a single thread) was sufficient for us- we don't have any topics with more than one partition.

I think introducing message-streams would be useful though- there've been a few other recent pull-requests/changes that suggest the current API isn't always sufficient.

I've been meaning to spend some more time going through the other conversations to try and take it on ready for a 0.3.0 release with an improved client API.

In short, I think its a good thing to do and would gladly appreciate the help :)

[sheelc]

Got it, thanks for that additional context!

I'll try adding message-streams and then just have messages take the first stream from the new message-streams (while preserving its current method signature).

[pingles]

To be honest, it's probably worth making messages work more generally- i.e it'll let you turn the messages in a given stream into a lazy sequence. Sounds like it could work :)

On Thu, Feb 12, 2015 at 4:19 PM, Sheel Choksi notifications@github.com wrote:

Got it, thanks for that additional context!

I'll try adding message-streams and then just have messages take the first stream from the new message-streams (while preserving its current method signature).

— Reply to this email directly or view it on GitHub https://github.com/pingles/clj-kafka/issues/41#issuecomment-74099873.

[sheelc]

Ah, that sounds like a good idea to have messages take a stream and output a lazy sequence. Makes it easy for people to directly call createMessageStreams if they need multi-topic/filtering and then having a way to transform the stream.

I was originally thinking of making that another helper function to avoid the breaking change, but since this seems to be aimed at 0.3.0, it's probably less of an issue.

[pingles]

Yep, definitely- I think its better this is seen as a completely different api style.

On Thu, Feb 12, 2015 at 5:13 PM, Sheel Choksi notifications@github.com wrote:

Ah, that sounds like a good idea to have messages take a stream and output a lazy sequence. Makes it easy for people to directly call createMessageStreams if they need multi-topic/filtering and then having a way to transform the stream.

I was originally thinking of making that another helper function to avoid the breaking change, but since this seems to be aimed at 0.3.0, it's probably less of an issue.

— Reply to this email directly or view it on GitHub https://github.com/pingles/clj-kafka/issues/41#issuecomment-74110812.

[pingles]

Hi,

I've made some changes that now offer this- it's been released is on Clojars as 0.3.0-0.8.1.1. I'll close this PR.

Thanks, Paul

[pingles]

d'oh... getting confused between pull requests and issues :)

[sheelc]

Awesome, the changes seem like they'll work for us. Looking forward to trying them out, thanks Paul!