Kafka produce with partition?

[scrogson]

Maybe I'm missing something, but I cannot see how to specify which partition to produce to for a kafka producer. send_to only seems to work with a StreamKey (topic) and something which implements Buffer. Am I missing something?

[tyt2y3]

Hi, welcome. There is an option to allow clients to specify a strategy https://kafka.apache.org/documentation/#consumerconfigs_partition.assignment.strategy

Edit: I was wrong. Forget what I said.

[scrogson]

In Kafka land, producers do not control which partition a message goes to, the broker (and hence the operations team) does.

While this may be true, it is also possible to specify a partition on the client when producing to Kafka: https://docs.rs/rdkafka/0.36.0/rdkafka/producer/struct.BaseRecord.html (also in FutureRecord).

In my experience with Erlang/Elixir, I’ve been doing this with a custom hashing strategy on the client. I hash the key based on the the number of partitions and this gives me the partition to produce the message to.

Having said that, I wasn’t aware you could have the broker do this, so thank you for pointing that out.

Even if the broker picks the partition, I believe you still need to set a key so that it can pick a consistent partition for you, otherwise it would be round robin (which I usually never want).

Even with all this, I'm still left puzzled on how you can set the key (StreamKey seems to be the topic unless I'm reading incorrectly). Also, how does one set message headers? If the payload is protobuf bytes for instance, it might be nice to set a content-type header...or propogate tracing headers.

I apologize if I'm completely missing something, but from what I have read (including the code), I cannot seem to understand how to accomplish this with the S: Buffer trait bounds.

[scrogson]

@tyt2y3 ok, good to know. I would be interested in contributing this but would need some consultation on the API.

What do you think of something simple like:

impl KafkaProducer {
    // existing functionality

    /// Send a record to a stream
    fn send_record<K, P>(
        &self,
        stream: &StreamKey,
        record: rdkafka::producer::FutureRecord<K, P>,
    ) -> KafkaResult<SendFuture>
    where
        K: rdkafka::message::ToBytes + ?Sized,
        P: rdkafka::message::ToBytes + ?Sized,
    {
        let fut = self
            .get()
            .send_result(record)
            .map_err(|(err, _raw)| stream_err(err))?;

        Ok(SendFuture {
            stream_key: Some(stream.to_owned()),
            fut,
        })
    }
}

This seems to compile.

[scrogson]

I don't think the StreamKey needs to be passed in if we can extract it from the record.topic.

[scrogson]

    /// Send a record to a stream
    fn send_record<K, P>(
        &self,
        record: rdkafka::producer::FutureRecord<K, P>,
    ) -> KafkaResult<SendFuture>
    where
        K: rdkafka::message::ToBytes + ?Sized,
        P: rdkafka::message::ToBytes + ?Sized,
    {
        let stream = StreamKey::new(record.topic.to_owned())?;

        let fut = self
            .get()
            .send_result(record)
            .map_err(|(err, _raw)| stream_err(err))?;

        Ok(SendFuture {
            stream_key: Some(stream),
            fut,
        })
    }

[scrogson]

#17