Does the application transit to RUNNING state? Also check `KafkaStreams#localThreadsMetadata()` what tasks are assigned?
You might also enable DEBUG logs for
`org.apache.kafka.clients.consumer.**` classes to see if the consumer
sends fetch request to the broker.
-Matthias
On 6/11/19 7:03 PM, Brian Putt wrote:
> The application just hangs (we let it sit for ~1 hour, small dataset as
> we're testing), we can restart it listening to 1 of the 3 topics we start
> it with and it chugs along, no problem. The same code is executed as
> separate application.ids listening to other topics without any issues.
> We'll try to increase our logging as nothing is currently being shown in
> the logs. Guessing we have our level set to WARN.
>
> Will certainly share updates as we figure it out.
>
> Thanks!
>
> On Tue, Jun 11, 2019 at 6:07 PM Matthias J. Sax <matth...@confluent.io>
> wrote:
>
>> What do you exactly observe?
>>
>> - Does the application rebalance correctly?
>> - Does it start processing?
>> - Anything in the logs about the status of the application?
>>
>>
>> -Matthias
>>
>> On 6/10/19 6:19 PM, Brian Putt wrote:
>>> Hello,
>>>
>>> I'm working with the kafka streams api and am running into issues where I
>>> subscribe to multiple topics and the consumer just hangs. It has a unique
>>> application.id and I can see in kafka that the consumer group has been
>>> created, but when I describe the group, I'll get: consumer group X has no
>>> active members
>>>
>>> The interesting thing is that this works when the topics list only
>> contains
>>> 1 topic. I'm not interested in other answers where we create multiple
>>> sources, ie: source1 = builder.stream("topic1") and source2 =
>>> builder.stream("topic2") as the interface for StreamsBuilder.stream
>> supports
>>> an array of topics.
>>>
>>> I've been able to subscribe to multiple topics before, I just can't
>>> replicate how we've done this. (This code is running in a different
>>> environment and working as expected, so not sure if it's a timing issue
>> or
>>> something else)
>>>
>>> List<String> topics = Arrays.asList("topic1", "topic2");
>>>
>>> StreamsBuilder builder = new StreamsBuilder();
>>> KStream<String, String> source = builder.stream(topics);
>>>
>>> source
>>> .transformValues(...)
>>> .map(key, value) -> ...)
>>> .to((key, value, record) -> ...);
>>>
>>> new KafkaStreams(builder.build(), props).start();
>>>
>>> This question has been posted on stackoverflow in case you want to
>>> answer there:
>> https://stackoverflow.com/questions/56535113/kafka-streams-listening-to-multiple-topics-hangs
>>>
>>
>>
>
signature.asc
Description: OpenPGP digital signature
