You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Stevo Slavić <ss...@gmail.com> on 2015/10/05 17:44:13 UTC
How to verify offsets topic exists?
Hello Apache Kafka community,
In my integration tests, with single 0.8.2.2 broker, for newly created
topic with single partition, after determining through topic metadata
request that partition has lead broker assigned, when I try to reset offset
for given consumer group, I first try to discover offset coordinator and
that lookup is throwing ConsumerCoordinatorNotAvailableException
On
https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
it is documented that broker returns ConsumerCoordinatorNotAvailableCode
for consumer metadata requests or offset commit requests if the offsets
topic has not yet been created.
I wonder if this is really the case, that the offsets topic has not been
created. Any tips how to ensure/verify that offsets topic exists?
Kind regards,
Stevo Slavic.
Re: How to verify offsets topic exists?
Posted by Stevo Slavić <ss...@gmail.com>.
If I'm not mistaken, replication factor of a topic does not get stored in
ZK - on creation replication factor gets translated to topic replica
assignment which gets stored in ZK.
Again, please correct me if wrong - it seems that only during topic
creation it is verified that all replicas in replica assignment are
actually live brokers. Once topic is created, topic can become under
replicated - brokers from topic replica assignment can go down, and even in
that case topic may still be usable (online) depending on other settings.
Would it make sense, as exception for consumer offsets topic only, or as
rule for all topics - to allow topic to be created with replica assignment
which is referencing brokers which do not exist (just rely on monotonically
increasing broker ids convention) or are just not alive? None of the
brokers referenced in replica assignment would have to be alive or maybe
require just one referenced broker to be alive, or maybe make it
configuration option - switch whether to enforce that brokers referenced in
replica assignment are alive, or min alive brokers in replica assignment
setting, or both.
Kind regards,
Stevo Slavic.
On Thu, Oct 8, 2015 at 3:32 PM, Stevo Slavić <ss...@gmail.com> wrote:
> There's another related bug - triggering offsets topic creation through
> requesting metadata about that topic does not work in case of single broker
> clean (no topics created yet) Kafka cluster running. In that case sequence
> returned by KafkaApis.getAliveBrokers is empty, and
> KafkaApis.getTopicMetadata will issue a AdminUtils.createTopic request
> with offsetsTopicReplicationFactor of 3, which will fail with
>
> ERROR [KafkaApi-0] error when handling request Name: TopicMetadataRequest;
> Version: 0; CorrelationId: 0; ClientId: ; Topics: __consumer_offsets
> (kafka.server.KafkaApis)
> kafka.admin.AdminOperationException: replication factor: 3 larger than
> available brokers: 1
> at kafka.admin.AdminUtils$.assignReplicasToBrokers(AdminUtils.scala:70)
> at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:171)
> at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:513)
> at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:503)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
> at scala.collection.immutable.Set$Set1.foreach(Set.scala:79)
> at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
> at
> scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)
> at scala.collection.SetLike$class.map(SetLike.scala:92)
> at scala.collection.AbstractSet.map(Set.scala:47)
> at kafka.server.KafkaApis.getTopicMetadata(KafkaApis.scala:503)
> at
> kafka.server.KafkaApis.handleTopicMetadataRequest(KafkaApis.scala:542)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:62)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)
> at java.lang.Thread.run(Thread.java:745)
>
> So this code in KafkaApis.getTopicMetadata
>
> val offsetsTopicReplicationFactor =
> if (aliveBrokers.length > 0)
> Math.min(config.offsetsTopicReplicationFactor, aliveBrokers.length)
> else
> config.offsetsTopicReplicationFactor
>
> should probably look like
>
> val offsetsTopicReplicationFactor =
> if (aliveBrokers.length > 0)
> Math.min(config.offsetsTopicReplicationFactor, aliveBrokers.length)
> else
> Math.min(config.offsetsTopicReplicationFactor,
> ZkUtils.getChildrenParentMayNotExist(zkClient,
> ZkUtils.BrokerIdsPath).length)
>
> Rationale: since aliveBrokers is actually information from (metadata)
> cache - if broker has in cache info that there are 0 brokers alive, that
> surely means that cache hasn't been set/updated ever, and in that case do
> not assume that there are enough (3) brokers up, but instead lookup in ZK
> how many brokers are alive, and based on that determine replication factor
> for offsets topic.
>
> Does this make sense?
>
> As workaround, I guess I will have to resort to explicitly creating
> offsets topic if it doesn't exist already.
>
> Kind regards,
> Stevo Slavic.
>
> On Tue, Oct 6, 2015 at 11:34 AM, Stevo Slavić <ss...@gmail.com> wrote:
>
>> Debugged, and found in KafkaApis.handleConsumerMetadataRequest that
>> consumer offsets topic gets created on first lookup of offsets topic
>> metadata, even when auto topic creation is disabled.
>>
>> In that method there is following call:
>>
>> // get metadata (and create the topic if necessary)
>> val offsetsTopicMetadata =
>> getTopicMetadata(Set(OffsetManager.OffsetsTopicName)).head
>>
>> and it returns on first call offsetsTopicMetadata with empty
>> partitionsMetadata sequence and errorCode 5.
>>
>> By the docs this means:
>> LeaderNotAvailable => This error is thrown if we are in the middle of a
>> leadership election and there is currently no leader for this partition and
>> hence it is unavailable for writes.
>>
>> Since leadership election takes some time, especially for default 50
>> partitions of consumer offsets topic, first call to lookup consumer
>> metadata request will always return an error. This sounds like a bug,
>> especially if it's not documented as to be expected behavior. I'd prefer if
>> Kafka broker/controller did this init of consumer offsets topic on startup.
>>
>> As workaround, as init step of my integration test, I will lookup offsets
>> topic metadata, until successful or timeout.
>>
>> Kind regards,
>> Stevo Slavic.
>>
>> On Tue, Oct 6, 2015 at 10:02 AM, Stevo Slavić <ss...@gmail.com> wrote:
>>
>>> Thanks Grant for quick reply!
>>>
>>> I've used AdminUtils.topicExists("__consumer_offsets") check and even
>>> 10sec after Kafka broker startup, the check fails.
>>>
>>> When, on which event, does this internal topic get created? Is there
>>> some broker config property preventing it from being created? Does one have
>>> to use high level consumer or make some special request (JoingGroup?) using
>>> simple consumer API to trigger consumer offsets topic init on broker?
>>>
>>> I'm using simple consumer API - I assume exclude.internal.topics,
>>> offsets.storage or dual.commit.enabled however configured shouldn't affect
>>> me, since I'm passing OffsetCommitRequest with version id 1, and even more
>>> I do not even reach point where commit is done, since lookup of consumer
>>> coordinator is throwing ConsumerCoordinatorNotAvailableException.
>>>
>>> Kind regards,
>>> Stevo Slavic.
>>>
>>> On Mon, Oct 5, 2015 at 5:59 PM, Grant Henke <gh...@cloudera.com> wrote:
>>>
>>>> Hi Stevo,
>>>>
>>>> There are a couple of options to verify the topic exists:
>>>>
>>>> 1. Consume from a topic with "offsets.storage=kafka". If its not
>>>> created
>>>> already, this should create it.
>>>> 2. List and describe the topic using the Kafka topics script. Ex:
>>>>
>>>> bin/kafka-topics.sh --zookeeper localhost:2181 --list
>>>>
>>>> bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic
>>>> __consumer_offsets
>>>>
>>>>
>>>> 1. Check the ZNode exists in Zookeeper. Ex:
>>>>
>>>> bin/zookeeper-shell.sh localhost:2181
>>>> ls /brokers/topics/__consumer_offsets
>>>>
>>>> get /brokers/topics/__consumer_offsets
>>>>
>>>>
>>>> Thanks,
>>>> Grant
>>>>
>>>> On Mon, Oct 5, 2015 at 10:44 AM, Stevo Slavić <ss...@gmail.com>
>>>> wrote:
>>>>
>>>> > Hello Apache Kafka community,
>>>> >
>>>> > In my integration tests, with single 0.8.2.2 broker, for newly created
>>>> > topic with single partition, after determining through topic metadata
>>>> > request that partition has lead broker assigned, when I try to reset
>>>> offset
>>>> > for given consumer group, I first try to discover offset coordinator
>>>> and
>>>> > that lookup is throwing ConsumerCoordinatorNotAvailableException
>>>> >
>>>> > On
>>>> >
>>>> >
>>>> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
>>>> > it is documented that broker returns
>>>> ConsumerCoordinatorNotAvailableCode
>>>> > for consumer metadata requests or offset commit requests if the
>>>> offsets
>>>> > topic has not yet been created.
>>>> >
>>>> > I wonder if this is really the case, that the offsets topic has not
>>>> been
>>>> > created. Any tips how to ensure/verify that offsets topic exists?
>>>> >
>>>> > Kind regards,
>>>> >
>>>> > Stevo Slavic.
>>>> >
>>>>
>>>>
>>>>
>>>> --
>>>> Grant Henke
>>>> Software Engineer | Cloudera
>>>> grant@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>>>>
>>>
>>>
>>
>
Re: How to verify offsets topic exists?
Posted by Stevo Slavić <ss...@gmail.com>.
There's another related bug - triggering offsets topic creation through
requesting metadata about that topic does not work in case of single broker
clean (no topics created yet) Kafka cluster running. In that case sequence
returned by KafkaApis.getAliveBrokers is empty, and
KafkaApis.getTopicMetadata will issue a AdminUtils.createTopic request with
offsetsTopicReplicationFactor of 3, which will fail with
ERROR [KafkaApi-0] error when handling request Name: TopicMetadataRequest;
Version: 0; CorrelationId: 0; ClientId: ; Topics: __consumer_offsets
(kafka.server.KafkaApis)
kafka.admin.AdminOperationException: replication factor: 3 larger than
available brokers: 1
at kafka.admin.AdminUtils$.assignReplicasToBrokers(AdminUtils.scala:70)
at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:171)
at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:513)
at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:503)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at scala.collection.immutable.Set$Set1.foreach(Set.scala:79)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
at
scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)
at scala.collection.SetLike$class.map(SetLike.scala:92)
at scala.collection.AbstractSet.map(Set.scala:47)
at kafka.server.KafkaApis.getTopicMetadata(KafkaApis.scala:503)
at
kafka.server.KafkaApis.handleTopicMetadataRequest(KafkaApis.scala:542)
at kafka.server.KafkaApis.handle(KafkaApis.scala:62)
at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)
at java.lang.Thread.run(Thread.java:745)
So this code in KafkaApis.getTopicMetadata
val offsetsTopicReplicationFactor =
if (aliveBrokers.length > 0)
Math.min(config.offsetsTopicReplicationFactor, aliveBrokers.length)
else
config.offsetsTopicReplicationFactor
should probably look like
val offsetsTopicReplicationFactor =
if (aliveBrokers.length > 0)
Math.min(config.offsetsTopicReplicationFactor, aliveBrokers.length)
else
Math.min(config.offsetsTopicReplicationFactor,
ZkUtils.getChildrenParentMayNotExist(zkClient,
ZkUtils.BrokerIdsPath).length)
Rationale: since aliveBrokers is actually information from (metadata) cache
- if broker has in cache info that there are 0 brokers alive, that surely
means that cache hasn't been set/updated ever, and in that case do not
assume that there are enough (3) brokers up, but instead lookup in ZK how
many brokers are alive, and based on that determine replication factor for
offsets topic.
Does this make sense?
As workaround, I guess I will have to resort to explicitly creating offsets
topic if it doesn't exist already.
Kind regards,
Stevo Slavic.
On Tue, Oct 6, 2015 at 11:34 AM, Stevo Slavić <ss...@gmail.com> wrote:
> Debugged, and found in KafkaApis.handleConsumerMetadataRequest that
> consumer offsets topic gets created on first lookup of offsets topic
> metadata, even when auto topic creation is disabled.
>
> In that method there is following call:
>
> // get metadata (and create the topic if necessary)
> val offsetsTopicMetadata =
> getTopicMetadata(Set(OffsetManager.OffsetsTopicName)).head
>
> and it returns on first call offsetsTopicMetadata with empty
> partitionsMetadata sequence and errorCode 5.
>
> By the docs this means:
> LeaderNotAvailable => This error is thrown if we are in the middle of a
> leadership election and there is currently no leader for this partition and
> hence it is unavailable for writes.
>
> Since leadership election takes some time, especially for default 50
> partitions of consumer offsets topic, first call to lookup consumer
> metadata request will always return an error. This sounds like a bug,
> especially if it's not documented as to be expected behavior. I'd prefer if
> Kafka broker/controller did this init of consumer offsets topic on startup.
>
> As workaround, as init step of my integration test, I will lookup offsets
> topic metadata, until successful or timeout.
>
> Kind regards,
> Stevo Slavic.
>
> On Tue, Oct 6, 2015 at 10:02 AM, Stevo Slavić <ss...@gmail.com> wrote:
>
>> Thanks Grant for quick reply!
>>
>> I've used AdminUtils.topicExists("__consumer_offsets") check and even
>> 10sec after Kafka broker startup, the check fails.
>>
>> When, on which event, does this internal topic get created? Is there some
>> broker config property preventing it from being created? Does one have to
>> use high level consumer or make some special request (JoingGroup?) using
>> simple consumer API to trigger consumer offsets topic init on broker?
>>
>> I'm using simple consumer API - I assume exclude.internal.topics,
>> offsets.storage or dual.commit.enabled however configured shouldn't affect
>> me, since I'm passing OffsetCommitRequest with version id 1, and even more
>> I do not even reach point where commit is done, since lookup of consumer
>> coordinator is throwing ConsumerCoordinatorNotAvailableException.
>>
>> Kind regards,
>> Stevo Slavic.
>>
>> On Mon, Oct 5, 2015 at 5:59 PM, Grant Henke <gh...@cloudera.com> wrote:
>>
>>> Hi Stevo,
>>>
>>> There are a couple of options to verify the topic exists:
>>>
>>> 1. Consume from a topic with "offsets.storage=kafka". If its not
>>> created
>>> already, this should create it.
>>> 2. List and describe the topic using the Kafka topics script. Ex:
>>>
>>> bin/kafka-topics.sh --zookeeper localhost:2181 --list
>>>
>>> bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic
>>> __consumer_offsets
>>>
>>>
>>> 1. Check the ZNode exists in Zookeeper. Ex:
>>>
>>> bin/zookeeper-shell.sh localhost:2181
>>> ls /brokers/topics/__consumer_offsets
>>>
>>> get /brokers/topics/__consumer_offsets
>>>
>>>
>>> Thanks,
>>> Grant
>>>
>>> On Mon, Oct 5, 2015 at 10:44 AM, Stevo Slavić <ss...@gmail.com> wrote:
>>>
>>> > Hello Apache Kafka community,
>>> >
>>> > In my integration tests, with single 0.8.2.2 broker, for newly created
>>> > topic with single partition, after determining through topic metadata
>>> > request that partition has lead broker assigned, when I try to reset
>>> offset
>>> > for given consumer group, I first try to discover offset coordinator
>>> and
>>> > that lookup is throwing ConsumerCoordinatorNotAvailableException
>>> >
>>> > On
>>> >
>>> >
>>> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
>>> > it is documented that broker returns
>>> ConsumerCoordinatorNotAvailableCode
>>> > for consumer metadata requests or offset commit requests if the offsets
>>> > topic has not yet been created.
>>> >
>>> > I wonder if this is really the case, that the offsets topic has not
>>> been
>>> > created. Any tips how to ensure/verify that offsets topic exists?
>>> >
>>> > Kind regards,
>>> >
>>> > Stevo Slavic.
>>> >
>>>
>>>
>>>
>>> --
>>> Grant Henke
>>> Software Engineer | Cloudera
>>> grant@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>>>
>>
>>
>
Re: How to verify offsets topic exists?
Posted by Stevo Slavić <ss...@gmail.com>.
Debugged, and found in KafkaApis.handleConsumerMetadataRequest that
consumer offsets topic gets created on first lookup of offsets topic
metadata, even when auto topic creation is disabled.
In that method there is following call:
// get metadata (and create the topic if necessary)
val offsetsTopicMetadata =
getTopicMetadata(Set(OffsetManager.OffsetsTopicName)).head
and it returns on first call offsetsTopicMetadata with empty
partitionsMetadata sequence and errorCode 5.
By the docs this means:
LeaderNotAvailable => This error is thrown if we are in the middle of a
leadership election and there is currently no leader for this partition and
hence it is unavailable for writes.
Since leadership election takes some time, especially for default 50
partitions of consumer offsets topic, first call to lookup consumer
metadata request will always return an error. This sounds like a bug,
especially if it's not documented as to be expected behavior. I'd prefer if
Kafka broker/controller did this init of consumer offsets topic on startup.
As workaround, as init step of my integration test, I will lookup offsets
topic metadata, until successful or timeout.
Kind regards,
Stevo Slavic.
On Tue, Oct 6, 2015 at 10:02 AM, Stevo Slavić <ss...@gmail.com> wrote:
> Thanks Grant for quick reply!
>
> I've used AdminUtils.topicExists("__consumer_offsets") check and even
> 10sec after Kafka broker startup, the check fails.
>
> When, on which event, does this internal topic get created? Is there some
> broker config property preventing it from being created? Does one have to
> use high level consumer or make some special request (JoingGroup?) using
> simple consumer API to trigger consumer offsets topic init on broker?
>
> I'm using simple consumer API - I assume exclude.internal.topics,
> offsets.storage or dual.commit.enabled however configured shouldn't affect
> me, since I'm passing OffsetCommitRequest with version id 1, and even more
> I do not even reach point where commit is done, since lookup of consumer
> coordinator is throwing ConsumerCoordinatorNotAvailableException.
>
> Kind regards,
> Stevo Slavic.
>
> On Mon, Oct 5, 2015 at 5:59 PM, Grant Henke <gh...@cloudera.com> wrote:
>
>> Hi Stevo,
>>
>> There are a couple of options to verify the topic exists:
>>
>> 1. Consume from a topic with "offsets.storage=kafka". If its not
>> created
>> already, this should create it.
>> 2. List and describe the topic using the Kafka topics script. Ex:
>>
>> bin/kafka-topics.sh --zookeeper localhost:2181 --list
>>
>> bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic
>> __consumer_offsets
>>
>>
>> 1. Check the ZNode exists in Zookeeper. Ex:
>>
>> bin/zookeeper-shell.sh localhost:2181
>> ls /brokers/topics/__consumer_offsets
>>
>> get /brokers/topics/__consumer_offsets
>>
>>
>> Thanks,
>> Grant
>>
>> On Mon, Oct 5, 2015 at 10:44 AM, Stevo Slavić <ss...@gmail.com> wrote:
>>
>> > Hello Apache Kafka community,
>> >
>> > In my integration tests, with single 0.8.2.2 broker, for newly created
>> > topic with single partition, after determining through topic metadata
>> > request that partition has lead broker assigned, when I try to reset
>> offset
>> > for given consumer group, I first try to discover offset coordinator and
>> > that lookup is throwing ConsumerCoordinatorNotAvailableException
>> >
>> > On
>> >
>> >
>> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
>> > it is documented that broker returns ConsumerCoordinatorNotAvailableCode
>> > for consumer metadata requests or offset commit requests if the offsets
>> > topic has not yet been created.
>> >
>> > I wonder if this is really the case, that the offsets topic has not been
>> > created. Any tips how to ensure/verify that offsets topic exists?
>> >
>> > Kind regards,
>> >
>> > Stevo Slavic.
>> >
>>
>>
>>
>> --
>> Grant Henke
>> Software Engineer | Cloudera
>> grant@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>>
>
>
Re: How to verify offsets topic exists?
Posted by Stevo Slavić <ss...@gmail.com>.
Thanks Grant for quick reply!
I've used AdminUtils.topicExists("__consumer_offsets") check and even 10sec
after Kafka broker startup, the check fails.
When, on which event, does this internal topic get created? Is there some
broker config property preventing it from being created? Does one have to
use high level consumer or make some special request (JoingGroup?) using
simple consumer API to trigger consumer offsets topic init on broker?
I'm using simple consumer API - I assume exclude.internal.topics,
offsets.storage or dual.commit.enabled however configured shouldn't affect
me, since I'm passing OffsetCommitRequest with version id 1, and even more
I do not even reach point where commit is done, since lookup of consumer
coordinator is throwing ConsumerCoordinatorNotAvailableException.
Kind regards,
Stevo Slavic.
On Mon, Oct 5, 2015 at 5:59 PM, Grant Henke <gh...@cloudera.com> wrote:
> Hi Stevo,
>
> There are a couple of options to verify the topic exists:
>
> 1. Consume from a topic with "offsets.storage=kafka". If its not created
> already, this should create it.
> 2. List and describe the topic using the Kafka topics script. Ex:
>
> bin/kafka-topics.sh --zookeeper localhost:2181 --list
>
> bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic
> __consumer_offsets
>
>
> 1. Check the ZNode exists in Zookeeper. Ex:
>
> bin/zookeeper-shell.sh localhost:2181
> ls /brokers/topics/__consumer_offsets
>
> get /brokers/topics/__consumer_offsets
>
>
> Thanks,
> Grant
>
> On Mon, Oct 5, 2015 at 10:44 AM, Stevo Slavić <ss...@gmail.com> wrote:
>
> > Hello Apache Kafka community,
> >
> > In my integration tests, with single 0.8.2.2 broker, for newly created
> > topic with single partition, after determining through topic metadata
> > request that partition has lead broker assigned, when I try to reset
> offset
> > for given consumer group, I first try to discover offset coordinator and
> > that lookup is throwing ConsumerCoordinatorNotAvailableException
> >
> > On
> >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
> > it is documented that broker returns ConsumerCoordinatorNotAvailableCode
> > for consumer metadata requests or offset commit requests if the offsets
> > topic has not yet been created.
> >
> > I wonder if this is really the case, that the offsets topic has not been
> > created. Any tips how to ensure/verify that offsets topic exists?
> >
> > Kind regards,
> >
> > Stevo Slavic.
> >
>
>
>
> --
> Grant Henke
> Software Engineer | Cloudera
> grant@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>
Re: How to verify offsets topic exists?
Posted by Grant Henke <gh...@cloudera.com>.
Hi Stevo,
There are a couple of options to verify the topic exists:
1. Consume from a topic with "offsets.storage=kafka". If its not created
already, this should create it.
2. List and describe the topic using the Kafka topics script. Ex:
bin/kafka-topics.sh --zookeeper localhost:2181 --list
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic
__consumer_offsets
1. Check the ZNode exists in Zookeeper. Ex:
bin/zookeeper-shell.sh localhost:2181
ls /brokers/topics/__consumer_offsets
get /brokers/topics/__consumer_offsets
Thanks,
Grant
On Mon, Oct 5, 2015 at 10:44 AM, Stevo Slavić <ss...@gmail.com> wrote:
> Hello Apache Kafka community,
>
> In my integration tests, with single 0.8.2.2 broker, for newly created
> topic with single partition, after determining through topic metadata
> request that partition has lead broker assigned, when I try to reset offset
> for given consumer group, I first try to discover offset coordinator and
> that lookup is throwing ConsumerCoordinatorNotAvailableException
>
> On
>
> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI
> it is documented that broker returns ConsumerCoordinatorNotAvailableCode
> for consumer metadata requests or offset commit requests if the offsets
> topic has not yet been created.
>
> I wonder if this is really the case, that the offsets topic has not been
> created. Any tips how to ensure/verify that offsets topic exists?
>
> Kind regards,
>
> Stevo Slavic.
>
--
Grant Henke
Software Engineer | Cloudera
grant@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke