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.

Sign up for GitHub

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

Jump to bottom

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

Merged
merged 7 commits into from
Jul 20, 2020
Merged

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

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b0432a9
adds consumer offset reset policy option to keda kafka scaler
grassiale Jul 10, 2020
c1967c8
fixing metadata label that was lost in rename
grassiale Jul 10, 2020
6de5df5
changes default offset reset policy to be coherent with current behavior
grassiale Jul 10, 2020
81f5605
adds metadata parse test for consumer offset reset parameter
grassiale Jul 10, 2020
0187474
uses offsetResetPolicy type in test data
grassiale Jul 12, 2020
f8a4648
renames offset reset policy property and sets default to latest
grassiale Jul 13, 2020
81246de
changes default outcome for offsetResetPolicy in tests
grassiale Jul 13, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 29 additions & 8 deletions pkg/scalers/kafka_scaler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,11 @@ type kafkaScaler struct {
}

type kafkaMetadata struct {
bootstrapServers []string
group string
topic string
lagThreshold int64
bootstrapServers []string
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 ;)


// auth
authMode kafkaAuthMode
Expand All @@ -42,6 +43,13 @@ type kafkaMetadata struct {
ca string
}

type offsetResetPolicy string

const (
latest offsetResetPolicy = "latest"
earliest offsetResetPolicy = "earliest"
)

type kafkaAuthMode string

const (
Expand All @@ -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.

)

var kafkaLog = logf.Log.WithName("kafka_scaler")
Expand Down Expand Up @@ -100,6 +109,16 @@ func parseKafkaMetadata(resolvedEnv, metadata, authParams map[string]string) (ka
}
meta.topic = metadata["topic"]

meta.consumerOffsetReset = defaultOffsetReset

if metadata["consumerOffsetReset"] != "" {
policy := offsetResetPolicy(metadata["consumerOffsetReset"])
if policy != earliest && policy != latest {
return meta, fmt.Errorf("err consumerOffsetReset policy %s given", policy)
}
meta.consumerOffsetReset = policy
}

meta.lagThreshold = defaultKafkaLagThreshold

if val, ok := metadata[lagThresholdMetricName]; ok {
Expand Down Expand Up @@ -295,11 +314,13 @@ func (s *kafkaScaler) getLagForPartition(partition int32, offsets *sarama.Offset
}

var lag int64
// For now, assume a consumer group that has no committed
// offset will read all messages from the topic. This may be
// something we want to allow users to configure.

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

} else {
lag = latestOffset
}
} else {
lag = latestOffset - consumerOffset
}
Expand Down
36 changes: 24 additions & 12 deletions pkg/scalers/kafka_scaler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,13 @@ import (
)

type parseKafkaMetadataTestData struct {
metadata map[string]string
isError bool
numBrokers int
brokers []string
group string
topic string
metadata map[string]string
isError bool
numBrokers int
brokers []string
group string
topic string
consumerOffsetReset offsetResetPolicy
}

// A complete valid metadata example for reference
Expand All @@ -33,16 +34,21 @@ var validWithoutAuthParams = map[string]string{}

var parseKafkaMetadataTestDataset = []parseKafkaMetadataTestData{
// failure, no bootstrapServers
{map[string]string{}, true, 0, nil, "", ""},

{map[string]string{}, true, 0, nil, "", "", ""},
// failure, no consumer group
{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", ""},
{map[string]string{"bootstrapServers": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", "", "earliest"},
// failure, no topic
{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, true, 1, []string{"foobar:9092"}, "my-group", ""},
{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group"}, true, 1, []string{"foobar:9092"}, "my-group", "", offsetResetPolicy("earliest")},
// success
{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic"},
{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 1, []string{"foobar:9092"}, "my-group", "my-topic", offsetResetPolicy("earliest")},
// success, more brokers
{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic"},
{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", offsetResetPolicy("earliest")},
// success, consumerOffsetReset policy earliest
{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "consumerOffsetReset": "earliest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", offsetResetPolicy("earliest")},
// failure, consumerOffsetReset policy wrong
{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "consumerOffsetReset": "foo"}, true, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", ""},
// success, consumerOffsetReset policy latest
{map[string]string{"bootstrapServers": "foo:9092,bar:9092", "consumerGroup": "my-group", "topic": "my-topic", "consumerOffsetReset": "latest"}, false, 2, []string{"foo:9092", "bar:9092"}, "my-group", "my-topic", offsetResetPolicy("latest")},
}

func TestGetBrokers(t *testing.T) {
Expand All @@ -67,6 +73,9 @@ func TestGetBrokers(t *testing.T) {
if meta.topic != testData.topic {
t.Errorf("Expected topic %s but got %s\n", testData.topic, meta.topic)
}
if err == nil && meta.consumerOffsetReset != testData.consumerOffsetReset {
t.Errorf("Expected consumerOffsetReset %s but got %s\n", testData.consumerOffsetReset, meta.consumerOffsetReset)
}

meta, err = parseKafkaMetadata(nil, testData.metadata, validWithoutAuthParams)

Expand All @@ -88,5 +97,8 @@ func TestGetBrokers(t *testing.T) {
if meta.topic != testData.topic {
t.Errorf("Expected topic %s but got %s\n", testData.topic, meta.topic)
}
if err == nil && meta.consumerOffsetReset != testData.consumerOffsetReset {
t.Errorf("Expected consumerOffsetReset %s but got %s\n", testData.consumerOffsetReset, meta.consumerOffsetReset)
}
}
}