Hi, I am seeing behaviour which I am not expecting when using topic filters.

TopicFilter sourceTopicFilter = new Whitelist("pageviews");
List<KafkaStream<Message>> streams consumer.createMessageStreamsByFilter(sourceTopicFilter, 3);

The topic has exactly 3 partitions and 3 streams are created, however only
the last one gets messages, the first two become stale.
I am not sure if the singe stream contains messages from all partitions or
When I however use the following code I get all 3 streams consuming for
each respective partition.

Map<String, List<KafkaStream<Message>>> topicStreams consumer.createMessageStreams(
    new HashMap<String, Integer>() {{
        put("pageviews", 3);
    }}
);
List<KafkaStream<Message>> streams = topicStreams.get("pageviews");

Am I missing something, is this expected ?

--
Michal Haris
Software Engineer

www.visualdna.com | t: +44 (0) 207 734 7033

This is likely caused by https://issues.apache.org/jira/browse/KAFKA-550.
The fix has been checked into trunk.

Thanks,

Jun

On Tue, Nov 20, 2012 at 4:44 AM, Michal Haris <[EMAIL PROTECTED]>wrote:

> Hi, I am seeing behaviour which I am not expecting when using topic
> filters.
>
> TopicFilter sourceTopicFilter = new Whitelist("pageviews");
> List<KafkaStream<Message>> streams > consumer.createMessageStreamsByFilter(sourceTopicFilter, 3);
>
> The topic has exactly 3 partitions and 3 streams are created, however only
> the last one gets messages, the first two become stale.
> I am not sure if the singe stream contains messages from all partitions or
> When I however use the following code I get all 3 streams consuming for
> each respective partition.
>
> Map<String, List<KafkaStream<Message>>> topicStreams > consumer.createMessageStreams(
>     new HashMap<String, Integer>() {{
>         put("pageviews", 3);
>     }}
> );
> List<KafkaStream<Message>> streams = topicStreams.get("pageviews");
>
> Am I missing something, is this expected ?
>
> --
> Michal Haris
> Software Engineer
>
> www.visualdna.com | t: +44 (0) 207 734 7033
>

Thanks Jun, will rebuild our maven libraries, we're running on trunk code
already :).

On 20 November 2012 17:15, Jun Rao <[EMAIL PROTECTED]> wrote:

> This is likely caused by https://issues.apache.org/jira/browse/KAFKA-550.
> The fix has been checked into trunk.
>
> Thanks,
>
> Jun
>
> On Tue, Nov 20, 2012 at 4:44 AM, Michal Haris <[EMAIL PROTECTED]
> >wrote:
>
> > Hi, I am seeing behaviour which I am not expecting when using topic
> > filters.
> >
> > TopicFilter sourceTopicFilter = new Whitelist("pageviews");
> > List<KafkaStream<Message>> streams > > consumer.createMessageStreamsByFilter(sourceTopicFilter, 3);
> >
> > The topic has exactly 3 partitions and 3 streams are created, however
> only
> > the last one gets messages, the first two become stale.
> > I am not sure if the singe stream contains messages from all partitions
> or
> > When I however use the following code I get all 3 streams consuming for
> > each respective partition.
> >
> > Map<String, List<KafkaStream<Message>>> topicStreams > > consumer.createMessageStreams(
> >     new HashMap<String, Integer>() {{
> >         put("pageviews", 3);
> >     }}
> > );
> > List<KafkaStream<Message>> streams = topicStreams.get("pageviews");
> >
> > Am I missing something, is this expected ?
> >
> > --
> > Michal Haris
> > Software Engineer
> >
> > www.visualdna.com | t: +44 (0) 207 734 7033
> >
>

--
Michal Haris
Software Engineer

www.visualdna.com | t: +44 (0) 207 734 7033