[v2] Kafka scaler isactive latest #996

grassiale · 2020-08-15T08:15:35Z

Hello, while implementing the e2e tests requested here: #985
I've come across a new issue that is not making the scaler behave correctly when the offset reset policy is set to latest (default).

When a new consumer is created, meaning it has no offsets committed, the sarama offset fetch will return a value of -1 for all the partitions of the topic it consumes from. If we set the lag to 0, as in the previous solution, the isActive method will always return false resulting in the scaling to never occur even if new messages are written in the topic.

In order to fix this, I propose the following:

return sarama.OffsetNewest (-1) when the new consumer with latest policy has no committed offset, for every partition.
This will result to have a negative lag at consumer creation.
The isActive method changes the logic in returning true so that if the offset policy is latest, the negative lag would make the scaling active. Scaling the deployment to the minimum number of pods.
When new messages arrive the running consumer pod will consume these messages, committing offsets, so that it is now able to scale to 0 and the logic in getLagForPartition will now return the correct lag offset when a new message is written in the topic.

With that in mind I also changed E2E tests so that:

They have a consumer that actually consumes from the topic, the previous twitter-function didn't.
They consider the earliest offset reset policy.
They test the behaviour introduced with the fix.

Checklist

A PR is opened to update the documentation on https://github.com/kedacore/keda-docs (probably not needed, bugfix)
Commits are signed with Developer Certificate of Origin (DCO)
Tests have been added

Fixes #

…olicy Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

… missing, wip. Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

zroubalik · 2020-08-15T13:11:00Z

@ppatierno PTAL ^

zroubalik · 2020-08-21T07:45:08Z

FYI @tbickford this is the Kafka issue we were talking about

zroubalik · 2020-08-31T09:41:34Z

@grassiale any update on this please?

grassiale · 2020-08-31T16:40:38Z

@grassiale any update on this please?

Hi.
Sorry, I've been away from keyboard, and still be for a couple of days. I'll be able to complete it then.

zroubalik · 2020-08-31T18:54:14Z

OK, thanks

Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

zroubalik

LGTM, but would like to have an ack from some Kafka guru 😄

ppatierno · 2020-09-04T07:38:42Z

pkg/scalers/kafka_scaler.go

@@ -317,7 +317,7 @@ func (s *kafkaScaler) getLagForPartition(partition int32, offsets *sarama.Offset

 	if consumerOffset == sarama.OffsetNewest || consumerOffset == sarama.OffsetOldest {
 		if s.metadata.offsetResetPolicy == latest {
-			lag = 0
+			lag = sarama.OffsetNewest


I don't know the sarama library but there is something that sounds strange to me first of all in the first if statement:

if consumerOffset == sarama.OffsetNewest || consumerOffset == sarama.OffsetOldest

When a consumer is new in a consumer group there is no offset committed of course. Why the sarama fetch offset should return OffsetNewest or OffsetOldest? It could return the one or the other if it knew the reset policy asked by the consumer (so returning OffsetNewest if latest and OffsetOldest if earliest) but it doesn't know, right?
It's not actually clear to me the sarama behaviour here tbh.

It is actually not Sarama deciding to return -1, but the kafka coordinator here: https://github.com/apache/kafka/blob/8bde3d476f23a9d9dd06b3360281a3a3fca2804e/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L481
From what I see, if no consumer group exists, or no offset is committed, the coordinator will return the value of OffsetFetchResponse.INVALID_OFFSET that is -1:
https://github.com/apache/kafka/blob/6da70f9b95d2f7cc1f32de4e405661a8015bfa7e/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java#L53

Maybe we could introduce the concept of invalid offset instead of using offsetNewest and Oldest that are related to the consumer and not the admin client?

Absolutely agree ... this -1 value has a different meaning them not related to what Sarama returns, so it's better having it separated.

Do you think we should return an error in getLagForPartition if the offset is invalid? This would help identify the case of missing offsets, hence having a more significant condition also for the isActive method.

absolutely yes, I don't like the if condition based on sarama newest and oldest + lag > 0 or lag != 0. An error should simplify and make it clearer.

Done. I didn't know whether to log an error or an info message with a low verbosity level, since this is not a fatal error as, for example, the other two in the getLagForPartition method.

…fka coordinator is invalid (-1) Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

ppatierno · 2020-09-04T11:46:34Z

pkg/scalers/kafka_scaler.go

+		lag, err := s.getLagForPartition(partition, offsets)
+		if err != nil {
+			return true, nil
+		}


only concern is, is any error really ok to saying isActive = true? Maybe we should check the exact error, in this case the invalid offset one?

Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

zroubalik · 2020-09-04T13:07:02Z

@grassiale @ppatierno thanks a lot!

…gy (kedacore#996)

grassiale added 2 commits August 15, 2020 10:24

changes kafka scaler isactive behavior on the basis of offset reset p…

bc25b9f

…olicy Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

kafka scaler e2e test now consider offset reset policies. One test is…

6eac8bb

… missing, wip. Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

grassiale force-pushed the kafka-scaler-isactive-latest branch from 552cafb to 6eac8bb Compare August 15, 2020 08:24

zroubalik changed the title ~~Kafka scaler isactive latest~~ [v2] Kafka scaler isactive latest Aug 19, 2020

zroubalik requested review from ahmelsayed and samuelmacko August 19, 2020 14:43

samuelmacko approved these changes Aug 27, 2020

View reviewed changes

grassiale marked this pull request as ready for review September 3, 2020 07:59

grassiale requested a review from zroubalik as a code owner September 3, 2020 07:59

grassiale force-pushed the kafka-scaler-isactive-latest branch from 67598d7 to 50d8b1f Compare September 3, 2020 08:02

final version of e2e test for kafka scaler latest offset policy

31fdbe2

Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

grassiale force-pushed the kafka-scaler-isactive-latest branch from 50d8b1f to 31fdbe2 Compare September 3, 2020 08:03

trims trailing whitespaces and adds eof line

af82bfd

Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

zroubalik approved these changes Sep 3, 2020

View reviewed changes

ppatierno reviewed Sep 4, 2020

View reviewed changes

getLagForPartition now returns an error if the offset fetched from ka…

dc3289b

…fka coordinator is invalid (-1) Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

ppatierno reviewed Sep 4, 2020

View reviewed changes

isactive returns true if error is not null and offset is invalid

f779ede

Signed-off-by: grassiale <alessandro.grassi01@gmail.com>

ppatierno approved these changes Sep 4, 2020

View reviewed changes

zroubalik merged commit 50c81f0 into kedacore:v2 Sep 4, 2020

PaulLiang1 mentioned this pull request Mar 1, 2022

Kafka scaler reports -2 and scale to maxReplicas #2612

Closed

SpiritZhou pushed a commit to SpiritZhou/keda that referenced this pull request Jul 18, 2023

docs(eventhub): provide dapr parameter for eventhub checkpoint strate…

f7e563c

…gy (kedacore#996)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[v2] Kafka scaler isactive latest #996

[v2] Kafka scaler isactive latest #996

grassiale commented Aug 15, 2020 •

edited

Loading

zroubalik commented Aug 15, 2020

zroubalik commented Aug 21, 2020

zroubalik commented Aug 31, 2020

grassiale commented Aug 31, 2020

zroubalik commented Aug 31, 2020

zroubalik left a comment

ppatierno Sep 4, 2020

grassiale Sep 4, 2020 •

edited

Loading

ppatierno Sep 4, 2020

grassiale Sep 4, 2020 •

edited

Loading

ppatierno Sep 4, 2020

grassiale Sep 4, 2020 •

edited

Loading

ppatierno Sep 4, 2020

grassiale Sep 4, 2020

zroubalik commented Sep 4, 2020

[v2] Kafka scaler isactive latest #996

[v2] Kafka scaler isactive latest #996

Conversation

grassiale commented Aug 15, 2020 • edited Loading

Checklist

zroubalik commented Aug 15, 2020

zroubalik commented Aug 21, 2020

zroubalik commented Aug 31, 2020

grassiale commented Aug 31, 2020

zroubalik commented Aug 31, 2020

zroubalik left a comment

Choose a reason for hiding this comment

ppatierno Sep 4, 2020

Choose a reason for hiding this comment

grassiale Sep 4, 2020 • edited Loading

Choose a reason for hiding this comment

ppatierno Sep 4, 2020

Choose a reason for hiding this comment

grassiale Sep 4, 2020 • edited Loading

Choose a reason for hiding this comment

ppatierno Sep 4, 2020

Choose a reason for hiding this comment

grassiale Sep 4, 2020 • edited Loading

Choose a reason for hiding this comment

ppatierno Sep 4, 2020

Choose a reason for hiding this comment

grassiale Sep 4, 2020

Choose a reason for hiding this comment

zroubalik commented Sep 4, 2020

grassiale commented Aug 15, 2020 •

edited

Loading

grassiale Sep 4, 2020 •

edited

Loading

grassiale Sep 4, 2020 •

edited

Loading

grassiale Sep 4, 2020 •

edited

Loading