v2.10.2

• [Feature] Expose Producer#current_variant as a public method. It returns the variant active for the current dispatch on the current fiber - the custom variant while inside a #with/#variant-wrapped call, otherwise the producer's default variant - so middleware and instrumentation listeners running synchronously within a dispatch can read the effective per-dispatch settings (topic_config, max_wait_timeout, default?). The lookup is fiber-local and dispatch-scoped: outside a variant-wrapped call (or from an asynchronous delivery callback) it returns the default variant.
• [Enhancement] Stop allocating one interpolated string per message in LoggerListener batch produce handlers. The quoted topic strings were only ever counted (quoting is a 1:1 mapping), never displayed, so counting the raw topic values yields the identical number with zero string allocations - relevant for large produce_many_* batches with the default logger listener attached.
• [Enhancement] Use Array#concat in Producer#buffer_many instead of appending messages one by one.
• [Enhancement] Skip building the message.acknowledged instrumentation payload in the delivery callback when nothing is subscribed to that event. The notifications bus already short-circuits on empty listeners, but only after the payload hash was allocated - once per delivered message on the polling thread. Mirrors the listener guard already used by the statistics callback. Late subscribers keep working as the check happens on each emission.
• [Enhancement] Resolve the fiber-local variant once per #produce call and once per #produce_many_sync wait phase instead of re-resolving it for every usage and for every waited delivery handle. For a 1,000-message sync batch this removes ~2,000 redundant fiber-local lookups.
• [Enhancement] Do not allocate the fiber-local variants hash on the Producer#current_variant read path. Previously every fiber that produced messages got a Hash pinned to it for the fiber's lifetime (per producer use), even when variants were never used - wasteful under fiber-per-request servers (Falcon, async). The hash is now only created by variant wrapper methods that actually need to write to it.
• [Enhancement] Cache the variant validation contract in a constant instead of instantiating a new Contracts::Variant on every Producer#with / Producer#variant call (mirrors the existing Transactions::CONTRACT pattern).
• [Enhancement] Cache the tombstone validation contract in a constant instead of instantiating a new Contracts::Tombstone per tombstone message, removing per-message allocations in the tombstone_* APIs (mirrors the existing Transactions::CONTRACT pattern).
• [Enhancement] Replace explicit Warning[:performance] opt-in with a dynamic approach using Warning.categories (available since Ruby 3.4) to automatically enable all stable opt-in warning categories in the test suite, including :strict_unused_block introduced in Ruby 4.0.
• [Fix] Prevent a deadlock between a transactional single-message dispatch and #close. A single produce_sync/produce_async on a transactional producer incremented the operations counter (which #close drains while holding @transaction_mutex) before acquiring @transaction_mutex for its per-message transaction - an inverted lock order. A dispatch that had counted itself but not yet taken @transaction_mutex could deadlock a concurrent #close permanently (the close wait loop has no timeout). Transactional dispatches now take @transaction_mutex before the operation is counted, matching #close's lock order (@transaction_mutex -> @operating_mutex -> operations counter).
• [Fix] Prevent a deadlock (ThreadError: deadlock; recursive locking) when closing an idempotent producer (with reload_on_idempotent_fatal_error enabled) that has buffered messages whose final flush surfaces a fatal librdkafka error. #close performs the final flush while already holding @operating_mutex, and the idempotent fatal-error reload tried to re-acquire that same mutex, leaving the producer stuck in :closing with the native client leaked. The idempotent reload is now skipped on the closing path, and the final buffer flush is best-effort so client teardown always completes.
• [Fix] Make concurrent idempotent fatal-error reload thread-safe. When several threads shared an idempotent producer (with reload_on_idempotent_fatal_error enabled), a single fatal librdkafka condition failed all their in-flight produces at once and each entered the reload path; the second reload ran reload! after the first had already reset @client to nil, raising NoMethodError. The idempotent reload now bails out if another thread already reloaded (mirroring the transactional path's return if @status.configured? guard). Additionally, Status#active? now classifies the lifecycle from a single atomic read and Producer#ensure_active! branches on one snapshot, so a concurrent configured -> connected transition during a reload can no longer make ensure_active! raise StatusInvalidError for a valid, active producer.
• [Fix] Stop #flush_async / #flush_sync from silently dropping valid buffered messages when the dispatch fails. #flush removes the batch from the internal buffer before dispatching it, and a failure (a single invalid message failing validation before anything is sent, or a mid-batch inline error such as queue full) previously discarded the entire taken batch - the removed messages were never restored. A failed flush now re-buffers the messages that never reached librdkafka (the whole batch on validation failure or on a transactional rollback, the unsent remainder otherwise) so they can be retried instead of being lost.
• [Fix] Make Producer#close fork-safe so the GC finalizer inherited by a forked child can no longer close the parent's client. #client registers an ObjectSpace finalizer that calls #close; that finalizer is inherited across fork, and a child that inherited a used producer, never touched it, and exited normally would run #close in the child - flushing and closing (with the real rdkafka client, rd_kafka_destroy on a fork-inherited handle, i.e. undefined behavior) a client owned by the parent. #close now detects when it runs in a process other than the one that built the client, drops the inherited references and finalizer, and returns without touching the native client (matching the existing fork guard on the #client path).
• [Fix] Guard the internal buffer appends in Producer#buffer and Producer#buffer_many with @buffer_mutex. The appends mutated the shared @messages buffer without the lock that flush/purge/close hold while swapping it for a fresh array, so a concurrent swap landing between reading @messages and appending could drop the message into an orphaned array that is never dispatched - silently losing buffered messages in the documented "buffer in one thread, flush in another" pattern.
• [Fix] Stop a nested same-producer variant call from clobbering the outer variant inside a variant transaction block. transaction is the only variant-wrapped method that yields user code, so a variant call nested inside it (another variant.produce_*, or a raw producer dispatch in the same scope) used to delete the shared Fiber.current.waterdrop_clients entry on return, making the rest of the block silently fall back to the default variant and dispatch with default topic_config (timeouts, compression, partitioner) instead of the altered one. The wrapper now saves and restores the previous entry instead of unconditionally deleting it (still deleting when there was none, so the fiber-local hash does not accumulate stale keys).
• [Fix] Stop ConnectionPool#shutdown and #reload from silently dropping in-flight messages. Both closed every pooled producer with close! (force), which flushes for the max wait timeout and then purges whatever has not drained - so on a slow or unreachable broker, queued produce_async messages were cancelled and lost with no delivery report. They now close producers gracefully by default (#reload always; #shutdown unless called with the new force: true), letting messages flush instead of being purged. Pass pool.shutdown(force: true) to keep the old force-and-purge behavior.
• [Fix] Close a race in the FD poller where a producer registered while the last one was being torn down could be left permanently unpolled (sync produces hang until timeout, async deliveries are never acknowledged). The poller thread decided to exit (last producer unregistered) and cleared its thread reference in two separate, unsynchronized steps, so a register landing in that gap saw the still-alive exiting thread, skipped starting a fresh one, and then had its producer's state closed by the exiting thread's cleanup. The thread now decides to stop and clears its reference in a single mutex section, so a racing register either keeps it running or starts a fresh thread; and the exit cleanup runs only on an abnormal exit, since a normal exit always leaves an empty registry and so can never close a producer registered in the gap.

v2.10.1

• [Fix] Prevent Producer#close from raising ThreadError: can't be called from trap context when invoked from a Ruby signal trap context (e.g. Puma's after_stopped DSL hook in single mode). close now detects this case and delegates to a background thread, joining it so the caller blocks until the producer is fully closed (#866).

v2.10.0

• [Fix] Clean up native rdkafka client, global instrumentation callbacks, and poller registration when init_transactions fails during producer client construction. Previously, each failed attempt permanently leaked native threads, pipe file descriptors, and callback registry entries because the started rd_kafka_t handle was abandoned without being destroyed.
• [Breaking] Skip emitting librdkafka statistics when nothing is subscribed to statistics.emitted at the time the underlying rdkafka client is constructed. When no listener is present at build time, statistics.interval.ms is forced to 0 regardless of user configuration and the statistics callback is not registered, saving substantial allocations in the hot path (no JSON parsing, no statistics hash materialization, no decoration work). To use statistics, subscribe a listener to statistics.emitted BEFORE the first producer use (before the underlying client is lazily initialized).
• [Breaking] Raise WaterDrop::Errors::StatisticsNotEnabledError when attempting to subscribe to statistics.emitted (either via block or via a listener that responds to on_statistics_emitted) on a monitor where librdkafka statistics have been disabled at client build time. This replaces the "silent nothing" failure mode with an immediate, actionable error that pinpoints the timing mistake.
• [Feature] Add tombstone API (#tombstone_sync, #tombstone_async, #tombstone_many_sync, #tombstone_many_async) for producing tombstone records (nil-payload messages) with required key and partition validation. Works with variants.
• [Fix] Add ensure_same_process! to Poller#unregister for fork safety. Without this, a child process that inherited a pre-fork producer would deadlock on producer.close because unregister waited on a latch the dead parent poller thread would never release.

v2.9.0

• [Fix] Use delete in the variant ensure block to avoid leaving stale nil entries in Fiber.current.waterdrop_clients and prevent memory leaks in long-running processes (#836).
• [Fix] Exclude test files, .github/, and log/ directories from gem releases to reduce package size.
• [Breaking] Switch default polling mode from :thread to :fd. If you experience any issues, you can revert to the previous behavior by setting config.polling.mode = :thread. The :thread mode will be deprecated in 2.10 and removed in 2.11.
• [Breaking] Statistics decorator now only decorates keys used by the built-in Datadog metrics listener (tx, txretries, txerrs, rxerrs) and skips unused subtrees (topics, broker window stats). This reduces decoration cost by ~425x on large clusters (10 brokers, 20 topics, 2000 partitions). Users who rely on other _d or _fd keys in custom instrumentation should provide a custom decorator via config.statistics_decorator.
• [Feature] Add config.statistics_decorator setting to allow full control over the StatisticsDecorator instance used for statistics decoration. Users can provide a custom decorator with different only_keys and excluded_keys to match their instrumentation needs.
• [Change] Upscale default timeout values 3x closer to librdkafka defaults to prevent intermediate timeouts during node recovery (message.timeout.ms: 50s → 150s, transaction.timeout.ms: 55s → 165s, max_wait_timeout: 60s → 180s).

Upgrade Notes: https://karafka.io/docs/Upgrades-WaterDrop-2.9/

v2.8.16

• [Feature] Add FD-based polling mode (config.polling.mode = :fd) as an alternative to the default thread-based polling. FD mode uses a single Ruby thread with IO.select for efficient multiplexing, providing 39-54% higher throughput, lower memory usage, and fewer threads compared to the default thread mode.
• [Feature] Add config.polling.fd.max_time setting (default: 100ms) to control maximum polling time per producer per cycle. This enables per-producer priority differentiation.
• [Feature] Add Producer#queue_size (alias: #queue_length) to return the count of messages pending in the librdkafka queue. This counts messages that have been dispatched to librdkafka but not yet transmitted to the Kafka broker.
• [Enhancement] Add poll_nb and poll_drain methods to rdkafka Producer for efficient non-blocking polling without GVL release overhead.
• [Enhancement] Add poller.producer_registered and poller.producer_unregistered instrumentation events for FD mode.
• [Change] Update to use max_wait_timeout_ms parameter in rdkafka AbstractHandle#wait for consistency with librdkafka conventions.
• [Change] Require karafka-rdkafka >= 0.24.0 due to max_wait_timeout_ms API change and FD polling support.

v2.8.15

• [Enhancement] Skip statistics decoration and emission when no listeners are subscribed to statistics.emitted events to reduce overhead.
• [Enhancement] Support late subscription to statistics.emitted by checking for listeners on each emission (every 5 seconds).

v2.8.14

• [Fix] Fatal error replacement not working without exact error unwind from the fatal envelope
• [Testing] Add WaterDrop::Producer::Testing module for injecting and querying librdkafka fatal errors in tests.
• [Testing] Add #trigger_test_fatal_error(error_code, reason) method to simulate fatal errors without requiring actual broker-side conditions.
• [Testing] Add #fatal_error method to query current fatal error state for validation in tests.
• [Testing] Add comprehensive test coverage for fatal error injection, reload behavior, and event instrumentation with real librdkafka errors.
• [Change] Require karafka-rdkafka >= 0.23.1 due to error handling fixes.

v2.8.13

• [Enhancement] Make fenced error skip-reload behavior configurable via new non_reloadable_errors setting (defaults to [:fenced] for backward compatibility).
• [Enhancement] Add producer.reload event allowing config modification before reload to escape fencing loops (#706).
• [Enhancement] Do not early initialize the new instance on reload.

v2.8.12

• [Enhancement] Introduce reload_on_idempotent_fatal_error to automatically reload librdkafka producer after fatal errors on idempotent (non-transactional) producers.
• [Enhancement] Add configurable backoff and retry limits for fatal error recovery to prevent infinite reload loops:
  • wait_backoff_on_idempotent_fatal_error (default: 5000ms) - backoff before retrying after idempotent fatal error reload
  • max_attempts_on_idempotent_fatal_error (default: 5) - max reload attempts for idempotent fatal errors
  • wait_backoff_on_transaction_fatal_error (default: 1000ms) - backoff after transactional fatal error reload
  • max_attempts_on_transaction_fatal_error (default: 10) - max reload attempts for transactional fatal errors
• [Enhancement] Ensure error.occurred is instrumented before idempotent fatal error reload for visibility.
• [Enhancement] Automatically reset fatal error reload attempts counter on successful produce/transaction to allow recovery.
• [Refactor] Extract idempotence-related logic into separate WaterDrop::Producer::Idempotence module.
• [Refactor] Initialize @idempotent and @transactional instance variables in Producer#initialize for consistent Ruby object shapes optimization.
• [Refactor] Add idempotent_reloadable? and idempotent_retryable? methods to encapsulate idempotent fatal error reload checks.
• [Refactor] Add transactional_retryable? method to encapsulate transactional fatal error reload retry checks.
• [Fix] Waterdrop config.kafka errors on frozen hash.
• [Fix] Producer#transactional? method now correctly computes transactional status when @transactional is initialized to nil.

v2.8.11

• [Enhancement] Provide fast-track for middleware-less flows (20% faster) for single message, 5000x faster for batches.
• [Enhancement] Optimize middlewares application by around 20%.
• [Change] Remove Ruby 3.1 according to the EOL schedule.
• [Fix] Connection pool timeout parameter now accepts milliseconds instead of seconds for consistency with other WaterDrop timeouts. The default timeout has been changed from 5 seconds to 5000 milliseconds (equivalent value).