There is a known issue that Kafka may return a wrong offset even if there
is no reset happening: https://issues.apache.org/jira/browse/KAFKA-7703
Best Regards,
Ryan
On Tue, Apr 30, 2019 at 10:41 AM Austin Weaver <aus...@flyrlabs.com> wrote:
> @deng - There was a short erroneous period where 2 streams were reading
> from the same topic and group id were running at the same time. We saw
> errors in this and stopped the extra stream. That being said, I would think
> regardless that the auto.offset.reset would kick in sine documentation says
> that it will kick in if there is no existing current offset or that the
> current offset no longer exists on the kafka topic? Moreover, that doesn't
> explain the fact that the spark logs that it is on one offset for that
> partition (5553330) - and then immediately errors out trying to read the
> old offset (4544296) that no longer exists?
>
> @Akshay - I am using Spark Streaming (D-streams) Here is a snippet of the
> kafka consumer configuration I am using (redacted some fields) -
>
> kakaConsumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
> kakaConsumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "");
> kakaConsumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
> kakaConsumerProperties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
> RoundRobinAssignor.class.getName());
> kakaConsumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> StringDeserializer.class);
> kakaConsumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> StringDeserializer.class);
> kakaConsumerProperties.put("auto.offset.reset", "earliest");
> kakaConsumerProperties.put("sasl.mechanism", "PLAIN");
> kakaConsumerProperties.put("sasl.jaas.config", "security.protocol");
> kakaConsumerProperties.put("security.protocol", "");
>
> and I'm using LocationStrategies.PreferConsistent()
>
> Thanks
>
> On Tue, Apr 30, 2019 at 5:56 AM Akshay Bhardwaj <
> akshay.bhardwaj1...@gmail.com> wrote:
>
>> Hi Austin,
>>
>> Are you using Spark Streaming or Structured Streaming?
>>
>> For better understanding, could you also provide sample code/config
>> params for your spark-kafka connector for the said streaming job?
>>
>>
>> Akshay Bhardwaj
>> +91-97111-33849
>>
>>
>> On Mon, Apr 29, 2019 at 10:34 PM Austin Weaver <aus...@flyrlabs.com>
>> wrote:
>>
>>> Hey guys, relatively new Spark Dev here and i'm seeing some kafka offset
>>> issues and was wondering if you guys could help me out.
>>>
>>> I am currently running a spark job on Dataproc and am getting errors
>>> trying to re-join a group and read data from a kafka topic. I have done
>>> some digging and am not sure what the issue is. I have auto.offset.reset set
>>> to earliest so it should being reading from the earliest available
>>> non-committed offset and initially my spark logs look like this :
>>>
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-11 to offset 5553330.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-2 to offset 5555553.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-3 to offset 5555484.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-4 to offset 5555586.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-5 to offset 5555502.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-6 to offset 5555561.
>>> 19/04/29 16:30:30 INFO
>>> org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
>>> clientId=consumer-1, groupId=demo-group] Resetting offset for
>>> partition demo.topic-7 to offset 5555542.```
>>>
>>> But then the very next line I get an error trying to read from a
>>> nonexistent offset on the server (you can see that the offset for the
>>> partition differs from the one listed above, so I have no idea why it would
>>> be attempting to read form that offset, here is the error on the next line:
>>>
>>> org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets
>>> out of range with no configured reset policy for partitions:
>>> {demo.topic-11=4544296}
>>>
>>> Any ideas to why my spark job is constantly going back to this offset
>>> (4544296), and not the one it outputs originally (5553330)?
>>>
>>> It seems to be contradicting itself w a) the actual offset it says its
>>> on and the one it attempts to read and b) saying no configured reset policy
>>> --
>>> Austin Weaver
>>> Software Engineer
>>> FLYR, Inc. www.flyrlabs.com
>>>
>>
>
> --
> Austin Weaver
> Software Engineer
> FLYR, Inc. www.flyrlabs.com
>
