Optimize experimental Kafka scaler and fix consumer group logic

[adrien-f]

TL;DR:

• Only request metadata on the topics needed
• Fix the logic to fetch topics from the consumer groups

---

I know we decided to implement the IAM auth in the sarama based scaler (the original) but there is still an optimization that would serve current users of the experimental scaler and potentially lower their Keda load and their broker.

Firstly, the Metadata request is not scoped for the requested topics (which can be empty):

keda/pkg/scalers/apache_kafka_scaler.go

Lines 425 to 427 in bcaf5c0

	metadata, err := s.client.Metadata(ctx, &kafka.MetadataRequest{
	Addr: s.client.Addr,
	})

If the list of topics is empty, we enter the branch to detect the topics & permissions based on the consumer group activity here:

https://github.com/kedacore/keda/blob/bcaf5c07e785e3e58e3be4e3707b518fdef6acde/pkg/scalers/apache_kafka_scaler.go#L441C3-L441C14

On my AWS MSK cluster, the version for the response is not supported by the segment-io library which causes the following to error out, as seen in segmentio/kafka-go#1212 (probably):

Got non-zero number of bytes remaining: 10

Alright, for the case of debugging, let's ignore the error. What happens next? From what I can see, this variable describeGrp is unused, and because the MetadataRequest is empty, the whole list of topics is processed and returned with their partitions. All the time. Which then cause Keda to request all consumer & producer offsets again and again, etc...

So to me, currently the behavior is buggy. The describeGrp should be processed to extract the list of topics and partitions for the consumer group which I also correct in this PR.

Checklist

• When introducing a new scaler, I agree with the scaling governance policy
• I have verified that my change is according to the deprecations & breaking changes policy
• Tests have been added
• Changelog has been updated and is aligned with our changelog requirements
• A PR is opened to update our Helm chart (repo) (if applicable, ie. when deployment manifests are modified)
• A PR is opened to update the documentation on (repo) (if applicable)
• Commits are signed with Developer Certificate of Origin (DCO - learn more)

Fixes #

Relates to #5531

[dttung2905]

Hi @adrien-f ,
Thanks for refactoring this piece of code. Generally its LGTM. Just a couple of small comments for me
Since there is a new helper function getTopicPartitionsFromConsumerGroup created, would that be possible to create a unit test for it ?

[dttung2905]

If this is a known error, can we put a link to the issue in the comment so that we can easily trace it back in the future ?

[adrien-f]

Done 👍

[adrien-f]

@dttung2905 thanks you for the review! I will look into adding unit tests, the issue being that this lib does not have an interface to create mock on. I will investigate and see what's possible to do.