kafkamdm: make kafka version configurable #446
Conversation
|
nice. looks good. you have tested this, i presume? |
|
Hi, i did not test against a kafka cluster but i checked if the config gets applied and what happens if i do not configure the version or configure a invalid version. Invalid version or unconfigured = The warning might be misleading due to the sarama error message "invalid version" maybe i should adjust it / remove the sarama error message. |
| @@ -218,6 +218,7 @@ topic | Y | string | N/A | | |||
| codec | Y | string | N/A | which compression to use. possible values: none, gzip, snappy | |||
| partitionBy | Y | string | N/A | which fields to shard by. possible values are: byOrg, bySeries, bySeriesWithTags | |||
| schemasFile | Y | string | N/A | | |||
| kafkaVersion | N | string | "" | Kafka version in semver format. All brokers must be this version or newer. (fallback to oldest stable version 1.0.0) | |||
There was a problem hiding this comment.
go.mod pins sarama to v1.23.0
That version's ParseKafkaVersion function returns V0_8_2_0 if no version could be parsed
| @@ -96,11 +95,19 @@ func NewKafkaMdm(key string, matcher matcher.Matcher, topic, codec, schemasFile, | |||
| log.Fatalf("kafkaMdm %q: failed to initialize partitioner. %s", r.key, err) | |||
| } | |||
|
|
|||
| kfkVersion, err := sarama.ParseKafkaVersion(kafkaVersion) | |||
There was a problem hiding this comment.
defaulting to an empty string and showing an error if the user doesn't specify a version, seems messy.
Why don't we default to sarama.MinVersion.String() ?
|
Georg Doser seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
This PR will fix #444
This implementation will add kafkaVersion as option to the kafkamdm route. If no version got specified carbon-relay-ng will print a warning and use sarama.DefaultVersion.
This means that users can now specify the kafka version but if they don't do it carbon-relay-ng will act as in the current implementation.
ParseKafkaVersion will return the same default value on error as sarama.NewConfig() would:
https://github.com/Shopify/sarama/blob/v1.23.0/utils.go#L196
https://github.com/Shopify/sarama/blob/v1.23.0/config.go#L439
We might discuss if the message regarding invalid version should be a warn or info.