kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Stevo Slavić <ssla...@gmail.com>
Subject Re: Explicit topic creation and topic metadata availability
Date Mon, 27 Oct 2014 08:55:08 GMT
OK,

thanks for heads up!

Is this requirement documented somewhere?
Would it make sense then to have AdminUtils, call setZkSerializer on
zkClient passed to it? Or maybe provide a factory method(s) for ZkClient in
AdminUtils, which would ensure  ZkSerializer is appropriate.

Kind regards,
Stevo Slavic.

On Mon, Oct 27, 2014 at 12:43 AM, Guozhang Wang <wangguoz@gmail.com> wrote:

> Hi Stevo,
>
> Sorry for the late reply.
>
> Yes, you have to use ZKStringSerializer when initializing zkClient. However
> this is not related to key.serializer.class of the producer (not the
> KafkaProducer, which is the new producer that does not need any key
> serializer any more).
>
> key.serializer.class and value serializer.class are used to specify how to
> encode / decode key value bytes of the message, while ZKStringSerializer
> specifies how to encode the bytes written in ZK metadata path.
>
> Guozhang
>
> On Wed, Oct 22, 2014 at 5:36 PM, Stevo Slavić <sslavic@gmail.com> wrote:
>
> > It seems that used ZkSerializer has to be aligned with KafkaProducer
> > configured key.serializer.class.
> >
> > On Thu, Oct 23, 2014 at 1:13 AM, Stevo Slavić <sslavic@gmail.com> wrote:
> >
> > > Still have to understand what is going on, but when I set
> > > kafka.utils.ZKStringSerializer to be ZkSerializer for ZkClient used in
> > > AdminUtils calls, KafkaProducer could see created topic...
> > > Default ZkSerializer is
> > > org.I0Itec.zkclient.serialize.SerializableSerializer.
> > >
> > > Kind regards,
> > > Stevo Slavic.
> > >
> > > On Wed, Oct 22, 2014 at 10:03 PM, Stevo Slavić <sslavic@gmail.com>
> > wrote:
> > >
> > >> Output on trunk is clean too, after clean build:
> > >>
> > >> ~/git/oss/kafka [trunk|✔]
> > >> 22:00 $ bin/kafka-topics.sh --zookeeper 127.0.0.1:50194 --topic
> > >> 059915e6-56ef-4b8e-8e95-9f676313a01c --describe
> > >> Error while executing topic command next on empty iterator
> > >> java.util.NoSuchElementException: next on empty iterator
> > >>     at scala.collection.Iterator$$anon$2.next(Iterator.scala:39)
> > >>     at scala.collection.Iterator$$anon$2.next(Iterator.scala:37)
> > >>     at scala.collection.IterableLike$class.head(IterableLike.scala:91)
> > >>     at scala.collection.AbstractIterable.head(Iterable.scala:54)
> > >>     at
> > >>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:137)
> > >>     at
> > >>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:127)
> > >>     at
> > >>
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > >>     at
> > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > >>     at kafka.admin.TopicCommand$.describeTopic(TopicCommand.scala:127)
> > >>     at kafka.admin.TopicCommand$.main(TopicCommand.scala:56)
> > >>     at kafka.admin.TopicCommand.main(TopicCommand.scala)
> > >>
> > >> On Wed, Oct 22, 2014 at 9:45 PM, Stevo Slavić <sslavic@gmail.com>
> > wrote:
> > >>
> > >>> kafka-topics.sh execution, from latest trunk:
> > >>>
> > >>> ~/git/oss/kafka [trunk|✔]
> > >>> 21:00 $ bin/kafka-topics.sh --zookeeper 127.0.0.1:50194 --topic
> > >>> 059915e6-56ef-4b8e-8e95-9f676313a01c --describe
> > >>> SLF4J: Class path contains multiple SLF4J bindings.
> > >>> SLF4J: Found binding in
> > >>>
> >
> [jar:file:/Users/d062007/git/oss/kafka/core/build/dependant-libs-2.10.1/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> > >>> SLF4J: Found binding in
> > >>>
> >
> [jar:file:/Users/d062007/git/oss/kafka/core/build/dependant-libs-2.10.1/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> > >>> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> > >>> explanation.
> > >>> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> > >>> Error while executing topic command next on empty iterator
> > >>> java.util.NoSuchElementException: next on empty iterator
> > >>>     at scala.collection.Iterator$$anon$2.next(Iterator.scala:39)
> > >>>     at scala.collection.Iterator$$anon$2.next(Iterator.scala:37)
> > >>>     at
> scala.collection.IterableLike$class.head(IterableLike.scala:91)
> > >>>     at scala.collection.AbstractIterable.head(Iterable.scala:54)
> > >>>     at
> > >>>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:170)
> > >>>     at
> > >>>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:160)
> > >>>     at
> > >>>
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > >>>     at
> > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > >>>     at
> kafka.admin.TopicCommand$.describeTopic(TopicCommand.scala:160)
> > >>>     at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
> > >>>     at kafka.admin.TopicCommand.main(TopicCommand.scala)
> > >>>
> > >>>
> > >>> Output from same command on 0.8.1 branch is better, but still same
> > >>> exception:
> > >>>
> > >>> ~/git/oss/kafka [0.8.1|✔]
> > >>> 21:12 $ bin/kafka-topics.sh --zookeeper 127.0.0.1:50194 --topic
> > >>> 059915e6-56ef-4b8e-8e95-9f676313a01c --describe
> > >>> Error while executing topic command null
> > >>> java.util.NoSuchElementException
> > >>>     at
> scala.collection.IterableLike$class.head(IterableLike.scala:101)
> > >>>     at scala.collection.immutable.Map$EmptyMap$.head(Map.scala:73)
> > >>>     at
> > >>>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:137)
> > >>>     at
> > >>>
> >
> kafka.admin.TopicCommand$$anonfun$describeTopic$1.apply(TopicCommand.scala:127)
> > >>>     at
> > >>>
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57)
> > >>>     at
> > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43)
> > >>>     at
> kafka.admin.TopicCommand$.describeTopic(TopicCommand.scala:127)
> > >>>     at kafka.admin.TopicCommand$.main(TopicCommand.scala:56)
> > >>>     at kafka.admin.TopicCommand.main(TopicCommand.scala)
> > >>>
> > >>> On Wed, Oct 22, 2014 at 5:30 PM, Guozhang Wang <wangguoz@gmail.com>
> > >>> wrote:
> > >>>
> > >>>> Hello Stevo,
> > >>>>
> > >>>> Your understanding about the configs are correct, and it is indeed
> > wired
> > >>>> that the producer gets the exception after topic is created. Could
> you
> > >>>> use
> > >>>> the kafka-topics command to check if the leaders exist?
> > >>>>
> > >>>> kafka-topics.sh --zookeeper XXX --topic [topic-name] describe
> > >>>>
> > >>>> Guozhang
> > >>>>
> > >>>> On Wed, Oct 22, 2014 at 5:57 AM, Stevo Slavić <sslavic@gmail.com>
> > >>>> wrote:
> > >>>>
> > >>>> > Hello Apache Kafka users,
> > >>>> >
> > >>>> > Using Kafka 0.8.1.1 (single instance with single ZK 3.4.6
running
> > >>>> locally),
> > >>>> > with auto topic creation disabled, in a test I have topic
created
> > with
> > >>>> > AdminUtils.createTopic (AdminUtils.topicExists returns true)
but
> > >>>> > KafkaProducer on send request keeps throwing
> > >>>> > UnknownTopicOrPartitionException even after 100 retries, both
when
> > >>>> > topic.metadata.refresh.interval.ms and retry.backoff.ms are
left
> at
> > >>>> > defaults, and when customized.
> > >>>> >
> > >>>> > Am I doing something wrong or is this a known bug?
> > >>>> >
> > >>>> > How long does it typically take for metadata to be refreshed?
> > >>>> > How long does it take for leader to be elected?
> > >>>> >
> > >>>> > Documentation for retry.backoff.ms states:
> > >>>> > "Before each retry, the producer refreshes the metadata of
> relevant
> > >>>> topics
> > >>>> > to see if a new leader has been elected. Since leader election
> takes
> > >>>> a bit
> > >>>> > of time, this property specifies the amount of time that the
> > producer
> > >>>> waits
> > >>>> > before refreshing the metadata."
> > >>>> >
> > >>>> > Do I understand this docs correctly - on failure to send a
> message,
> > >>>> such as
> > >>>> > unknown topic, if retries are configured producer will wait
for
> > >>>> configured
> > >>>> > retry.backoff.ms, then it will initiate and wait for metadata
> > >>>> refresh to
> > >>>> > complete, and only then retry sending?
> > >>>> >
> > >>>> > Kind regards,
> > >>>> > Stevo Slavic.
> > >>>> >
> > >>>>
> > >>>>
> > >>>>
> > >>>> --
> > >>>> -- Guozhang
> > >>>>
> > >>>
> > >>>
> > >>
> > >
> >
>
>
>
> --
> -- Guozhang
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message