switch to using rd_kafka_message_t directly