[kafka_confluent] should propagate partitionkey on an event level

[bread-amammay]

Issue

Currently you have to interact with

sdk-go/protocol/kafka_confluent/v2/option.go

Lines 137 to 151 in 7fef294

	// WithMessageKey returns back a new context with the given messageKey.
	func WithMessageKey(ctx context.Context, messageKey string) context.Context {
	return context.WithValue(ctx, keyForMessageKey, messageKey)
	}
	// MessageKeyFrom looks in the given context and returns `messageKey` as a string if found and valid, otherwise "".
	func MessageKeyFrom(ctx context.Context) string {
	c := ctx.Value(keyForMessageKey)
	if c != nil {
	if s, ok := c.(string); ok {
	return s
	}
	}
	return ""
	}

to set the partition key on an outgoing event. If you have multiple events are you sending on a single context you have to keep swapping values.

Proposal

I propose that the paradigm from kafka_sarama is copied over where it checks to see if the partition key is set on an a cloud event extension, if it is, set that on the outgoing record.

sdk-go/protocol/kafka_sarama/v2/write_producer_message.go

Lines 38 to 48 in f1c0d0a

	transformers = append(transformers, binding.TransformerFunc(func(r binding.MessageMetadataReader, w binding.MessageMetadataWriter) error {
	ext := r.GetExtension(partitionKey)
	if !types.IsZero(ext) {
	extStr, err := types.Format(ext)
	if err != nil {
	return err
	}
	key = extStr
	}
	return nil
	}))

• tagging @yanmxa for thoughts since i seen you wrote the implementation of Confluent kafka binding #988

[yanmxa]

Thanks @bread-amammay!
I see that both implementations use the same way to inject the partitionKey for the Sender. The difference is that the sarma implementation also adds partitionKey by event extension. But what if the key exists in both context and extension?

I agree with the idea of supporting the partitionKey at the event level. But we need to document their priorities.

Do you want to have a try? :)