add raising flag on wait in handler (#346)

Author: mensfeld

CHANGELOG.md

@@ -1,140 +1,141 @@
 # Rdkafka Changelog
 
 ## 0.14.0 (Unreleased)
+- [Enhancement] Add `raise_response_error` flag to the `Rdkafka::AbstractHandle`.
 - [Enhancement] Allow for setting `statistics_callback` as nil to reset predefined settings configured by a different gem (mensfeld)
-* [Enhancement] Get consumer position (thijsc & mensfeld)
-* [Enhancement] Provide `#purge` to remove any outstanding requests from the producer (mensfeld)
-* [Enhancement] Update `librdkafka` to `2.2.0` (mensfeld)
-* [Enhancement] Introduce producer partitions count metadata cache (mensfeld)
-* [Enhancement] Increase metadata timeout request from `250 ms` to `2000 ms` default to allow for remote cluster operations via `rdkafka-ruby` (mensfeld)
-* [Enhancement] Introduce `#name` for producers and consumers (mensfeld)
-* [Enhancement] Include backtrace in non-raised binded errors (mensfeld)
-* [Fix] Reference to Opaque is not released when Admin, Consumer or Producer is closed (mensfeld)
-* [Fix] Trigger `#poll` on native kafka creation to handle oauthbearer cb (mensfeld)
-* [Fix] `#flush` does not handle the timeouts errors by making it return `true` if all flushed or `false` if failed. We do **not** raise an exception here to keep it backwards compatible (mensfeld)
-* [Change] Remove support for Ruby 2.6 due to it being EOL and WeakMap incompatibilities (mensfeld)
-* [Change] Update Kafka Docker with Confluent KRaft (mensfeld)
-* [Change] Update librdkafka repo reference from edenhill to confluentinc (mensfeld)
+- [Enhancement] Get consumer position (thijsc & mensfeld)
+- [Enhancement] Provide `#purge` to remove any outstanding requests from the producer (mensfeld)
+- [Enhancement] Update `librdkafka` to `2.2.0` (mensfeld)
+- [Enhancement] Introduce producer partitions count metadata cache (mensfeld)
+- [Enhancement] Increase metadata timeout request from `250 ms` to `2000 ms` default to allow for remote cluster operations via `rdkafka-ruby` (mensfeld)
+- [Enhancement] Introduce `#name` for producers and consumers (mensfeld)
+- [Enhancement] Include backtrace in non-raised binded errors (mensfeld)
+- [Fix] Reference to Opaque is not released when Admin, Consumer or Producer is closed (mensfeld)
+- [Fix] Trigger `#poll` on native kafka creation to handle oauthbearer cb (mensfeld)
+- [Fix] `#flush` does not handle the timeouts errors by making it return `true` if all flushed or `false` if failed. We do **not** raise an exception here to keep it backwards compatible (mensfeld)
+- [Change] Remove support for Ruby 2.6 due to it being EOL and WeakMap incompatibilities (mensfeld)
+- [Change] Update Kafka Docker with Confluent KRaft (mensfeld)
+- [Change] Update librdkafka repo reference from edenhill to confluentinc (mensfeld)
 
 ## 0.13.0 (2023-07-24)
-* Support cooperative sticky partition assignment in the rebalance callback (methodmissing)
-* Support both string and symbol header keys (ColinDKelley)
-* Handle tombstone messages properly (kgalieva)
-* Add topic name to delivery report (maeve)
-* Allow string partitioner config (mollyegibson)
-* Fix documented type for DeliveryReport#error (jimmydo)
-* Bump librdkafka to 2.0.2 (lmaia)
-* Use finalizers to cleanly exit producer and admin (thijsc)
-* Lock access to the native kafka client (thijsc)
-* Fix potential race condition in multi-threaded producer (mensfeld)
-* Fix leaking FFI resources in specs (mensfeld)
-* Improve specs stability (mensfeld)
-* Make metadata request timeout configurable (mensfeld)
-* call_on_partitions_assigned and call_on_partitions_revoked only get a tpl passed in (thijsc)
+- Support cooperative sticky partition assignment in the rebalance callback (methodmissing)
+- Support both string and symbol header keys (ColinDKelley)
+- Handle tombstone messages properly (kgalieva)
+- Add topic name to delivery report (maeve)
+- Allow string partitioner config (mollyegibson)
+- Fix documented type for DeliveryReport#error (jimmydo)
+- Bump librdkafka to 2.0.2 (lmaia)
+- Use finalizers to cleanly exit producer and admin (thijsc)
+- Lock access to the native kafka client (thijsc)
+- Fix potential race condition in multi-threaded producer (mensfeld)
+- Fix leaking FFI resources in specs (mensfeld)
+- Improve specs stability (mensfeld)
+- Make metadata request timeout configurable (mensfeld)
+- call_on_partitions_assigned and call_on_partitions_revoked only get a tpl passed in (thijsc)
 
 ## 0.12.0 (2022-06-17)
-* Bumps librdkafka to 1.9.0
-* Fix crash on empty partition key (mensfeld)
-* Pass the delivery handle to the callback (gvisokinskas)
+- Bumps librdkafka to 1.9.0
+- Fix crash on empty partition key (mensfeld)
+- Pass the delivery handle to the callback (gvisokinskas)
 
 ## 0.11.0 (2021-11-17)
-* Upgrade librdkafka to 1.8.2
-* Bump supported minimum Ruby version to 2.6
-* Better homebrew path detection
+- Upgrade librdkafka to 1.8.2
+- Bump supported minimum Ruby version to 2.6
+- Better homebrew path detection
 
 ## 0.10.0 (2021-09-07)
-* Upgrade librdkafka to 1.5.0
-* Add error callback config
+- Upgrade librdkafka to 1.5.0
+- Add error callback config
 
 ## 0.9.0 (2021-06-23)
-* Fixes for Ruby 3.0
-* Allow any callable object for callbacks (gremerritt)
-* Reduce memory allocations in Rdkafka::Producer#produce (jturkel)
-* Use queue as log callback to avoid unsafe calls from trap context (breunigs)
-* Allow passing in topic configuration on create_topic (dezka)
-* Add each_batch method to consumer (mgrosso)
+- Fixes for Ruby 3.0
+- Allow any callable object for callbacks (gremerritt)
+- Reduce memory allocations in Rdkafka::Producer#produce (jturkel)
+- Use queue as log callback to avoid unsafe calls from trap context (breunigs)
+- Allow passing in topic configuration on create_topic (dezka)
+- Add each_batch method to consumer (mgrosso)
 
 ## 0.8.1 (2020-12-07)
-* Fix topic_flag behaviour and add tests for Metadata (geoff2k)
-* Add topic admin interface (geoff2k)
-* Raise an exception if @native_kafka is nil (geoff2k)
-* Option to use zstd compression (jasonmartens)
+- Fix topic_flag behaviour and add tests for Metadata (geoff2k)
+- Add topic admin interface (geoff2k)
+- Raise an exception if @native_kafka is nil (geoff2k)
+- Option to use zstd compression (jasonmartens)
 
 ## 0.8.0 (2020-06-02)
-* Upgrade librdkafka to 1.4.0
-* Integrate librdkafka metadata API and add partition_key (by Adithya-copart)
-* Ruby 2.7 compatibility fix (by Geoff Thé)A
-* Add error to delivery report (by Alex Stanovsky)
-* Don't override CPPFLAGS and LDFLAGS if already set on Mac (by Hiroshi Hatake)
-* Allow use of Rake 13.x and up (by Tomasz Pajor)
+- Upgrade librdkafka to 1.4.0
+- Integrate librdkafka metadata API and add partition_key (by Adithya-copart)
+- Ruby 2.7 compatibility fix (by Geoff Thé)A
+- Add error to delivery report (by Alex Stanovsky)
+- Don't override CPPFLAGS and LDFLAGS if already set on Mac (by Hiroshi Hatake)
+- Allow use of Rake 13.x and up (by Tomasz Pajor)
 
 ## 0.7.0 (2019-09-21)
-* Bump librdkafka to 1.2.0 (by rob-as)
-* Allow customizing the wait time for delivery report availability (by mensfeld)
+- Bump librdkafka to 1.2.0 (by rob-as)
+- Allow customizing the wait time for delivery report availability (by mensfeld)
 
 ## 0.6.0 (2019-07-23)
-* Bump librdkafka to 1.1.0 (by Chris Gaffney)
-* Implement seek (by breunigs)
+- Bump librdkafka to 1.1.0 (by Chris Gaffney)
+- Implement seek (by breunigs)
 
 ## 0.5.0 (2019-04-11)
-* Bump librdkafka to 1.0.0 (by breunigs)
-* Add cluster and member information (by dmexe)
-* Support message headers for consumer & producer (by dmexe)
-* Add consumer rebalance listener (by dmexe)
-* Implement pause/resume partitions (by dmexe)
+- Bump librdkafka to 1.0.0 (by breunigs)
+- Add cluster and member information (by dmexe)
+- Support message headers for consumer & producer (by dmexe)
+- Add consumer rebalance listener (by dmexe)
+- Implement pause/resume partitions (by dmexe)
 
 ## 0.4.2 (2019-01-12)
-* Delivery callback for producer
-* Document list param of commit method
-* Use default Homebrew openssl location if present
-* Consumer lag handles empty topics
-* End iteration in consumer when it is closed
-* Add support for storing message offsets
-* Add missing runtime dependency to rake
+- Delivery callback for producer
+- Document list param of commit method
+- Use default Homebrew openssl location if present
+- Consumer lag handles empty topics
+- End iteration in consumer when it is closed
+- Add support for storing message offsets
+- Add missing runtime dependency to rake
 
 ## 0.4.1 (2018-10-19)
-* Bump librdkafka to 0.11.6
+- Bump librdkafka to 0.11.6
 
 ## 0.4.0 (2018-09-24)
-* Improvements in librdkafka archive download
-* Add global statistics callback
-* Use Time for timestamps, potentially breaking change if you
+- Improvements in librdkafka archive download
+- Add global statistics callback
+- Use Time for timestamps, potentially breaking change if you
   rely on the previous behavior where it returns an integer with
   the number of milliseconds.
-* Bump librdkafka to 0.11.5
-* Implement TopicPartitionList in Ruby so we don't have to keep
+- Bump librdkafka to 0.11.5
+- Implement TopicPartitionList in Ruby so we don't have to keep
   track of native objects.
-* Support committing a topic partition list
-* Add consumer assignment method
+- Support committing a topic partition list
+- Add consumer assignment method
 
 ## 0.3.5 (2018-01-17)
-* Fix crash when not waiting for delivery handles
-* Run specs on Ruby 2.5
+- Fix crash when not waiting for delivery handles
+- Run specs on Ruby 2.5
 
 ## 0.3.4 (2017-12-05)
-* Bump librdkafka to 0.11.3
+- Bump librdkafka to 0.11.3
 
 ## 0.3.3 (2017-10-27)
-* Fix bug that prevent display of `RdkafkaError` message
+- Fix bug that prevent display of `RdkafkaError` message
 
 ## 0.3.2 (2017-10-25)
-* `add_topic` now supports using a partition count
-* Add way to make errors clearer with an extra message
-* Show topics in subscribe error message
-* Show partition and topic in query watermark offsets error message
+- `add_topic` now supports using a partition count
+- Add way to make errors clearer with an extra message
+- Show topics in subscribe error message
+- Show partition and topic in query watermark offsets error message
 
 ## 0.3.1 (2017-10-23)
-* Bump librdkafka to 0.11.1
-* Officially support ranges in `add_topic` for topic partition list.
-* Add consumer lag calculator
+- Bump librdkafka to 0.11.1
+- Officially support ranges in `add_topic` for topic partition list.
+- Add consumer lag calculator
 
 ## 0.3.0 (2017-10-17)
-* Move both add topic methods to one `add_topic` in `TopicPartitionList`
-* Add committed offsets to consumer
-* Add query watermark offset to consumer
+- Move both add topic methods to one `add_topic` in `TopicPartitionList`
+- Add committed offsets to consumer
+- Add query watermark offset to consumer
 
 ## 0.2.0 (2017-10-13)
-* Some refactoring and add inline documentation
+- Some refactoring and add inline documentation
 
 ## 0.1.x (2017-09-10)
-* Initial working version including producing and consuming
+- Initial working version including producing and consuming

lib/rdkafka/abstract_handle.rb

@@ -48,12 +48,13 @@ def pending?
     #   If this is nil it does not time out.
     # @param wait_timeout [Numeric] Amount of time we should wait before we recheck if the
     #   operation has completed
+    # @param raise_response_error [Boolean] should we raise error when waiting finishes
     #
     # @return [Object] Operation-specific result
     #
     # @raise [RdkafkaError] When the operation failed
     # @raise [WaitTimeoutError] When the timeout has been reached and the handle is still pending
-    def wait(max_wait_timeout: 60, wait_timeout: 0.1)
+    def wait(max_wait_timeout: 60, wait_timeout: 0.1, raise_response_error: true)
       timeout = if max_wait_timeout
                   monotonic_now + max_wait_timeout
                 else
@@ -67,7 +68,7 @@ def wait(max_wait_timeout: 60, wait_timeout: 0.1)
             )
           end
           sleep wait_timeout
-        elsif self[:response] != 0
+        elsif self[:response] != 0 && raise_response_error
           raise_error
         else
           return create_result

lib/rdkafka/version.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module Rdkafka
-  VERSION = "0.13.0"
+  VERSION = "0.14.0.rc1"
   LIBRDKAFKA_VERSION = "2.2.0"
   LIBRDKAFKA_SOURCE_SHA256 = "af9a820cbecbc64115629471df7c7cecd40403b6c34bfdbb9223152677a47226"
 end