-
Notifications
You must be signed in to change notification settings - Fork 177
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
out_rdkafka2: add patch for new version of rdkafka
Signed-off-by: Thomas Tych <thomas.tych@gmail.com>
- Loading branch information
Showing
5 changed files
with
96 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,15 @@ | ||
class Rdkafka::Producer | ||
# return false if producer is forcefully closed, otherwise return true | ||
def close(timeout = nil) | ||
@closing = true | ||
# Wait for the polling thread to finish up | ||
# If the broker isn't alive, the thread doesn't exit | ||
if timeout | ||
thr = @polling_thread.join(timeout) | ||
return !!thr | ||
else | ||
@polling_thread.join | ||
return true | ||
end | ||
end | ||
end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,27 @@ | ||
# This is required for `rdkafka` version >= 0.12.0 | ||
# Overriding the close method in order to provide a time limit for when it should be forcibly closed | ||
class Rdkafka::Producer::Client | ||
# return false if producer is forcefully closed, otherwise return true | ||
def close(timeout=nil) | ||
return unless @native | ||
|
||
# Indicate to polling thread that we're closing | ||
@polling_thread[:closing] = true | ||
# Wait for the polling thread to finish up | ||
thread = @polling_thread.join(timeout) | ||
|
||
Rdkafka::Bindings.rd_kafka_destroy(@native) | ||
|
||
@native = nil | ||
|
||
return !thread.nil? | ||
end | ||
end | ||
|
||
class Rdkafka::Producer | ||
def close(timeout = nil) | ||
ObjectSpace.undefine_finalizer(self) | ||
|
||
return @client.close(timeout) | ||
end | ||
end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,44 @@ | ||
class Rdkafka::NativeKafka | ||
# return false if producer is forcefully closed, otherwise return true | ||
def close(timeout=nil, object_id=nil) | ||
return true if closed? | ||
|
||
synchronize do | ||
# Indicate to the outside world that we are closing | ||
@closing = true | ||
|
||
thread_status = :unknown | ||
if @polling_thread | ||
# Indicate to polling thread that we're closing | ||
@polling_thread[:closing] = true | ||
|
||
# Wait for the polling thread to finish up, | ||
# this can be aborted in practice if this | ||
# code runs from a finalizer. | ||
thread_status = @polling_thread.join(timeout) | ||
end | ||
|
||
# Destroy the client after locking both mutexes | ||
@poll_mutex.lock | ||
|
||
# This check prevents a race condition, where we would enter the close in two threads | ||
# and after unlocking the primary one that hold the lock but finished, ours would be unlocked | ||
# and would continue to run, trying to destroy inner twice | ||
if @inner | ||
Rdkafka::Bindings.rd_kafka_destroy(@inner) | ||
@inner = nil | ||
@opaque = nil | ||
end | ||
|
||
!thread_status.nil? | ||
end | ||
end | ||
end | ||
|
||
class Rdkafka::Producer | ||
def close(timeout = nil) | ||
return true if closed? | ||
ObjectSpace.undefine_finalizer(self) | ||
@native_kafka.close(timeout) | ||
end | ||
end |