Allow specifying a create time for messages

lib/kafka/producer.rb

@@ -178,12 +178,11 @@ def initialize(cluster:, logger:, instrumenter:, compressor:, ack_timeout:, requ
     # @param topic [String] the topic that the message should be written to.
     # @param partition [Integer] the partition that the message should be written to.
     # @param partition_key [String] the key that should be used to assign a partition.
+    # @param create_time [Time] the timestamp that should be set on the message.
     #
     # @raise [BufferOverflow] if the maximum buffer size has been reached.
     # @return [nil]
-    def produce(value, key: nil, topic:, partition: nil, partition_key: nil)
-      create_time = Time.now
-
+    def produce(value, key: nil, topic:, partition: nil, partition_key: nil, create_time: Time.now)
       message = PendingMessage.new(
         value && value.to_s,
         key && key.to_s,

spec/functional/producer_spec.rb

@@ -7,6 +7,17 @@
 
   let!(:topic) { create_random_topic(num_partitions: 3) }
 
+  example "setting a create_time value" do
+    timestamp = Time.now
+
+    producer.produce("hello", topic: topic, partition: 0, create_time: timestamp)
+    producer.deliver_messages
+
+    message = kafka.fetch_messages(topic: topic, partition: 0, offset: :earliest).last
+
+    expect(message.create_time.to_i).to eq timestamp.to_i
+  end
+
   example "writing messages using the buffered producer" do
     value1 = rand(10_000).to_s
     value2 = rand(10_000).to_s