Why does `FutureRecord` require me to put the key in an `async` block or specify the type in `rdkafka`?

I'm trying to create a FutureProducer that would send a simple message to the topic, but cannot do so without specifying the key apparently: the checker complains:

rdkafka::producer::future_producer::FutureRecord
pub fn payload(self, payload: &'a P) -> FutureRecord<'a, K, P>
Sets the destination payload of the record.

type inside `async` block must be known in this context
cannot infer type for type parameter `K`rustcE0698

Furthermore,when i do specify the key, they arrive mangled in the topic and i don't understand why. I think there is one too many things going on here, can you help me understand?

    async {
        loop {
            let msg = consumer.recv().await.unwrap();
            println!("Offset :{:?}", msg.offset());
            let dest_topic = "compression_data_control";
            let offset     = msg.offset().to_le_bytes();    // msg.offset is i64, but doesn't have `ToBytes` which `FutureRecord` requires apparently.
            println!("encoded the offset as {:?}", i64::from_le_bytes(offset));
            let r = FutureRecord::to(dest_topic)
                .payload(msg.payload().unwrap())
                .key(&offset);             // <---- if i don't specify this
                                           //       i get the above error
                   

            let future = producer.send(r, Timeout::After(Duration::from_millis(15000)));
            future.await.map_or(-1, |_| {println!("Processed offset {:?} successfully.", &offset); 0});

        }
    }.await;

When i do it this way, the keys arrive in the topic like so: ,�X. What am i doing wrong?