Make ConsumerRecord to be copyable

Author: kenneth-jia

include/kafka/ConsumerRecord.h

@@ -87,8 +87,8 @@ class ConsumerRecord
     std::string toString() const;
 
 private:
-    using rd_kafka_message_unique_ptr = std::unique_ptr<rd_kafka_message_t, void(*)(rd_kafka_message_t*)>;
-    rd_kafka_message_unique_ptr _rk_msg;
+    using rd_kafka_message_shared_ptr = std::shared_ptr<rd_kafka_message_t>;
+    rd_kafka_message_shared_ptr _rk_msg;
 };
 
 inline Headers

include/kafka/ProducerRecord.h

@@ -13,13 +13,13 @@ namespace KAFKA_API::clients::producer {
 /**
  * A key/value pair to be sent to Kafka.
  * This consists of a topic name to which the record is being sent, an optional partition number, and an optional key and value.
+ * Note: `ProducerRecord` would not take the ownership from the memory block of `Value`.
  */
 class ProducerRecord
 {
 public:
     using Id  = std::uint64_t;
 
-    // Note: ProducerRecord would not take the ownership from these parameters,
     ProducerRecord(Topic topic, Partition partition, const Key& key, const Value& value)
        : _topic(std::move(topic)), _partition(partition), _key(key), _value(value) {}
 

tests/integration/TestKafkaConsumer.cc

@@ -60,6 +60,12 @@ TEST(KafkaConsumer, BasicPoll)
     records = KafkaTestUtility::ConsumeMessagesUntilTimeout(consumer);
     EXPECT_EQ(messages.size(), records.size());
 
+    // Copyable ConsumerRecord
+    {
+        auto recordsCopy = records;
+        recordsCopy.clear();
+    }
+
     // Check messages
     std::size_t rcvMsgCount = 0;
     for (auto& record: records)