You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@samza.apache.org by Gaurav Agarwal <ga...@gmail.com> on 2017/05/02 03:47:29 UTC
Possible bug causing reset in offsets [Version: 0.12.0].
Hi All,
We recently observed an issue in our Samza application (version
0.12.0) where we found that that the message offsets "jumped back" causing
many of the messages to be re-processed.
On digging deeper - here is what we found:
- There was a some network related issues at that time causing temporary
communication losses between nodes.
- BrokerProxy tried to addTopicPartition() and while doing that, its
isValidOffset() method failed.
- This caused it to read the reset the offset using the configured setting
- which we have set to auto.offset.reset=smallest
Now, the validation check simply tries to fetch the message at the given
offset and if there is an exception while doing it, it assumes that the
offset is invalid. I think that the exception ReplicaNotAvailableException is
being considered harmless, but I am not sure what is the expected behavior
in case of any other exception. What about network related exceptions (that
are transient)?
I am pasting below the logs that show a *simple network error,* and
immediately after that the message that Samza thinks that the offset was
invalid and hence it is falling-back to reset.
Is there any workaround to get past this problem? I would have thought that
ideally, there could be only handful of well-known error codes that would
indicate that the offset into a kafka topic/partition is invalid.
--
thanks,
gaurav
INFO [2017-05-01 15:19:27,327] [U:1,898,F:1,159,T:3,056,M:3,056]
system.kafka.DefaultFetchSimpleConsumer:[Logging$class:info:76] -
[SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for client
samza_consumer-job4-1] - Reconnect due to error:
java.io.EOFException
at org.apache.kafka.common.network.NetworkReceive.
readFromReadableChannel(NetworkReceive.java:83)
at kafka.network.BlockingChannel.readCompletely(
BlockingChannel.scala:129)
at kafka.network.BlockingChannel.receive(BlockingChannel.scala:120)
at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.
scala:86)
at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$
$sendRequest(SimpleConsumer.scala:83)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:132)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
apply$mcV$sp$1.apply(SimpleConsumer.scala:132)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$
apply$mcV$sp$1.apply(SimpleConsumer.scala:132)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(
SimpleConsumer.scala:131)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(
SimpleConsumer.scala:131)
at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(
SimpleConsumer.scala:131)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:130)
at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.fetch(
DefaultFetchSimpleConsumer.scala:48)
at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.
defaultFetch(DefaultFetchSimpleConsumer.scala:41)
at org.apache.samza.system.kafka.GetOffset.isValidOffset(
GetOffset.scala:60)
at org.apache.samza.system.kafka.BrokerProxy.addTopicPartition(
BrokerProxy.scala:99)
at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$
refreshBrokers$2.refresh$1(KafkaSystemConsumer.scala:213)
at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$
refreshBrokers$2.apply(KafkaSystemConsumer.scala:226)
at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$
refreshBrokers$2.apply(KafkaSystemConsumer.scala:192)
at org.apache.samza.util.ExponentialSleepStrategy.run(
ExponentialSleepStrategy.scala:82)
at org.apache.samza.system.kafka.KafkaSystemConsumer.refreshBrokers(
KafkaSystemConsumer.scala:191)
at org.apache.samza.system.kafka.KafkaSystemConsumer$$anon$1.
abdicate(KafkaSystemConsumer.scala:293)
at org.apache.samza.system.kafka.BrokerProxy.abdicate(
BrokerProxy.scala:207)
at org.apache.samza.system.kafka.BrokerProxy$$anonfun$
handleErrors$2.apply(BrokerProxy.scala:245)
at org.apache.samza.system.kafka.BrokerProxy$$anonfun$
handleErrors$2.apply(BrokerProxy.scala:245)
at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
at org.apache.samza.system.kafka.BrokerProxy.handleErrors(
BrokerProxy.scala:245)
at org.apache.samza.system.kafka.BrokerProxy.org
<http://org.apache.samza.system.kafka.brokerproxy.org/>$apache$samza$
system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:186)
at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$
run$1.apply(BrokerProxy.scala:147)
at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$
run$1.apply(BrokerProxy.scala:134)
at org.apache.samza.util.ExponentialSleepStrategy.run(
ExponentialSleepStrategy.scala:82)
at org.apache.samza.system.kafka.BrokerProxy$$anon$1.run(
BrokerProxy.scala:133)
at java.lang.Thread.run(Thread.java:745)
WARN [2017-05-01 15:19:27,378] [U:1,904,F:1,153,T:3,056,M:3,056]
system.kafka.BrokerProxy:[Logging$class:warn:74] -
[SAMZA-BROKER-PROXY-BrokerProxy
thread pointed at platform1:9092 for client samza_consumer-job4-1] - It
appears that we received an invalid or empty offset Some(45039137) for
[Topic3,17]. Attempting to use Kafka's auto.offset.reset setting. This can
result in data loss if processing continues.
INFO [2017-05-01 15:19:27,398] [U:1,904,F:1,152,T:3,056,M:3,056]
system.kafka.KafkaSystemConsumer:[Logging$class:info:63] -
[SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform3:9092 for client
samza_consumer-job4-1] - Refreshing brokers for: Map([Topic3,17] ->
45039137)
INFO [2017-05-01 15:19:27,405] [U:1,904,F:1,152,T:3,056,M:3,056]
system.kafka.GetOffset:[Logging$class:info:63] -
[SAMZA-BROKER-PROXY-BrokerProxy
thread pointed at platform1:9092 for client samza_consumer-job4-1] -
Checking if auto.offset.reset is defined for topic Topic3
INFO [2017-05-01 15:19:27,432] [U:1,904,F:1,152,T:3,056,M:3,056]
system.kafka.GetOffset:[Logging$class:info:63] -
[SAMZA-BROKER-PROXY-BrokerProxy
thread pointed at platform1:9092 for client samza_consumer-job4-1] - Got
reset of type smallest.
def addTopicPartition(tp: TopicAndPartition, nextOffset: Option[String]) = {
debug("Adding new topic and partition %s to queue for %s" format (tp, host))
if (nextOffsets.containsKey(tp)) {
toss("Already consuming TopicPartition %s" format tp)
}
val offset = if (nextOffset.isDefined &&
offsetGetter.isValidOffset(simpleConsumer, tp, nextOffset.get)) {
nextOffset
.get
.toLong
} else {
warn("It appears that we received an invalid or empty offset %s
for %s. Attempting to use Kafka's auto.offset.reset setting. This can
result in data loss if processing continues." format (nextOffset, tp))
offsetGetter.getResetOffset(simpleConsumer, tp)
}
debug("Got offset %s for new topic and partition %s." format (offset, tp))
nextOffsets += tp -> offset
metrics.topicPartitions(host, port).set(nextOffsets.size)
}
def isValidOffset(consumer: DefaultFetchSimpleConsumer,
topicAndPartition: TopicAndPartition, offset: String) = {
info("Validating offset %s for topic and partition %s" format
(offset, topicAndPartition))
try {
val messages = consumer.defaultFetch((topicAndPartition, offset.toLong))
if (messages.hasError) {
KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic,
topicAndPartition.partition))
}
info("Able to successfully read from offset %s for topic and
partition %s. Using it to instantiate consumer." format (offset,
topicAndPartition))
true
} catch {
case e: OffsetOutOfRangeException => false
}
}
Re: Possible bug causing reset in offsets [Version: 0.12.0].
Posted by Jagadish Venkatraman <ja...@gmail.com>.
Hi Gaurav (and others in this thread),
My apologies for the late reply, This e-mail appears to have missed my
inbox somehow.
From your logs, It appears that there was a leadership change happening on
the Kafka side for this topic partition? If so, I would actually expect the
follower's offset to trail the leader's offset.
If you want to avoid this behavior, you can set *acks = -1* in your produce
request. That will ensure that the produce request succeeds iff. all
replicas in the ISR acknowledge the write. This will give you higher
durability guarantees at the cost of increased latency.
Hope that helps.
Best,
Jagadish
On Tue, May 2, 2017 at 11:44 AM, Gaurav Agarwal <ga...@gmail.com>
wrote:
> Some more logs from Kakja
>
> WARN [2017-05-01 15:21:19,132]
> kafka.server.ReplicaFetcherThread:[Logging$class:warn:83] -
> [ReplicaFetcherThread-0-3] - [ReplicaFetcherThread-0-3], Replica 0 for
> partition [Topic3,17] reset its fetch offset from 45039137 to current
> leader 3's latest offset 45039132
>
>
> INFO [2017-05-01 15:21:19,150] kafka.log.Log:[Logging$class:info:68] -
> [ReplicaFetcherThread-0-3] - Truncating log Topic3-17 to offset 45039132.
>
>
> ERROR [2017-05-01 15:21:19,248]
> kafka.server.ReplicaFetcherThread:[Logging$class:error:97] -
> [ReplicaFetcherThread-0-3] - [ReplicaFetcherThread-0-3], Current offset
> 45039137 for partition [Topic3,17] out of range; reset offset to 45039132
>
>
> So it appears that Samza does not have an issue here. It was Kafka itself
> that went back on offsets, leading to an out-of-bounds offset query by
> Samza.
>
>
> We need to now dig a bit more into Kafka !
>
> On Tue, May 2, 2017 at 11:35 PM, Gaurav Agarwal <ga...@gmail.com>
> wrote:
>
> > Looking further, the reason for this "jump back" seems not so straight
> > forward:
> > In Kafka's Simple Consumer code:
> >
> > private def sendRequest(request: RequestOrResponse): NetworkReceive = {
> > lock synchronized {
> > var response: NetworkReceive = null
> > try {
> > getOrMakeConnection()
> > blockingChannel.send(request)
> > response = blockingChannel.receive()
> > } catch {
> > case e : ClosedByInterruptException =>
> > throw e
> > // Should not observe this exception when running Kafka with Java
> 1.8
> > case e: AsynchronousCloseException =>
> > throw e
> > case e : Throwable =>
> > info("Reconnect due to error:", e)
> > // retry once
> > try {
> > reconnect()
> > blockingChannel.send(request)
> > response = blockingChannel.receive()
> > } catch {
> > case e: Throwable =>
> > disconnect()
> > throw e
> > }
> > }
> > response
> > }
> > }
> >
> >
> > Note that first exception is being printed (which is what is logged), but
> > the kafka client retries and throws back whatever exception it received.
> It
> > could be that this exception is an instance of
> OffsetOutOfRangeException that
> > caused Samza to thing that the offset is invalid.
> >
> > However, I am unable to understand what could have caused this? Did Kafka
> > return this exception accidentally or did Samza ask for the offset that
> was
> > beyond what was present in kafka queue?
> >
> >
> > On Tue, May 2, 2017 at 9:31 AM, Gaurav Agarwal <gauravagarwal4@gmail.com
> >
> > wrote:
> >
> >> This also seems somewhat related to the mail on this group a few days
> >> back - with subject 'Messages lost after broker failure'.
> >>
> >> If someone had set auto.offset.reset to largest, then reverse would
> >> happen - i.e samza skipping over kafka partition queue in face of such
> >> failures.
> >>
> >> On Tue, May 2, 2017 at 9:17 AM, Gaurav Agarwal <
> gauravagarwal4@gmail.com>
> >> wrote:
> >>
> >>> Hi All,
> >>>
> >>> We recently observed an issue in our Samza application (version
> >>> 0.12.0) where we found that that the message offsets "jumped back"
> causing
> >>> many of the messages to be re-processed.
> >>>
> >>> On digging deeper - here is what we found:
> >>>
> >>> - There was a some network related issues at that time causing
> >>> temporary communication losses between nodes.
> >>> - BrokerProxy tried to addTopicPartition() and while doing that, its
> >>> isValidOffset() method failed.
> >>> - This caused it to read the reset the offset using the configured
> >>> setting - which we have set to auto.offset.reset=smallest
> >>>
> >>> Now, the validation check simply tries to fetch the message at the
> given
> >>> offset and if there is an exception while doing it, it assumes that the
> >>> offset is invalid. I think that the exception ReplicaNotAvailableE
> >>> xception is being considered harmless, but I am not sure what is the
> >>> expected behavior in case of any other exception. What about network
> >>> related exceptions (that are transient)?
> >>>
> >>> I am pasting below the logs that show a *simple network error,* and
> >>> immediately after that the message that Samza thinks that the offset
> was
> >>> invalid and hence it is falling-back to reset.
> >>>
> >>> Is there any workaround to get past this problem? I would have thought
> >>> that ideally, there could be only handful of well-known error codes
> that
> >>> would indicate that the offset into a kafka topic/partition is invalid.
> >>>
> >>> --
> >>> thanks,
> >>> gaurav
> >>>
> >>>
> >>> INFO [2017-05-01 15:19:27,327] [U:1,898,F:1,159,T:3,056,M:3,056]
> >>> system.kafka.DefaultFetchSimpleConsumer:[Logging$class:info:76] -
> >>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> >>> client samza_consumer-job4-1] - Reconnect due to error:
> >>> java.io.EOFException
> >>> at org.apache.kafka.common.network.NetworkReceive.
> readFromReada
> >>> bleChannel(NetworkReceive.java:83)
> >>> at kafka.network.BlockingChannel.
> readCompletely(BlockingChannel
> >>> .scala:129)
> >>> at kafka.network.BlockingChannel.
> receive(BlockingChannel.scala:
> >>> 120)
> >>> at kafka.consumer.SimpleConsumer.
> liftedTree1$1(SimpleConsumer.s
> >>> cala:86)
> >>> at kafka.consumer.SimpleConsumer.
> kafka$consumer$SimpleConsumer$
> >>> $sendRequest(SimpleConsumer.scala:83)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1$$anonfun$appl
> >>> y$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:132)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1$$anonfun$appl
> >>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1$$anonfun$appl
> >>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
> >>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1.apply$mcV$sp(
> >>> SimpleConsumer.scala:131)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1.apply(SimpleC
> >>> onsumer.scala:131)
> >>> at kafka.consumer.SimpleConsumer$
> $anonfun$fetch$1.apply(SimpleC
> >>> onsumer.scala:131)
> >>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> >>> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:
> 130)
> >>> at org.apache.samza.system.kafka.
> DefaultFetchSimpleConsumer.fet
> >>> ch(DefaultFetchSimpleConsumer.scala:48)
> >>> at org.apache.samza.system.kafka.
> DefaultFetchSimpleConsumer.def
> >>> aultFetch(DefaultFetchSimpleConsumer.scala:41)
> >>> at org.apache.samza.system.kafka.
> GetOffset.isValidOffset(GetOff
> >>> set.scala:60)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy.addTopicPartition(
> >>> BrokerProxy.scala:99)
> >>> at org.apache.samza.system.kafka.
> KafkaSystemConsumer$$anonfun$r
> >>> efreshBrokers$2.refresh$1(KafkaSystemConsumer.scala:213)
> >>> at org.apache.samza.system.kafka.
> KafkaSystemConsumer$$anonfun$r
> >>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:226)
> >>> at org.apache.samza.system.kafka.
> KafkaSystemConsumer$$anonfun$r
> >>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:192)
> >>> at org.apache.samza.util.ExponentialSleepStrategy.run(
> Exponenti
> >>> alSleepStrategy.scala:82)
> >>> at org.apache.samza.system.kafka.
> KafkaSystemConsumer.refreshBro
> >>> kers(KafkaSystemConsumer.scala:191)
> >>> at org.apache.samza.system.kafka.
> KafkaSystemConsumer$$anon$1.ab
> >>> dicate(KafkaSystemConsumer.scala:293)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy.abdicate(BrokerPro
> >>> xy.scala:207)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy$$anonfun$handleErr
> >>> ors$2.apply(BrokerProxy.scala:245)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy$$anonfun$handleErr
> >>> ors$2.apply(BrokerProxy.scala:245)
> >>> at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy.handleErrors(Broke
> >>> rProxy.scala:245)
> >>> at org.apache.samza.system.kafka.BrokerProxy.org
> >>> <http://org.apache.samza.system.kafka.brokerproxy.org/>$apache$samz
> >>> a$system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:186)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy$$anon$1$$anonfun$r
> >>> un$1.apply(BrokerProxy.scala:147)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy$$anon$1$$anonfun$r
> >>> un$1.apply(BrokerProxy.scala:134)
> >>> at org.apache.samza.util.ExponentialSleepStrategy.run(
> Exponenti
> >>> alSleepStrategy.scala:82)
> >>> at org.apache.samza.system.kafka.
> BrokerProxy$$anon$1.run(Broker
> >>> Proxy.scala:133)
> >>> at java.lang.Thread.run(Thread.java:745)
> >>>
> >>> WARN [2017-05-01 15:19:27,378] [U:1,904,F:1,153,T:3,056,M:3,056]
> >>> system.kafka.BrokerProxy:[Logging$class:warn:74] -
> >>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> >>> client samza_consumer-job4-1] - It appears that we received an invalid
> or
> >>> empty offset Some(45039137) for [Topic3,17]. Attempting to use Kafka's
> >>> auto.offset.reset setting. This can result in data loss if processing
> >>> continues.
> >>> INFO [2017-05-01 15:19:27,398] [U:1,904,F:1,152,T:3,056,M:3,056]
> >>> system.kafka.KafkaSystemConsumer:[Logging$class:info:63] -
> >>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform3:9092 for
> >>> client samza_consumer-job4-1] - Refreshing brokers for:
> Map([Topic3,17] ->
> >>> 45039137)
> >>> INFO [2017-05-01 15:19:27,405] [U:1,904,F:1,152,T:3,056,M:3,056]
> >>> system.kafka.GetOffset:[Logging$class:info:63] -
> >>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> >>> client samza_consumer-job4-1] - Checking if auto.offset.reset is
> defined
> >>> for topic Topic3
> >>> INFO [2017-05-01 15:19:27,432] [U:1,904,F:1,152,T:3,056,M:3,056]
> >>> system.kafka.GetOffset:[Logging$class:info:63] -
> >>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> >>> client samza_consumer-job4-1] - Got reset of type smallest.
> >>>
> >>> def addTopicPartition(tp: TopicAndPartition, nextOffset:
> Option[String]) = {
> >>> debug("Adding new topic and partition %s to queue for %s" format
> (tp, host))
> >>> if (nextOffsets.containsKey(tp)) {
> >>> toss("Already consuming TopicPartition %s" format tp)
> >>> }
> >>>
> >>> val offset = if (nextOffset.isDefined && offsetGetter.isValidOffset(simpleConsumer,
> tp, nextOffset.get)) {
> >>> nextOffset
> >>> .get
> >>> .toLong
> >>> } else {
> >>> warn("It appears that we received an invalid or empty offset %s
> for %s. Attempting to use Kafka's auto.offset.reset setting. This can
> result in data loss if processing continues." format (nextOffset, tp))
> >>> offsetGetter.getResetOffset(simpleConsumer, tp)
> >>> }
> >>>
> >>> debug("Got offset %s for new topic and partition %s." format
> (offset, tp))
> >>> nextOffsets += tp -> offset
> >>> metrics.topicPartitions(host, port).set(nextOffsets.size)
> >>> }
> >>>
> >>> def isValidOffset(consumer: DefaultFetchSimpleConsumer,
> topicAndPartition: TopicAndPartition, offset: String) = {
> >>> info("Validating offset %s for topic and partition %s" format
> (offset, topicAndPartition))
> >>>
> >>> try {
> >>> val messages = consumer.defaultFetch((topicAndPartition,
> offset.toLong))
> >>>
> >>> if (messages.hasError) {
> >>> KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic,
> topicAndPartition.partition))
> >>> }
> >>>
> >>> info("Able to successfully read from offset %s for topic and
> partition %s. Using it to instantiate consumer." format (offset,
> topicAndPartition))
> >>>
> >>> true
> >>> } catch {
> >>> case e: OffsetOutOfRangeException => false
> >>> }
> >>> }
> >>>
> >>>
> >>
> >
>
--
Jagadish V,
Graduate Student,
Department of Computer Science,
Stanford University
Re: Possible bug causing reset in offsets [Version: 0.12.0].
Posted by Gaurav Agarwal <ga...@gmail.com>.
Some more logs from Kakja
WARN [2017-05-01 15:21:19,132]
kafka.server.ReplicaFetcherThread:[Logging$class:warn:83] -
[ReplicaFetcherThread-0-3] - [ReplicaFetcherThread-0-3], Replica 0 for
partition [Topic3,17] reset its fetch offset from 45039137 to current
leader 3's latest offset 45039132
INFO [2017-05-01 15:21:19,150] kafka.log.Log:[Logging$class:info:68] -
[ReplicaFetcherThread-0-3] - Truncating log Topic3-17 to offset 45039132.
ERROR [2017-05-01 15:21:19,248]
kafka.server.ReplicaFetcherThread:[Logging$class:error:97] -
[ReplicaFetcherThread-0-3] - [ReplicaFetcherThread-0-3], Current offset
45039137 for partition [Topic3,17] out of range; reset offset to 45039132
So it appears that Samza does not have an issue here. It was Kafka itself
that went back on offsets, leading to an out-of-bounds offset query by
Samza.
We need to now dig a bit more into Kafka !
On Tue, May 2, 2017 at 11:35 PM, Gaurav Agarwal <ga...@gmail.com>
wrote:
> Looking further, the reason for this "jump back" seems not so straight
> forward:
> In Kafka's Simple Consumer code:
>
> private def sendRequest(request: RequestOrResponse): NetworkReceive = {
> lock synchronized {
> var response: NetworkReceive = null
> try {
> getOrMakeConnection()
> blockingChannel.send(request)
> response = blockingChannel.receive()
> } catch {
> case e : ClosedByInterruptException =>
> throw e
> // Should not observe this exception when running Kafka with Java 1.8
> case e: AsynchronousCloseException =>
> throw e
> case e : Throwable =>
> info("Reconnect due to error:", e)
> // retry once
> try {
> reconnect()
> blockingChannel.send(request)
> response = blockingChannel.receive()
> } catch {
> case e: Throwable =>
> disconnect()
> throw e
> }
> }
> response
> }
> }
>
>
> Note that first exception is being printed (which is what is logged), but
> the kafka client retries and throws back whatever exception it received. It
> could be that this exception is an instance of OffsetOutOfRangeException that
> caused Samza to thing that the offset is invalid.
>
> However, I am unable to understand what could have caused this? Did Kafka
> return this exception accidentally or did Samza ask for the offset that was
> beyond what was present in kafka queue?
>
>
> On Tue, May 2, 2017 at 9:31 AM, Gaurav Agarwal <ga...@gmail.com>
> wrote:
>
>> This also seems somewhat related to the mail on this group a few days
>> back - with subject 'Messages lost after broker failure'.
>>
>> If someone had set auto.offset.reset to largest, then reverse would
>> happen - i.e samza skipping over kafka partition queue in face of such
>> failures.
>>
>> On Tue, May 2, 2017 at 9:17 AM, Gaurav Agarwal <ga...@gmail.com>
>> wrote:
>>
>>> Hi All,
>>>
>>> We recently observed an issue in our Samza application (version
>>> 0.12.0) where we found that that the message offsets "jumped back" causing
>>> many of the messages to be re-processed.
>>>
>>> On digging deeper - here is what we found:
>>>
>>> - There was a some network related issues at that time causing
>>> temporary communication losses between nodes.
>>> - BrokerProxy tried to addTopicPartition() and while doing that, its
>>> isValidOffset() method failed.
>>> - This caused it to read the reset the offset using the configured
>>> setting - which we have set to auto.offset.reset=smallest
>>>
>>> Now, the validation check simply tries to fetch the message at the given
>>> offset and if there is an exception while doing it, it assumes that the
>>> offset is invalid. I think that the exception ReplicaNotAvailableE
>>> xception is being considered harmless, but I am not sure what is the
>>> expected behavior in case of any other exception. What about network
>>> related exceptions (that are transient)?
>>>
>>> I am pasting below the logs that show a *simple network error,* and
>>> immediately after that the message that Samza thinks that the offset was
>>> invalid and hence it is falling-back to reset.
>>>
>>> Is there any workaround to get past this problem? I would have thought
>>> that ideally, there could be only handful of well-known error codes that
>>> would indicate that the offset into a kafka topic/partition is invalid.
>>>
>>> --
>>> thanks,
>>> gaurav
>>>
>>>
>>> INFO [2017-05-01 15:19:27,327] [U:1,898,F:1,159,T:3,056,M:3,056]
>>> system.kafka.DefaultFetchSimpleConsumer:[Logging$class:info:76] -
>>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>>> client samza_consumer-job4-1] - Reconnect due to error:
>>> java.io.EOFException
>>> at org.apache.kafka.common.network.NetworkReceive.readFromReada
>>> bleChannel(NetworkReceive.java:83)
>>> at kafka.network.BlockingChannel.readCompletely(BlockingChannel
>>> .scala:129)
>>> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:
>>> 120)
>>> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.s
>>> cala:86)
>>> at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$
>>> $sendRequest(SimpleConsumer.scala:83)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>>> y$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:132)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
>>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(
>>> SimpleConsumer.scala:131)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
>>> onsumer.scala:131)
>>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
>>> onsumer.scala:131)
>>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>>> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:130)
>>> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.fet
>>> ch(DefaultFetchSimpleConsumer.scala:48)
>>> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.def
>>> aultFetch(DefaultFetchSimpleConsumer.scala:41)
>>> at org.apache.samza.system.kafka.GetOffset.isValidOffset(GetOff
>>> set.scala:60)
>>> at org.apache.samza.system.kafka.BrokerProxy.addTopicPartition(
>>> BrokerProxy.scala:99)
>>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>>> efreshBrokers$2.refresh$1(KafkaSystemConsumer.scala:213)
>>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:226)
>>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:192)
>>> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
>>> alSleepStrategy.scala:82)
>>> at org.apache.samza.system.kafka.KafkaSystemConsumer.refreshBro
>>> kers(KafkaSystemConsumer.scala:191)
>>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anon$1.ab
>>> dicate(KafkaSystemConsumer.scala:293)
>>> at org.apache.samza.system.kafka.BrokerProxy.abdicate(BrokerPro
>>> xy.scala:207)
>>> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
>>> ors$2.apply(BrokerProxy.scala:245)
>>> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
>>> ors$2.apply(BrokerProxy.scala:245)
>>> at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
>>> at org.apache.samza.system.kafka.BrokerProxy.handleErrors(Broke
>>> rProxy.scala:245)
>>> at org.apache.samza.system.kafka.BrokerProxy.org
>>> <http://org.apache.samza.system.kafka.brokerproxy.org/>$apache$samz
>>> a$system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:186)
>>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
>>> un$1.apply(BrokerProxy.scala:147)
>>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
>>> un$1.apply(BrokerProxy.scala:134)
>>> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
>>> alSleepStrategy.scala:82)
>>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1.run(Broker
>>> Proxy.scala:133)
>>> at java.lang.Thread.run(Thread.java:745)
>>>
>>> WARN [2017-05-01 15:19:27,378] [U:1,904,F:1,153,T:3,056,M:3,056]
>>> system.kafka.BrokerProxy:[Logging$class:warn:74] -
>>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>>> client samza_consumer-job4-1] - It appears that we received an invalid or
>>> empty offset Some(45039137) for [Topic3,17]. Attempting to use Kafka's
>>> auto.offset.reset setting. This can result in data loss if processing
>>> continues.
>>> INFO [2017-05-01 15:19:27,398] [U:1,904,F:1,152,T:3,056,M:3,056]
>>> system.kafka.KafkaSystemConsumer:[Logging$class:info:63] -
>>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform3:9092 for
>>> client samza_consumer-job4-1] - Refreshing brokers for: Map([Topic3,17] ->
>>> 45039137)
>>> INFO [2017-05-01 15:19:27,405] [U:1,904,F:1,152,T:3,056,M:3,056]
>>> system.kafka.GetOffset:[Logging$class:info:63] -
>>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>>> client samza_consumer-job4-1] - Checking if auto.offset.reset is defined
>>> for topic Topic3
>>> INFO [2017-05-01 15:19:27,432] [U:1,904,F:1,152,T:3,056,M:3,056]
>>> system.kafka.GetOffset:[Logging$class:info:63] -
>>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>>> client samza_consumer-job4-1] - Got reset of type smallest.
>>>
>>> def addTopicPartition(tp: TopicAndPartition, nextOffset: Option[String]) = {
>>> debug("Adding new topic and partition %s to queue for %s" format (tp, host))
>>> if (nextOffsets.containsKey(tp)) {
>>> toss("Already consuming TopicPartition %s" format tp)
>>> }
>>>
>>> val offset = if (nextOffset.isDefined && offsetGetter.isValidOffset(simpleConsumer, tp, nextOffset.get)) {
>>> nextOffset
>>> .get
>>> .toLong
>>> } else {
>>> warn("It appears that we received an invalid or empty offset %s for %s. Attempting to use Kafka's auto.offset.reset setting. This can result in data loss if processing continues." format (nextOffset, tp))
>>> offsetGetter.getResetOffset(simpleConsumer, tp)
>>> }
>>>
>>> debug("Got offset %s for new topic and partition %s." format (offset, tp))
>>> nextOffsets += tp -> offset
>>> metrics.topicPartitions(host, port).set(nextOffsets.size)
>>> }
>>>
>>> def isValidOffset(consumer: DefaultFetchSimpleConsumer, topicAndPartition: TopicAndPartition, offset: String) = {
>>> info("Validating offset %s for topic and partition %s" format (offset, topicAndPartition))
>>>
>>> try {
>>> val messages = consumer.defaultFetch((topicAndPartition, offset.toLong))
>>>
>>> if (messages.hasError) {
>>> KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic, topicAndPartition.partition))
>>> }
>>>
>>> info("Able to successfully read from offset %s for topic and partition %s. Using it to instantiate consumer." format (offset, topicAndPartition))
>>>
>>> true
>>> } catch {
>>> case e: OffsetOutOfRangeException => false
>>> }
>>> }
>>>
>>>
>>
>
Re: Possible bug causing reset in offsets [Version: 0.12.0].
Posted by Gaurav Agarwal <ga...@gmail.com>.
Looking further, the reason for this "jump back" seems not so straight
forward:
In Kafka's Simple Consumer code:
private def sendRequest(request: RequestOrResponse): NetworkReceive = {
lock synchronized {
var response: NetworkReceive = null
try {
getOrMakeConnection()
blockingChannel.send(request)
response = blockingChannel.receive()
} catch {
case e : ClosedByInterruptException =>
throw e
// Should not observe this exception when running Kafka with Java 1.8
case e: AsynchronousCloseException =>
throw e
case e : Throwable =>
info("Reconnect due to error:", e)
// retry once
try {
reconnect()
blockingChannel.send(request)
response = blockingChannel.receive()
} catch {
case e: Throwable =>
disconnect()
throw e
}
}
response
}
}
Note that first exception is being printed (which is what is logged), but
the kafka client retries and throws back whatever exception it received. It
could be that this exception is an instance of OffsetOutOfRangeException that
caused Samza to thing that the offset is invalid.
However, I am unable to understand what could have caused this? Did Kafka
return this exception accidentally or did Samza ask for the offset that was
beyond what was present in kafka queue?
On Tue, May 2, 2017 at 9:31 AM, Gaurav Agarwal <ga...@gmail.com>
wrote:
> This also seems somewhat related to the mail on this group a few days back
> - with subject 'Messages lost after broker failure'.
>
> If someone had set auto.offset.reset to largest, then reverse would
> happen - i.e samza skipping over kafka partition queue in face of such
> failures.
>
> On Tue, May 2, 2017 at 9:17 AM, Gaurav Agarwal <ga...@gmail.com>
> wrote:
>
>> Hi All,
>>
>> We recently observed an issue in our Samza application (version
>> 0.12.0) where we found that that the message offsets "jumped back" causing
>> many of the messages to be re-processed.
>>
>> On digging deeper - here is what we found:
>>
>> - There was a some network related issues at that time causing temporary
>> communication losses between nodes.
>> - BrokerProxy tried to addTopicPartition() and while doing that, its
>> isValidOffset() method failed.
>> - This caused it to read the reset the offset using the configured
>> setting - which we have set to auto.offset.reset=smallest
>>
>> Now, the validation check simply tries to fetch the message at the given
>> offset and if there is an exception while doing it, it assumes that the
>> offset is invalid. I think that the exception ReplicaNotAvailableE
>> xception is being considered harmless, but I am not sure what is the
>> expected behavior in case of any other exception. What about network
>> related exceptions (that are transient)?
>>
>> I am pasting below the logs that show a *simple network error,* and
>> immediately after that the message that Samza thinks that the offset was
>> invalid and hence it is falling-back to reset.
>>
>> Is there any workaround to get past this problem? I would have thought
>> that ideally, there could be only handful of well-known error codes that
>> would indicate that the offset into a kafka topic/partition is invalid.
>>
>> --
>> thanks,
>> gaurav
>>
>>
>> INFO [2017-05-01 15:19:27,327] [U:1,898,F:1,159,T:3,056,M:3,056]
>> system.kafka.DefaultFetchSimpleConsumer:[Logging$class:info:76] -
>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>> client samza_consumer-job4-1] - Reconnect due to error:
>> java.io.EOFException
>> at org.apache.kafka.common.network.NetworkReceive.readFromReada
>> bleChannel(NetworkReceive.java:83)
>> at kafka.network.BlockingChannel.readCompletely(BlockingChannel
>> .scala:129)
>> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:
>> 120)
>> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.s
>> cala:86)
>> at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$
>> $sendRequest(SimpleConsumer.scala:83)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>> y$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:132)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
>> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(
>> SimpleConsumer.scala:131)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
>> onsumer.scala:131)
>> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
>> onsumer.scala:131)
>> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:130)
>> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.fet
>> ch(DefaultFetchSimpleConsumer.scala:48)
>> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.def
>> aultFetch(DefaultFetchSimpleConsumer.scala:41)
>> at org.apache.samza.system.kafka.GetOffset.isValidOffset(GetOff
>> set.scala:60)
>> at org.apache.samza.system.kafka.BrokerProxy.addTopicPartition(
>> BrokerProxy.scala:99)
>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>> efreshBrokers$2.refresh$1(KafkaSystemConsumer.scala:213)
>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:226)
>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
>> efreshBrokers$2.apply(KafkaSystemConsumer.scala:192)
>> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
>> alSleepStrategy.scala:82)
>> at org.apache.samza.system.kafka.KafkaSystemConsumer.refreshBro
>> kers(KafkaSystemConsumer.scala:191)
>> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anon$1.ab
>> dicate(KafkaSystemConsumer.scala:293)
>> at org.apache.samza.system.kafka.BrokerProxy.abdicate(BrokerPro
>> xy.scala:207)
>> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
>> ors$2.apply(BrokerProxy.scala:245)
>> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
>> ors$2.apply(BrokerProxy.scala:245)
>> at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
>> at org.apache.samza.system.kafka.BrokerProxy.handleErrors(Broke
>> rProxy.scala:245)
>> at org.apache.samza.system.kafka.BrokerProxy.org
>> <http://org.apache.samza.system.kafka.brokerproxy.org/>$apache$samz
>> a$system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:186)
>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
>> un$1.apply(BrokerProxy.scala:147)
>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
>> un$1.apply(BrokerProxy.scala:134)
>> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
>> alSleepStrategy.scala:82)
>> at org.apache.samza.system.kafka.BrokerProxy$$anon$1.run(Broker
>> Proxy.scala:133)
>> at java.lang.Thread.run(Thread.java:745)
>>
>> WARN [2017-05-01 15:19:27,378] [U:1,904,F:1,153,T:3,056,M:3,056]
>> system.kafka.BrokerProxy:[Logging$class:warn:74] -
>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>> client samza_consumer-job4-1] - It appears that we received an invalid or
>> empty offset Some(45039137) for [Topic3,17]. Attempting to use Kafka's
>> auto.offset.reset setting. This can result in data loss if processing
>> continues.
>> INFO [2017-05-01 15:19:27,398] [U:1,904,F:1,152,T:3,056,M:3,056]
>> system.kafka.KafkaSystemConsumer:[Logging$class:info:63] -
>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform3:9092 for
>> client samza_consumer-job4-1] - Refreshing brokers for: Map([Topic3,17] ->
>> 45039137)
>> INFO [2017-05-01 15:19:27,405] [U:1,904,F:1,152,T:3,056,M:3,056]
>> system.kafka.GetOffset:[Logging$class:info:63] -
>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>> client samza_consumer-job4-1] - Checking if auto.offset.reset is defined
>> for topic Topic3
>> INFO [2017-05-01 15:19:27,432] [U:1,904,F:1,152,T:3,056,M:3,056]
>> system.kafka.GetOffset:[Logging$class:info:63] -
>> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
>> client samza_consumer-job4-1] - Got reset of type smallest.
>>
>> def addTopicPartition(tp: TopicAndPartition, nextOffset: Option[String]) = {
>> debug("Adding new topic and partition %s to queue for %s" format (tp, host))
>> if (nextOffsets.containsKey(tp)) {
>> toss("Already consuming TopicPartition %s" format tp)
>> }
>>
>> val offset = if (nextOffset.isDefined && offsetGetter.isValidOffset(simpleConsumer, tp, nextOffset.get)) {
>> nextOffset
>> .get
>> .toLong
>> } else {
>> warn("It appears that we received an invalid or empty offset %s for %s. Attempting to use Kafka's auto.offset.reset setting. This can result in data loss if processing continues." format (nextOffset, tp))
>> offsetGetter.getResetOffset(simpleConsumer, tp)
>> }
>>
>> debug("Got offset %s for new topic and partition %s." format (offset, tp))
>> nextOffsets += tp -> offset
>> metrics.topicPartitions(host, port).set(nextOffsets.size)
>> }
>>
>> def isValidOffset(consumer: DefaultFetchSimpleConsumer, topicAndPartition: TopicAndPartition, offset: String) = {
>> info("Validating offset %s for topic and partition %s" format (offset, topicAndPartition))
>>
>> try {
>> val messages = consumer.defaultFetch((topicAndPartition, offset.toLong))
>>
>> if (messages.hasError) {
>> KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic, topicAndPartition.partition))
>> }
>>
>> info("Able to successfully read from offset %s for topic and partition %s. Using it to instantiate consumer." format (offset, topicAndPartition))
>>
>> true
>> } catch {
>> case e: OffsetOutOfRangeException => false
>> }
>> }
>>
>>
>
Re: Possible bug causing reset in offsets [Version: 0.12.0].
Posted by Gaurav Agarwal <ga...@gmail.com>.
This also seems somewhat related to the mail on this group a few days back
- with subject 'Messages lost after broker failure'.
If someone had set auto.offset.reset to largest, then reverse would happen
- i.e samza skipping over kafka partition queue in face of such failures.
On Tue, May 2, 2017 at 9:17 AM, Gaurav Agarwal <ga...@gmail.com>
wrote:
> Hi All,
>
> We recently observed an issue in our Samza application (version
> 0.12.0) where we found that that the message offsets "jumped back" causing
> many of the messages to be re-processed.
>
> On digging deeper - here is what we found:
>
> - There was a some network related issues at that time causing temporary
> communication losses between nodes.
> - BrokerProxy tried to addTopicPartition() and while doing that, its
> isValidOffset() method failed.
> - This caused it to read the reset the offset using the configured
> setting - which we have set to auto.offset.reset=smallest
>
> Now, the validation check simply tries to fetch the message at the given
> offset and if there is an exception while doing it, it assumes that the
> offset is invalid. I think that the exception ReplicaNotAvailableException is
> being considered harmless, but I am not sure what is the expected behavior
> in case of any other exception. What about network related exceptions (that
> are transient)?
>
> I am pasting below the logs that show a *simple network error,* and
> immediately after that the message that Samza thinks that the offset was
> invalid and hence it is falling-back to reset.
>
> Is there any workaround to get past this problem? I would have thought
> that ideally, there could be only handful of well-known error codes that
> would indicate that the offset into a kafka topic/partition is invalid.
>
> --
> thanks,
> gaurav
>
>
> INFO [2017-05-01 15:19:27,327] [U:1,898,F:1,159,T:3,056,M:3,056]
> system.kafka.DefaultFetchSimpleConsumer:[Logging$class:info:76] -
> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> client samza_consumer-job4-1] - Reconnect due to error:
> java.io.EOFException
> at org.apache.kafka.common.network.NetworkReceive.readFromReada
> bleChannel(NetworkReceive.java:83)
> at kafka.network.BlockingChannel.readCompletely(BlockingChannel
> .scala:129)
> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:
> 120)
> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.s
> cala:86)
> at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$
> $sendRequest(SimpleConsumer.scala:83)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
> y$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:132)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$appl
> y$mcV$sp$1.apply(SimpleConsumer.scala:132)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(
> SimpleConsumer.scala:131)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
> onsumer.scala:131)
> at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleC
> onsumer.scala:131)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:130)
> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.fet
> ch(DefaultFetchSimpleConsumer.scala:48)
> at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.def
> aultFetch(DefaultFetchSimpleConsumer.scala:41)
> at org.apache.samza.system.kafka.GetOffset.isValidOffset(GetOff
> set.scala:60)
> at org.apache.samza.system.kafka.BrokerProxy.addTopicPartition(
> BrokerProxy.scala:99)
> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
> efreshBrokers$2.refresh$1(KafkaSystemConsumer.scala:213)
> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
> efreshBrokers$2.apply(KafkaSystemConsumer.scala:226)
> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anonfun$r
> efreshBrokers$2.apply(KafkaSystemConsumer.scala:192)
> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
> alSleepStrategy.scala:82)
> at org.apache.samza.system.kafka.KafkaSystemConsumer.refreshBro
> kers(KafkaSystemConsumer.scala:191)
> at org.apache.samza.system.kafka.KafkaSystemConsumer$$anon$1.ab
> dicate(KafkaSystemConsumer.scala:293)
> at org.apache.samza.system.kafka.BrokerProxy.abdicate(BrokerPro
> xy.scala:207)
> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
> ors$2.apply(BrokerProxy.scala:245)
> at org.apache.samza.system.kafka.BrokerProxy$$anonfun$handleErr
> ors$2.apply(BrokerProxy.scala:245)
> at scala.collection.mutable.HashSet.foreach(HashSet.scala:78)
> at org.apache.samza.system.kafka.BrokerProxy.handleErrors(Broke
> rProxy.scala:245)
> at org.apache.samza.system.kafka.BrokerProxy.org
> <http://org.apache.samza.system.kafka.brokerproxy.org/>$apache$
> samza$system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:186)
> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
> un$1.apply(BrokerProxy.scala:147)
> at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$r
> un$1.apply(BrokerProxy.scala:134)
> at org.apache.samza.util.ExponentialSleepStrategy.run(Exponenti
> alSleepStrategy.scala:82)
> at org.apache.samza.system.kafka.BrokerProxy$$anon$1.run(Broker
> Proxy.scala:133)
> at java.lang.Thread.run(Thread.java:745)
>
> WARN [2017-05-01 15:19:27,378] [U:1,904,F:1,153,T:3,056,M:3,056]
> system.kafka.BrokerProxy:[Logging$class:warn:74] -
> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> client samza_consumer-job4-1] - It appears that we received an invalid or
> empty offset Some(45039137) for [Topic3,17]. Attempting to use Kafka's
> auto.offset.reset setting. This can result in data loss if processing
> continues.
> INFO [2017-05-01 15:19:27,398] [U:1,904,F:1,152,T:3,056,M:3,056]
> system.kafka.KafkaSystemConsumer:[Logging$class:info:63] -
> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform3:9092 for
> client samza_consumer-job4-1] - Refreshing brokers for: Map([Topic3,17] ->
> 45039137)
> INFO [2017-05-01 15:19:27,405] [U:1,904,F:1,152,T:3,056,M:3,056]
> system.kafka.GetOffset:[Logging$class:info:63] -
> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> client samza_consumer-job4-1] - Checking if auto.offset.reset is defined
> for topic Topic3
> INFO [2017-05-01 15:19:27,432] [U:1,904,F:1,152,T:3,056,M:3,056]
> system.kafka.GetOffset:[Logging$class:info:63] -
> [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at platform1:9092 for
> client samza_consumer-job4-1] - Got reset of type smallest.
>
> def addTopicPartition(tp: TopicAndPartition, nextOffset: Option[String]) = {
> debug("Adding new topic and partition %s to queue for %s" format (tp, host))
> if (nextOffsets.containsKey(tp)) {
> toss("Already consuming TopicPartition %s" format tp)
> }
>
> val offset = if (nextOffset.isDefined && offsetGetter.isValidOffset(simpleConsumer, tp, nextOffset.get)) {
> nextOffset
> .get
> .toLong
> } else {
> warn("It appears that we received an invalid or empty offset %s for %s. Attempting to use Kafka's auto.offset.reset setting. This can result in data loss if processing continues." format (nextOffset, tp))
> offsetGetter.getResetOffset(simpleConsumer, tp)
> }
>
> debug("Got offset %s for new topic and partition %s." format (offset, tp))
> nextOffsets += tp -> offset
> metrics.topicPartitions(host, port).set(nextOffsets.size)
> }
>
> def isValidOffset(consumer: DefaultFetchSimpleConsumer, topicAndPartition: TopicAndPartition, offset: String) = {
> info("Validating offset %s for topic and partition %s" format (offset, topicAndPartition))
>
> try {
> val messages = consumer.defaultFetch((topicAndPartition, offset.toLong))
>
> if (messages.hasError) {
> KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic, topicAndPartition.partition))
> }
>
> info("Able to successfully read from offset %s for topic and partition %s. Using it to instantiate consumer." format (offset, topicAndPartition))
>
> true
> } catch {
> case e: OffsetOutOfRangeException => false
> }
> }
>
>