Skip to content

Commit

Permalink
Allow specifying a create time for messages
Browse files Browse the repository at this point in the history
  • Loading branch information
dasch committed Nov 10, 2017
1 parent 6b720d1 commit a6b5404
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 3 deletions.
5 changes: 2 additions & 3 deletions lib/kafka/producer.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
11 changes: 11 additions & 0 deletions spec/functional/producer_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down

0 comments on commit a6b5404

Please sign in to comment.