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

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

Merged
merged 7 commits into from
Jul 20, 2020
+53 −20
Merged

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

Changes from 1 commit
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 = latest
)

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["offsetReset"])
grassiale marked this conversation as resolved.
Show resolved Hide resolved
if policy != earliest && policy != latest {
return meta, fmt.Errorf("err offsetReset policy %s given", policy)
grassiale marked this conversation as resolved.
Show resolved Hide resolved
}
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