automatic offset reset question

[Richard Rossel <he...@gmail.com>]

Hi,
I have a stream process that from time to time starts re-ingesting old data
which affects the whole pipeline.
What I think is happening is the consumer group gets reset to the earliest,
but since:
  * the consumer group name is always the same
  * Replica set is 3
  * auto.offset.reset is 'none'
Then, I'm not sure what is the reason for this behaviour.

through log I found out a broker wasn't available, which was the consumer
coordinator:

INFO  AbstractCoordinator:657 (Group coordinator kafka-bkr-0.data:9092 (id:
2147483647 rack: null) is unavailable or invalid, will attempt rediscovery)

ERROR ConsumerCoordinator:748 - Offset commit failed on partition
p011060-syslog0-0 at offset 5499432626: This is not the correct coordinator.

ERROR ConsumerCoordinator:748 Offset commit failed on partition
p011060-syslog0-4 at offset 5498978605: The coordinator is loading and
hence can't process requests.

 ERROR ConsumerCoordinator:748 Offset commit failed on partition
p011060-syslog0-5 at offset 5499236332: Specified group generation id is
not valid.

WARN  AssignedStreamsTasks:202 stream task 0_5 got migrated to another
thread already. Closing it as zombie.

Then the whole stream crashes with:

Caused by: org.apache.kafka.clients.consumer.CommitFailedException: Commit
cannot be completed since the group has already rebalanced and assigned the
partitions to another member. This means that the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms, which
typically implies that the poll loop is spending too much time message
processing. You can address this either by increasing the session timeout
or by reducing the maximum size of batches returned in poll() with
max.poll.records.


Next time that application starts, the consumer will start reading from the
earliest offset.

I have been trying to replicate this scenario in a dev environment,
shutting down brokers, but everything works as it should be, not offsets
get reset.

Any ideas on what's going on or how can I replicate this scenario?
I thought that using auto.offset.reset = none It will avoid any offset
reset, why doesn't it work?

Setting:
 16 brokers, all running 2.3.0, replica 3 on all topics.
Some of the Consumer config:
  auto.commit.interval.ms = 5000
  auto.offset.reset = none
  enable.auto.commit = false
  connections.max.idle.ms = 540000

Thanks.-



-- 
Richard Rossel
Atlanta - GA

Re: automatic offset reset question

[Ezequiel Puig <es...@gmail.com>]

Hello,

Can you check the offsets.retention.minutes paramether


From the documentation (
https://kafka.apache.org/documentation/#upgrade_200_notable) :


   - KIP-186 <https://cwiki.apache.org/confluence/x/oYtjB> increases the
   default offset retention time from 1 day to 7 days. This makes it less
   likely to "lose" offsets in an application that commits infrequently. It
   also increases the active set of offsets and therefore can increase memory
   usage on the broker. Note that the console consumer currently enables
   offset commit by default and can be the source of a large number of offsets
   which this change will now preserve for 7 days instead of 1. You can
   preserve the existing behavior by setting the broker config
   offsets.retention.minutes to 1440.


Best regards,
Ezequiel PUIG


Le ven. 15 janv. 2021 à 23:02, Richard Rossel <he...@gmail.com> a
écrit :

> Hi,
> I have a stream process that from time to time starts re-ingesting old data
> which affects the whole pipeline.
> What I think is happening is the consumer group gets reset to the earliest,
> but since:
>   * the consumer group name is always the same
>   * Replica set is 3
>   * auto.offset.reset is 'none'
> Then, I'm not sure what is the reason for this behaviour.
>
> through log I found out a broker wasn't available, which was the consumer
> coordinator:
>
> INFO  AbstractCoordinator:657 (Group coordinator kafka-bkr-0.data:9092 (id:
> 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery)
>
> ERROR ConsumerCoordinator:748 - Offset commit failed on partition
> p011060-syslog0-0 at offset 5499432626: This is not the correct
> coordinator.
>
> ERROR ConsumerCoordinator:748 Offset commit failed on partition
> p011060-syslog0-4 at offset 5498978605: The coordinator is loading and
> hence can't process requests.
>
>  ERROR ConsumerCoordinator:748 Offset commit failed on partition
> p011060-syslog0-5 at offset 5499236332: Specified group generation id is
> not valid.
>
> WARN  AssignedStreamsTasks:202 stream task 0_5 got migrated to another
> thread already. Closing it as zombie.
>
> Then the whole stream crashes with:
>
> Caused by: org.apache.kafka.clients.consumer.CommitFailedException: Commit
> cannot be completed since the group has already rebalanced and assigned the
> partitions to another member. This means that the time between subsequent
> calls to poll() was longer than the configured max.poll.interval.ms, which
> typically implies that the poll loop is spending too much time message
> processing. You can address this either by increasing the session timeout
> or by reducing the maximum size of batches returned in poll() with
> max.poll.records.
>
>
> Next time that application starts, the consumer will start reading from the
> earliest offset.
>
> I have been trying to replicate this scenario in a dev environment,
> shutting down brokers, but everything works as it should be, not offsets
> get reset.
>
> Any ideas on what's going on or how can I replicate this scenario?
> I thought that using auto.offset.reset = none It will avoid any offset
> reset, why doesn't it work?
>
> Setting:
>  16 brokers, all running 2.3.0, replica 3 on all topics.
> Some of the Consumer config:
>   auto.commit.interval.ms = 5000
>   auto.offset.reset = none
>   enable.auto.commit = false
>   connections.max.idle.ms = 540000
>
> Thanks.-
>
>
>
> --
> Richard Rossel
> Atlanta - GA
>