v0.28.0

• [Enhancement] Bump librdkafka to 2.14.2. Maintenance release: fixes duplicate groups in ListConsumerGroups when multiple brokers return the same group, a data race in timers, and bumps bundled OpenSSL/libcurl/zstd/zlib/cJSON dependencies (several CVEs).
• [Enhancement] Add Consumer#metadata and Producer#metadata, mirroring Admin#metadata, so cluster/topic metadata can be fetched from an existing consumer or producer handle without opening a dedicated admin connection.
• [Enhancement] Name the failing partition and topic in the RdkafkaError raised for per-partition list_offsets errors (previously a bare error code), preserving the per-partition context the pre-batching Consumer#lag watermark errors carried.
• [Enhancement] Add Consumer#list_offsets, mirroring Admin#list_offsets, so batched offset queries (one ListOffsets request carrying all requested partitions, fanned out to the partition leaders by librdkafka) can be issued on an existing consumer handle without opening a dedicated admin connection, and rebuild Consumer#lag on top of it: lag is now computed from a single batched end-offsets query instead of one blocking query_watermark_offsets broker roundtrip per partition. To receive the results, consumer clients now register the librdkafka background event callback, so librdkafka spawns its internally-managed background thread for consumers as well. Consumer#lag keeps its exact pre-batching semantics: it forwards the consumer's configured isolation.level to the batched query (librdkafka resolves the per-partition watermark query with that level, so end offsets stay LSO-based for the default read_committed), surfaces a timeout of the batched query as an RdkafkaError (timed_out), and now also raises ClosedConsumerError when called on a closed consumer.
• [Enhancement] Extract the admin background-event result handlers into one class per operation under lib/rdkafka/callbacks/ (CreateTopicHandler, DescribeConfigsHandler, etc., all subclasses of Callbacks::BaseHandler). Callbacks::BackgroundEventCallback is now a thin dispatcher that maps the event type to its handler and destroys the event. Purely internal reorganization (Rdkafka::Callbacks is private) with no behavior or API change.
• [Enhancement] Expose replicas and isrs (in-sync replica broker ids) on each partition in topic metadata (Metadata#topics partition hashes). The base struct #to_h skips FFI pointer members, so these two arrays were dropped entirely and the partition replica assignment was unavailable to callers (e.g. for planning replication-factor changes). PartitionMetadata#to_h now dereferences both pointers into arrays of broker ids.
• [Enhancement] Reuse per-thread scratch pointers in Consumer::Headers.from_native instead of allocating them for every consumed message. Previously each message paid one native pointer allocation just to check for headers and three more when headers were present; now the scratch pointers are allocated once per thread/fiber and reused, removing all per-message native scratch allocations from the consumer hot path.
• [Enhancement] Remove the unused DeliveryHandle :topic_name struct field and the per-message allocation that populated it. The delivery callback copied the topic name into a native FFI::MemoryPointer on every delivered message, retained for the lifetime of the handle, yet nothing ever read it: the topic is already available via DeliveryHandle#topic (a Ruby attribute set during produce) and DeliveryReport#topic_name, both of which work exactly as before.
• [Fix] Stop poll_batch/poll_batch_nb from discarding a whole batch when one message fails to build. If building a Consumer::Message raised (e.g. a header-read RdkafkaError), the exception propagated out of the batch loop and the ensure destroyed every remaining message, so all the messages already built in that batch were lost even though their offsets had been stored - permanent silent loss of up to max_items messages. A per-message build error is now surfaced inline as an RdkafkaError in the returned array (the same way native error events already are), so the rest of the batch is preserved and the failure is visible to the caller. (Single-message poll/poll_nb still raise as before.)
• [Fix] Add the missing closed_consumer_check to Consumer#position. Every sibling offset method guards against a closed consumer, but position did not: with an explicit list it raised ClosedInnerError instead of ClosedConsumerError, and with no list it raised ClosedConsumerError naming assignment rather than position. It now raises a consistent ClosedConsumerError for position.
• [Fix] Stop leaking the native rd_kafka_topic_conf_t in Producer#set_topic_config when a per-topic config value is rejected. The conf is built with rd_kafka_topic_conf_new and only handed to rd_kafka_topic_new afterwards; if rd_kafka_topic_conf_set returned a non-:config_ok result the method raised before rd_kafka_topic_new was ever called, so librdkafka never took ownership of the conf and it leaked (repeatable per produce with an invalid topic_config:). The conf is now destroyed before raising. The rd_kafka_topic_new path is unchanged: librdkafka frees the conf there on both success and failure, so it is never destroyed by us.
• [Fix] Raise instead of silently dropping a rejected incremental_alter_configs entry. rd_kafka_ConfigResource_add_incremental_config returns an rd_kafka_error_t for an invalid op_type, an empty/nil name, or a nil value on a non-delete op; the result was ignored, so the entry was dropped, the alter request still reported success, and the error object leaked. The error is now surfaced as an RdkafkaError (raised before the request is sent) and the native error object is freed.
• [Fix] Let PartitionsCountCache adopt a lower partition count once the cached entry has expired. The cache prioritizes higher counts (partition counts only grow during normal operation), but it did so unconditionally: after the TTL expired it fetched the true lower count, discarded it, and re-armed the TTL on the stale higher count - permanently. A topic recreated with fewer partitions then made produce (with a partition key) fail with unknown_partition for the dropped partitions until process restart. A lower value is now adopted on the first refresh after expiry, while still being ignored within the TTL window so a transient or racy lower read cannot clobber a correct higher count.
• [Fix] Make the Consumer GC finalizer close the consumer and destroy its consumer queue, not just the native client. The finalizer used the generic NativeKafka one, which went straight to rd_kafka_destroy and skipped rd_kafka_consumer_close and rd_kafka_queue_destroy. A consumer that used poll_batch (which takes a consumer-queue reference) and was then garbage-collected without an explicit close left that reference dangling, which could make rd_kafka_destroy block inside the finalizer (process hang at GC/shutdown) or leak the handle. The consumer now installs its own finalizer that mirrors #close (the queue pointer is shared with the finalizer via a holder so it never captures the consumer and prevents collection).
• [Fix] Stop admin operations from leaking when their arguments are rejected. describe_configs and incremental_alter_configs allocated the background queue and AdminOptions and registered their handle before parsing input, so a missing key raised a KeyError out of the building loop that orphaned the handle in the process-global registry forever (and leaked the queue, AdminOptions and any ConfigResources already built); list_offsets likewise allocated its native topic-partition list before validating the offset specs. All three now parse and validate input up front, so a bad argument raises with nothing allocated.
• [Fix] Destroy the native topic-partition list in TopicPartitionList#to_native_tpl when population fails partway. The list is allocated with rd_kafka_topic_partition_list_new and only handed back (for the caller to destroy) on success; if building it raised - e.g. a non-string metadata value or an offset FFI cannot coerce to int64 - the half-built list leaked, since destruction is fully manual (the previous doc comment claiming GC handled it was wrong). It is now destroyed before the error propagates.
• [Fix] Allocate the admin result-count out-parameter as :size_t instead of :int32. Every librdkafka rd_kafka_*_result_*(result, size_t *cntp) accessor writes a full native size_t (8 bytes on 64-bit), but the count pointer was a 4-byte FFI::MemoryPointer.new(:int32) across the create/delete topic, create partitions, create/delete ACL, describe/incremental-alter configs and config-synonyms paths - a 4-byte heap overflow on every admin result parse (benign on little-endian, where the low word still reads the correct count, but undefined behavior). Now uses :size_t, matching the already-correct list-offsets and get_err_descs paths.
• [Fix] Destroy admin API background events after processing. librdkafka requires the application to destroy each background event, but rd_kafka_event_destroy was never called (nor even bound), so every admin operation leaked its entire result event with all result arrays and strings. Reports are now built inside the callback (copying event-owned memory into Ruby objects) before the event is destroyed, and DescribeConfigsReport/IncrementalAlterConfigsReport no longer destroy the event-owned ConfigResource array, which also fixes a double free on repeated wait calls on the same handle. As part of this, the internal FFI struct fields on admin operation handles (e.g. handle[:error_string], handle[:result_name], handle[:config_entries], handle[:response_string], handle[:matching_acls]) were removed; they were never part of the public API (use handle.wait and the returned report objects, whose interfaces are unchanged).
• [Fix] Resolve admin operation handles from the event error when an admin operation fails at the operation level (e.g. brokers unreachable, or the client closed with the request in flight). librdkafka delivers such failures as a result event with the error set and an empty results array, but the create topic, delete topic, create partitions, delete groups, create ACL and delete ACL handlers indexed results[0] unconditionally. That raised inside the background event callback, so the handle was never unlocked and wait blocked until its own timeout and raised WaitTimeoutError, discarding the real error. These handlers now check the event error first and resolve the handle with the actual error code (the describe configs, incremental alter configs, describe ACL and list offsets handlers already did).
• [Fix] Stop leaking the native rd_kafka_conf_t when client creation fails. native_config raised ConfigError mid-build (e.g. on an invalid option) without destroying the conf, and native_kafka raised ClientCreationError on a null rd_kafka_new without destroying it either. Both paths now call rd_kafka_conf_destroy before re-raising (librdkafka keeps app ownership of the conf on rd_kafka_new failure, so this is safe). Multi-KB leak per failed creation, relevant for supervisors retrying client creation on transient SASL/SSL misconfig.
• [Fix] Stop Metadata from leaking the native metadata struct (and a topic reference) on every retried fetch. retry restarts the begin block without running its ensure, so each retried attempt reassigned the pointers and only the last attempt's rd_kafka_metadata struct was ever destroyed; up to METADATA_MAX_RETRIES whole-cluster structs leaked per call (the leader_not_available case is routine during topic creation/leader election). Each attempt now frees its own native resources, and a failed fetch that never allocated a struct no longer calls rd_kafka_metadata_destroy on a NULL pointer.
• [Fix] Free the librdkafka-allocated string in Consumer#cluster_id and Consumer#member_id (previously copied via a :string binding but never freed) and fix the rd_kafka_clusterid arity to pass timeout_ms. Consumer#cluster_id now accepts a timeout_ms (default Defaults::CONSUMER_CLUSTER_ID_TIMEOUT_MS).
• [Fix] Guard the message delivery callback so a raising user delivery_callback can no longer skip the handle unlock or crash the producer. DeliveryCallback invoked the user callback and only then unlocked the handle, with no rescue; if the callback raised, the handle stayed pending (so wait blocked until its timeout and raised WaitTimeoutError for a message that was actually delivered) and the exception unwound out of the FFI callback on librdkafka's polling thread (abort_on_exception = true), taking down the whole process. The user callback is now wrapped so exceptions are logged and swallowed (matching the rebalance callback) and the handle is always unlocked in an ensure.
• [Fix] Stop Producer#produce from orphaning the delivery handle in the process-global registry when it fails after registering it. The handle was only removed on a non-zero rd_kafka_producev return, so any exception between registration and that check (a concurrent close making with_inner raise ClosedInnerError, or a header value whose #to_s raises) leaked the handle forever - it survives producer close and accumulates in apps that recreate/close producers. produce now removes the handle on any such failure before re-raising.
• [Fix] Attach rd_kafka_query_watermark_offsets with blocking: true so it releases the GVL during its broker round-trip. It was the only synchronous network call bound without the flag, so Consumer#query_watermark_offsets (and Consumer#lag, which calls it once per partition) froze every other Ruby thread in the process - including producer polling threads - for up to timeout_ms. Matches the neighboring rd_kafka_offsets_for_times binding.
• [Fix] Stabilize the flaky Consumer#lag "calculates the consumer lag" spec on overloaded CI. The manual consumer.commit could raise no_offset when the default 5s background auto-commit had already committed the stored offsets, or when the auto offset store had not yet caught up with poll. The spec now raises auto.commit.interval.ms to 60s (matching the existing #seek/pause specs) and lets the offset store settle before committing. Backported from rdkafka-ruby (karafka#912).
• [Fix] Fix the NULL background-queue cleanup branches in Admin#delete_group, Admin#delete_acl and Admin#describe_acl, which referenced undefined local variables (delete_topic_ptr/new_acl_ptr). When rd_kafka_queue_get_background returned NULL, those branches raised a NameError instead of the intended ConfigError and leaked the already-allocated native request object (the rd_kafka_DeleteGroup_t / ACL binding filter), and delete_group additionally called the wrong destructor. They now destroy the correct object and raise ConfigError.
• [Fix] Forward broker_message and instance_name through RdkafkaError.build. On the rd_kafka_error_t pointer path build called build_from_c without passing either, so a caller-supplied broker message was discarded (falling back to rd_kafka_err2str) and the instance name was lost; the Bindings::Message path also dropped instance_name. Both are now forwarded, and build_from_c accepts instance_name.
• [Fix] Synchronize AbstractHandle::REGISTRY mutations. The handle registry is a plain Hash mutated from producing/consuming threads and the background polling thread (which removes handles from FFI callbacks). That is effectively safe on MRI under the GVL but not on JRuby, where a lost write could leave a handle unregistered (never unlocked, so wait times out for a delivered message) or never removed (permanent leak). register/remove now guard the Hash with a mutex.
• [Fix] Stop the Metadata retry loop from clobbering the request timeout and blocking for minutes. Each retry overwrote the per-request timeout with the exponential backoff value, so the first retries ran with a far-too-short request timeout (e.g. 200ms vs the 2,000ms default, near-guaranteeing another timeout) while later ones inflated it to ~100s; the cumulative sleeps alone reached ~204s. The request timeout is now left unchanged across retries, the per-attempt backoff is capped at Defaults::METADATA_RETRY_BACKOFF_MAX_MS (1,000ms), and the loop is bounded by a wall-clock budget (Defaults::METADATA_RETRY_BUDGET_MS, 5,000ms) after a floor of Defaults::METADATA_MIN_ATTEMPTS (3) tries. This keeps a synchronous metadata fetch from blocking for minutes while still giving a slow broker a few attempts: fast-recovering clusters retry quickly within the budget, an unresponsive one fails in ~5s (a bit over if its requests each consume the full timeout), and at least 3 attempts always happen.
• [Fix] Cache the partition count for a missing topic so produce with a partition_key to a not-yet-created topic no longer runs a blocking metadata query on every message. The unknown_topic_or_part result was resolved to RD_KAFKA_PARTITION_UA outside the partition-count cache, so nothing was stored and each call performed a fresh synchronous metadata RPC (the promised negative cache did not exist). The miss is now resolved inside the cache block, so it is cached like any other value and reused for the cache TTL.