Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[v2]adds consumer offset reset policy option to keda kafka scaler #925

Merged

Conversation

grassiale
Copy link
Contributor

@grassiale grassiale commented Jul 9, 2020

Signed-off-by: grassiale [email protected]

This PR is aimed at solving how Keda Kafka scaler handles the absence of committed offset with relation to the offset policy the consumer is following.
At the moment, if you create a scaledobject for a new consumer group having no committed offset, keda scaler will return a lag value equal to the latest offset in the topic it is consuming from. This often causes the scaler to scale relative deployment replicas to the max, because the latest offset in the topic is probably higher than the lagTreshold. This is fine for consumers having auto.offset.reset=earliest, since they will start to consume right-away all the messages in the topic; but is wrong for
auto.offset.reset=latest consumers, that will only consume new messages on the topic. The lag for this kind of consumer is now initially set to 0.

Checklist

@zroubalik
Copy link
Member

@grassiale thanks, this is great. Could you please open this PR against v2 branch? Most likely we won't release any new v1 release , so v2 is the right place for this change.

@grassiale grassiale changed the base branch from master to v2 July 10, 2020 09:37
@grassiale grassiale force-pushed the feature/kafka-scaler-offset-reset-policy branch from ce2b760 to 472a3b7 Compare July 10, 2020 09:42
@grassiale grassiale force-pushed the feature/kafka-scaler-offset-reset-policy branch from 472a3b7 to b0432a9 Compare July 10, 2020 09:53
pkg/scalers/kafka_scaler.go Outdated Show resolved Hide resolved
pkg/scalers/kafka_scaler.go Outdated Show resolved Hide resolved
if consumerOffset == sarama.OffsetNewest || consumerOffset == sarama.OffsetOldest {
lag = latestOffset
if s.metadata.consumerOffsetReset == latest {
lag = 0
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shouldn't this be lag = latestOffset ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when you have no offset committed and you create a new consumer with reset latest policy then the lag should be 0 since the consumer is aligned with the latest offset on the topic.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I am not a Kafka expert, but the properties naming is pretty confusing

@zroubalik
Copy link
Member

@ppatierno PTAL if you have a time :)

@zroubalik
Copy link
Member

@grassiale could you please add a test that is covering this property?

@grassiale grassiale marked this pull request as draft July 10, 2020 15:13
@grassiale
Copy link
Contributor Author

Converting to draft because I'm implementing tests.

@@ -57,6 +65,7 @@ const (
lagThresholdMetricName = "lagThreshold"
kafkaMetricType = "External"
defaultKafkaLagThreshold = 10
defaultOffsetReset = earliest
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in Kafka the default is "latest", why are you setting "earliest" out of curiosity?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, you are right on the default in kafka. I set latest as a default because, right now, if you create a kafka scaler, it will behave as if we set earliest, if no offset is committed. So I would not break the current behavior for anyone upgrading from a previous version and not seeing the option.
But maybe, since this is a new major version of Keda, it could be safer to be coherent with Kafka defaults.
Let me know what you think, I can change it accordingly to the decision.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes I would agree, let's see what @zroubalik thinks.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree, it makes sense to set it to Kafka default, a new major release is a great fit for a change like this. We could add a small note to the docs about it.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've done the change and also changed the docs, let me know if it needs further clarification there.

brokers []string
group string
topic string
consumerOffsetReset string
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would use the custom type offsetResetPolicy instead of string here.

@grassiale grassiale changed the title adds consumer offset reset policy option to keda kafka scaler [v2]adds consumer offset reset policy option to keda kafka scaler Jul 12, 2020
group string
topic string
lagThreshold int64
consumerOffsetReset offsetResetPolicy
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can't we find a better name instead of consumerOffsetReset. Actually the type name would be a great name for the struct field as well so offsetResetPolicy. It's not the first time that a field name is the same as the type name, or?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Absolutely ;)

Copy link
Contributor

@ppatierno ppatierno left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM other than the comments I left :-)

@zroubalik
Copy link
Member

LGTM other than the comments I left :-)

@ppatierno thanks for the review!

@grassiale grassiale force-pushed the feature/kafka-scaler-offset-reset-policy branch from 83f9c9b to f8a4648 Compare July 13, 2020 09:35
@grassiale
Copy link
Contributor Author

grassiale commented Jul 13, 2020

I also started to add a small E2E test for this particular case, let me know if you don't think this is necessary.

@zroubalik
Copy link
Member

@grassiale e2e tests are highly appreciated. So would be really great if you could add a scenario for this, but it is not mandatory. But if you want to test functionality that is not that much relevant (or is not direct affected by) KEDA but it is more testing Kafka capability, you don't have to add it.

@grassiale
Copy link
Contributor Author

Hi @zroubalik , I've been able to implement some e2e tests for this case, the thing is:

  • the current consumer being used in e2e tests is not committing offsets, and this is a problem since lag for keda in "latest" configuration will always be 0.
  • I had to tweak some other stuff in the tests in order for the tests I've added to work.
    So maybe this is a bit out of scope for this PR. Shall we go on and finalize this pr and maybe I shall open an issue for this?

@zroubalik
Copy link
Member

  So maybe this is a bit out of scope for this PR. Shall we go on and finalize this pr and maybe I shall open an issue for this?

Agree, makes sense.

@grassiale grassiale marked this pull request as ready for review July 16, 2020 10:38
@zroubalik zroubalik merged commit e107f26 into kedacore:v2 Jul 20, 2020
zroubalik pushed a commit that referenced this pull request Aug 6, 2020
* adds consumer offset reset policy option to keda kafka scaler
Signed-off-by: grassiale <[email protected]>
@ahmelsayed ahmelsayed mentioned this pull request Aug 11, 2020
3 tasks
SpiritZhou pushed a commit to SpiritZhou/keda that referenced this pull request Jul 18, 2023
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

3 participants