Kafka PubSub: Add ability to set `partitionNumber` in message metadata

[alicejgibbons]

Describe the feature

Kafka has a few per-call metadata fields already, for message headers and partition keys. As is done with partitionKey (key attribute), the request is to add partitionNumber (representing the exact partition number) as an option for Kafka message broker in order to be able to publish to a specific Kafka partition number.

This is useful for specific cases of high volume partitions when they need to be pinned to specific producers and consumers.

Release Note

RELEASE NOTE: ADD Kafka PubSub:partitionNumber in message metadata

[passuied]

@alicejgibbons I'm confused by this request. I don't think Kafka allows you to specify the partition number. You can only set the message key, which is currently set through partitionKey or __key publish metadata... Kafka will then balance the key to a partition in a deterministic way...
More over, the partition number is already available to the consumer in metadata as shown in the doc

Am I missing something?

[olitomlinson]

I agree with @passuied -- I don't think setting the partition id explicitly is a thing. Message keys get hashed to a partition Id, by kafka.

[alicejgibbons]

Looking at the constructor here the logic looks to be the following:

"If a valid partition number is specified that partition will be used when sending the record. If no partition is specified but a key is present a partition will be chosen using a hash of the key. If neither key nor partition is present a partition will be assigned in a round-robin fashion."

Thus, the Java SDK seems to accept either partition or Key (or combinations of both)? Will need to check go

[vikram-mahadevkar]

Hi @alicejgibbons,

Although Java Producer API does not explicitly provide constructor with only partitionId and value; key in all existing constructors is not mandatory.

From a Dapr feature implementation perspective, the following can be considered:

• Dapr publish API already supports per-call metadata key metadata.partitionKey. On the same lines we can support one more per-call metadata key named "metadata.partitionNumber"
• Application can decide to pass both/any-one/none based on need basis:
  • If only metadata.partitionNumber is passed then metadata.partitionNumber can be used as exact partition
  • If only metadata.partitionKey is passed then metadata.partitionKey can be used to derive partition using hash of key (same as current Dapr behaviour if metadata.partitionKey is passed)
  • If both are passed then, metadata.partitionNumber should be used as exact partition and metadata.partitionKey should be ignored
  • If both are not passed then partition can be assigned in round-robin fashion. (same as current Dapr behaviour in case metadata.partitionKey is not passed)

[alicejgibbons]

Thanks @vikram-mahadevkar. This looks doable using the current Kafka implementation in Go. See the example under "Manual partitioner" here: https://pkg.go.dev/github.com/Shopify/sarama#Partitioner

[passuied]

@alicejgibbons, you're right that you seem to be able to overridde the partition logic and you could implement such feature. I'm still uneasy about where that would be useful as we typically rely on the message key and ordering per partition but I guess there may be scenarios where you want full control at the producer level and ordering is not a huge concern? Also what happens if the partition # specified doesn't exist? Would it fallback to round robin or something else?

[alicejgibbons]

@passuied just to clarify, there would be no change to the default behaviour here. So today when a key is provided Partition = hash(key) % number_of_partitions and then when a key isn't provided a random partition is chosen (not round-robin by default in sarama).

This would be a new per-message metadata property that would follow the logic Vikram posted above, and if the partition number doesn't exist then agree it should fall back to default behaviour.

[passuied]

That was my understanding but thanks for confirming

[github-actions]

This issue has been automatically marked as stale because it has not had activity in the last 30 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue, help wanted or triaged/resolved) or other activity occurs. Thank you for your contributions.

[alicejgibbons]

not stale, putting in 1.17