Skip to content

Various AsyncProducer improvements #855

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Aug 4, 2020
Merged

Various AsyncProducer improvements #855

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5f22fbd
Create a unique `Mutex` for each `AsyncProducer`
cstyles Jul 28, 2020
142a93c
Only synchronize once in `ensure_threads_running!`
cstyles Jul 28, 2020
4b6c560
Factor out `worker_thread_alive?` and `timer_thread_alive?` methods
cstyles Jul 28, 2020
95ea335
Return early in `ensure_threads_running!` if there's nothing to do
cstyles Jul 28, 2020
2205415
Simplify the code a bit inside `ensure_threads_running!`
cstyles Jul 28, 2020
2d26a53
Ensure threads are running when calling `deliver_messages`
cstyles Jul 28, 2020
d9f9363
Ensure threads are running when calling `shutdown`
cstyles Jul 28, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 18 additions & 9 deletions lib/kafka/async_producer.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,6 @@ module Kafka
# producer.shutdown
#
class AsyncProducer
THREAD_MUTEX = Mutex.new

# Initializes a new AsyncProducer.
#
# @param sync_producer [Kafka::Producer] the synchronous producer that should
Expand Down Expand Up @@ -94,6 +92,8 @@ def initialize(sync_producer:, max_queue_size: 1000, delivery_threshold: 0, deli

# The timer will no-op if the delivery interval is zero.
@timer = Timer.new(queue: @queue, interval: delivery_interval)

@thread_mutex = Mutex.new
end

# Produces a message to the specified topic.
Expand Down Expand Up @@ -131,6 +131,8 @@ def produce(value, topic:, **options)
# @see Kafka::Producer#deliver_messages
# @return [nil]
def deliver_messages
ensure_threads_running!

@queue << [:deliver_messages, nil]

nil
Expand All @@ -142,6 +144,8 @@ def deliver_messages
# @see Kafka::Producer#shutdown
# @return [nil]
def shutdown
ensure_threads_running!

@timer_thread && @timer_thread.exit
@queue << [:shutdown, nil]
@worker_thread && @worker_thread.join
Expand All @@ -152,17 +156,22 @@ def shutdown
private

def ensure_threads_running!
THREAD_MUTEX.synchronize do
@worker_thread = nil unless @worker_thread && @worker_thread.alive?
@worker_thread ||= Thread.new { @worker.run }
end
return if worker_thread_alive? && timer_thread_alive?

THREAD_MUTEX.synchronize do
@timer_thread = nil unless @timer_thread && @timer_thread.alive?
@timer_thread ||= Thread.new { @timer.run }
@thread_mutex.synchronize do
@worker_thread = Thread.new { @worker.run } unless worker_thread_alive?
@timer_thread = Thread.new { @timer.run } unless timer_thread_alive?
end
end

def worker_thread_alive?
!!@worker_thread && @worker_thread.alive?
end

def timer_thread_alive?
!!@timer_thread && @timer_thread.alive?
end

def buffer_overflow(topic, message)
@instrumenter.instrument("buffer_overflow.async_producer", {
topic: topic,
Expand Down