[exporter/kafka_exporter]: Add topic-based partitioning support for grouping same topics on the same partition

[karakayasemi]

Description

When TopicFromAttribute is configured, messages with the same topic attribute value are grouped on the same partition for correct topic routing.

Link to tracking issue

Fixes #37470

Testing

• Added topic_partitioning test cases to both logs and metrics partitioning tests
• Tests verify that the same topic attributes get the same partition keys, different topics get different keys

Documentation

• Created a changelog entry documenting the enhancement for end users

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: karakayasemi / name: Semih Serhat Karakaya (b197464)

[axw]

Thanks for the PR @karakayasemi. Adding the topic to the hash sounds good, but I think it needs to be additive rather than replacing the resource attributes hash.

[axw]

This doesn't seem quite right: this will completely ignore the resource attributes, as you're overwriting hash. Can we combine them?

How about this?

m := resourceLogs.Resource().Attributes()
if e.config.TopicFromAttribute != "" {
	if rv, ok := resourceLogs.Resource().Attributes().Get(e.config.TopicFromAttribute); ok && rv.Str() != "" {
		m2 := pcommon.NewMap()
		m2.PutStr("topic", rv.Str())
		m.CopyTo(m2.PutEmptyMap("resource"))
		m = m2
	}
}
hash := pdatautil.MapHash(m)

[karakayasemi]

Hey @axw, thank you for the review. I simplified the logic mainly to resolve the wrong-topic issue.

When partitioning is enabled, the topic is already included in the resource attributes, so the hash will naturally differ when the topic value differs. So, I decided not to touch the current partitioning logic.

When partitioning is disabled, the problem mentioned in the issue occurs. Everything goes within a single batch and with a single topic, but in that case, we still need to at least separate by topic. That’s the behavior I am aiming for in the new implementation. Could you please review one more time?

[axw]

@karakayasemi sorry, I think I was a bit confused before and probably read the line above your addition as if e.config.PartitionLogsByResourceAttributes rather than if !...

Reading it again, I'm not sure why setting topic_from_attribute should control partitioning? They should be separate concepts. If you're just trying to fix #37470, I think we can skip the hashing and just yield one result per ResourceX.

I'll leave a new comment to show what I mean.

[axw]

Suggested change

	hash := pdatautil.ValueHash(pcommon.NewValueStr(""))
	if rv, ok := resourceLogs.Resource().Attributes().Get(e.config.TopicFromAttribute); ok && rv.Str() != "" {
	hash = pdatautil.ValueHash(pcommon.NewValueStr(rv.Str()))
	}
	newLogs := plog.NewLogs()
	resourceLogs.CopyTo(newLogs.ResourceLogs().AppendEmpty())
	if !yield(hash[:], newLogs) {
	return
	}
	newLogs := plog.NewLogs()
	resourceLogs.CopyTo(newLogs.ResourceLogs().AppendEmpty())
	if !yield(nil, newLogs) {
	return
	}

I think this is all that's needed? i.e. rather than yielding the original ld, yield one per ResourceLogs within it.

[github-actions]

This PR was marked stale due to lack of activity. It will be closed in 14 days.