KEDA 2.3 kafka_scaler error "invalid offset found for topic"

[carmeloacosta]

There is some issue/bug in the kafka_scaler that makes it get lost regarding some Kafta topic partition offset. Once this situation happens, an error (one for each affected partition) is shown in the keda-operator logs :

INFO kafka_scaler invalid offset found for topic our_topic in group our_Group and partition 71, probably no offset is committed yet

Once we reach this situation, any “affected partition” will no trigger any new message, regardless the amount of new messages incoming to that specific partition. But if we sent them to a “non-affected” partition, KEDA keeps on working fine. This situation is unpredictable, at least for us, and we do not know when will a partition start being affected in this way. When we destroy the kafka topic, and create again it fresh-new, all partitions are unaffected. But, as time goes by, the number of "affected partitions" keeps on growing. Losing the events history, when recreating the topic, is harmful for us, so it is not a real option.

Regarding the KEDA configuration, this is a snippet of our ScaledJobs:

> API Version:  keda.sh/v1alpha1
>  Kind:         ScaledJob
>  Metadata:
>    Finalizers:
>      finalizer.keda.sh
>    Generation:        2
>    Resource Version:  103335517
>  Spec:                                      
>    Failed Jobs History Limit:  100
>    Job Target Ref:                      
>      Active Deadline Seconds:  86400
>      Backoff Limit:            0              
>      Parallelism:              1               
>    Max Replica Count:     100           
>    Polling Interval:      10                  
>    Scaling Strategy:                         
>      Strategy:                     accurate 
>    Successful Jobs History Limit:  100
>    Triggers:                                       
>      Authentication Ref:                      
>        Name:  keda-trigger-auth-kafka-credential
>      Metadata:                                               
>        Allow Idle Consumers:  true                    
>        Bootstrap Servers:     <our-own-instance-in.confluent.cloud:9092>
>        Consumer Group:       <our-group>                                             
>        Lag Threshold:         1                                                               
>        Offset Reset Policy:   latest                                                       
>        Topic:                 <our-topic>                                                     
>      Type:                    kafka

Due to our business logic, we cannot use Offset Reset Policy 'earliest' since we need to quickly respond to the new events, without going through all the previous ones.

[zroubalik]

@grassiale WDYT?

[zroubalik]

Can't you initialize initialize offsets (for all consumer groups) before starting to consume the messages, ie. scaling the jobs?

[carmeloacosta]

In fact, another "interesting" issue is that we need to initialize each new Kafka topic, sending and consuming a few dummy Kafka messages, before being able to use that new Kafka topic with KEDA. Otherwise, KEDA won't trigger any new scaledjob regardless of the amount of new messages that arrives to the Kafka topic set as trigger.

[grassiale]

There is some issue/bug in the kafka_scaler that makes it get lost regarding some Kafta topic partition offset. Once this situation happens, an error (one for each affected partition) is shown in the keda-operator logs :

INFO kafka_scaler invalid offset found for topic our_topic in group our_Group and partition 71, probably no offset is committed yet

Once we reach this situation, any “affected partition” will no trigger any new message, regardless the amount of new messages incoming to that specific partition. But if we sent them to a “non-affected” partition, KEDA keeps on working fine. This situation is unpredictable, at least for us, and we do not know when will a partition start being affected in this way. When we destroy the kafka topic, and create again it fresh-new, all partitions are unaffected. But, as time goes by, the number of "affected partitions" keeps on growing. Losing the events history, when recreating the topic, is harmful for us, so it is not a real option.

Regarding the KEDA configuration, this is a snippet of our ScaledJobs:

> API Version:  keda.sh/v1alpha1
>  Kind:         ScaledJob
>  Metadata:
>    Finalizers:
>      finalizer.keda.sh
>    Generation:        2
>    Resource Version:  103335517
>  Spec:                                      
>    Failed Jobs History Limit:  100
>    Job Target Ref:                      
>      Active Deadline Seconds:  86400
>      Backoff Limit:            0              
>      Parallelism:              1               
>    Max Replica Count:     100           
>    Polling Interval:      10                  
>    Scaling Strategy:                         
>      Strategy:                     accurate 
>    Successful Jobs History Limit:  100
>    Triggers:                                       
>      Authentication Ref:                      
>        Name:  keda-trigger-auth-kafka-credential
>      Metadata:                                               
>        Allow Idle Consumers:  true                    
>        Bootstrap Servers:     <our-own-instance-in.confluent.cloud:9092>
>        Consumer Group:       <our-group>                                             
>        Lag Threshold:         1                                                               
>        Offset Reset Policy:   latest                                                       
>        Topic:                 <our-topic>                                                     
>      Type:                    kafka

Due to our business logic, we cannot use Offset Reset Policy 'earliest' since we need to quickly respond to the new events, without going through all the previous ones.

Hello!

Do you have a low message rate?
Is your consumer scaled to 0 and only starts when new messages arrive? Does it happen frequently that a partition loses its committed offsets?

I'm asking because for what I know you will only lose committed offsets when a consumer is not connected to Kafka for a certain amount of time. One week by default.

It never occurred to me that offsets were missing for some partitions only and not all of them.

[carmeloacosta]

Hello!

The message rate is relatively low, about 1 message every few hours, although it is expected to increase.
No, it is frequent having a partition in a "desynchronized state" (i.e, with an "invalid offset found for topic" log). In fact, after weeks of production use only a few partitions, out of a 100 partition topic, got into that state. We initially created the 100-partition topic with KEDA 2.2 (maxReplicaCount : 100) and now, with KEDA 2.3 and allowIdleConsumers we are planning to reduce the amount of partitions. We are not sure if the amount of partitions, combined with the "relatively" low message rate could be producing this negative effect.