I work with Kafka 0.8 and zookeeper 3.3.5. In fact, we have a dozen of those that we consume without any problems.
Recently, we started to feed and consume a new topic that has strange behavior. The understated displacement was suddenly reset. It respects the auto.offset.reset policy that we set (actually the smallest), but I can’t understand why in this thread its reset is suddenly reset.
I use a high level consumer.
Here are some ERROR logs I found: We have a bunch of this error logs:
[2015-03-26 05: 21: 17,789] INFO Fetching metadata from broker id: 1, host: 172.16.23.1, port: 9092 with correlation id 47 for 1 topic (s) Set (MyTopic) (kafka.cl
ient.ClientUtils $)
[2015-03-26 05: 21: 17,789] ERROR Producer connection to 172.16.23.1:9092 unsuccessful (kafka.producer.SyncProducer)
java.nio.channels.ClosedByInterruptException
at java.nio.channels.spi.AbstractInterruptibleChannel.end (AbstractInterruptibleChannel.java:202)
at sun.nio.ch.SocketChannelImpl.connect (SocketChannelImpl.java:681)
at kafka.network.BlockingChannel.connect (BlockingChannel.scala: 57)
at kafka.producer.SyncProducer.connect (SyncProducer.scala: 141)
at kafka.producer.SyncProducer.getOrMakeConnection (SyncProducer.scala: 156)
at kafka.producer.SyncProducer.kafka $ producer $ SyncProducer $$ doSend (SyncProducer.scala: 68)
at kafka.producer.SyncProducer.send (SyncProducer.scala: 112)
at kafka.client.ClientUtils $ .fetchTopicMetadata (ClientUtils.scala: 53)
at kafka.client.ClientUtils $ .fetchTopicMetadata (ClientUtils.scala: 88)
at kafka.consumer.ConsumerFetcherManager $ LeaderFinderThread.doWork (ConsumerFetcherManager.scala: 66)
at kafka.utils.ShutdownableThread.run (ShutdownableThread.scala: 51)
Every time this problem occurs, I see that WARN log:
[2015-03-26 05: 21: 30,596] WARN Reconnect due to socket error: null (kafka.consumer.SimpleConsumer)
And then the real problem arises:
[2015-03-26 05: 21: 47,551] INFO Connected to 172.16.23.5:9092 for producing (kafka.producer.SyncProducer)
[2015-03-26 05: 21: 47,552] INFO Disconnecting from 172.16.23.5:9092 (kafka.producer.SyncProducer)
[2015-03-26 05: 21: 47,553] INFO [ConsumerFetcherManager-1427047649942] Added fetcher for partitions ArrayBuffer ([[[MyTopic, 0], initOffset 45268422051 to br
oker id:5,host:172.16.23.5,port:9092] ) (kafka.consumer.ConsumerFetcherManager)
[2015-03-26 05:21:47,553] INFO [ConsumerFetcherThread-MyTopic_group-1427047649884-699191d4-0-5], Starting (kafka.consumer.Cons
umerFetcherThread)
[2015-03-26 05:21:50,388] ERROR [ConsumerFetcherThread-MyTopic_group-1427047649884-699191d4-0-5], Current offset 45268422051 for partition [MyTopic,0] out of range; reset offset to 1948447612 (kafka.consumer.ConsumerFetcherThread)
[2015-03-26 05:21:50,490] ERROR [ConsumerFetcherThread-MyTopic_group-1427047649884-699191d4-0-5], Current offset 1948447612 for partition [MyTopic,0] out of range; reset offset to 1948447612 (kafka.consumer.ConsumerFetcherThread)
[2015-03-26 05:21:50,591] ERROR [ConsumerFetcherThread-MyTopic_group-1427047649884-699191d4-0-5], Current offset 1948447612 for partition [MyTopic,0] out of range; reset offset to 1948447612 (kafka.consumer.ConsumerFetcherThread)
[2015-03-26 05:21:50,692] ERROR [ConsumerFetcherThread-MyTopic_group-1427047649884-699191d4-0-5], Current offset 1948447612 for partition [MyTopic,0] out of range; reset offset to 1948447612 (kafka.consumer.ConsumerFetcherThread)
:
-, ?
-, , reset , auto.offset.reset ?
.