use failed partition count result shorter than the cache (#404)

Author: mensfeld

CHANGELOG.md

@@ -6,10 +6,12 @@
 - [Enhancement] Alias `topic_name` as `topic` in the delivery report (mensfeld)
 - [Enhancement] Provide `label` producer handler and report reference for improved traceability (mensfeld)
 - [Enhancement] Include the error when invoking `create_result` on producer handle (mensfeld)
-- [Enhancement] Skip intermediate array creation on delivery report callback execution (one per message).
+- [Enhancement] Skip intermediate array creation on delivery report callback execution (one per message) (mensfeld).
+- [Enhancement] Report `-1` instead of `nil` in case `partition_count` failure (mensfeld).
 - [Fix] Fix return type on `#rd_kafka_poll` (mensfeld)
 - [Fix] `uint8_t` does not exist on Apple Silicon (mensfeld)
 - [Fix] Missing ACL `RD_KAFKA_RESOURCE_BROKER` constant reference (mensfeld)
+- [Fix] Partition cache caches invalid nil result for `PARTITIONS_COUNT_TTL` (mensfeld)
 - [Change] Rename `matching_acl_pattern_type` to `matching_acl_resource_pattern_type` to align the whole API (mensfeld)
 
 ## 0.15.0 (2023-12-03)

lib/rdkafka/producer.rb

@@ -40,10 +40,16 @@ def initialize(native_kafka, partitioner_name)
           topic_metadata = ::Rdkafka::Metadata.new(inner, topic).topics&.first
         end
 
-        cache[topic] = [
-          monotonic_now,
-          topic_metadata ? topic_metadata[:partition_count] : nil
-        ]
+        partition_count = topic_metadata ? topic_metadata[:partition_count] : -1
+
+        # This approach caches the failure to fetch only for 1 second. This will make sure, that
+        # we do not cache the failure for too long but also "buys" us a bit of time in case there
+        # would be issues in the cluster so we won't overaload it with consecutive requests
+        cache[topic] = if partition_count.positive?
+                         [monotonic_now, partition_count]
+                       else
+                         [monotonic_now - PARTITIONS_COUNT_TTL + 5, partition_count]
+                       end
       end
     end
 
@@ -137,14 +143,15 @@ def purge
     # Partition count for a given topic.
     #
     # @param topic [String] The topic name.
-    # @return [Integer] partition count for a given topic
+    # @return [Integer] partition count for a given topic or `-1` if it could not be obtained.
     #
     # @note If 'allow.auto.create.topics' is set to true in the broker, the topic will be
     #   auto-created after returning nil.
     #
     # @note We cache the partition count for a given topic for given time.
     #   This prevents us in case someone uses `partition_key` from querying for the count with
-    #   each message. Instead we query once every 30 seconds at most
+    #   each message. Instead we query once every 30 seconds at most if we have a valid partition
+    #   count or every 5 seconds in case we were not able to obtain number of partitions
     def partition_count(topic)
       closed_producer_check(__method__)
 
@@ -194,7 +201,7 @@ def produce(topic:, payload: nil, key: nil, partition: nil, partition_key: nil,
       if partition_key
         partition_count = partition_count(topic)
         # If the topic is not present, set to -1
-        partition = Rdkafka::Bindings.partitioner(partition_key, partition_count, @partitioner_name) if partition_count
+        partition = Rdkafka::Bindings.partitioner(partition_key, partition_count, @partitioner_name) if partition_count.positive?
       end
 
       # If partition is nil, use -1 to let librdafka set the partition randomly or