Class: Rdkafka::Producer::PartitionsCountCache

Inherits:
Object
  • Object
show all
Includes:
Helpers::Time
Defined in:
lib/rdkafka/producer/partitions_count_cache.rb

Overview

Note:

Design considerations:

Caching mechanism for Kafka topic partition counts to avoid frequent cluster queries

This cache is designed to optimize the process of obtaining partition counts for topics. It uses several strategies to minimize Kafka cluster queries:

  1. Statistics-based updates When statistics callbacks are enabled (via statistics.interval.ms), we leverage this data to proactively update the partition counts cache. This approach costs approximately 0.02ms of processing time during each statistics interval (typically every 5 seconds) but eliminates the need for explicit blocking metadata queries.

  2. Edge case handling If a user configures statistics.interval.ms much higher than the default cache TTL (Defaults::PARTITIONS_COUNT_CACHE_TTL_MSms), the cache will still function correctly. When statistics updates don't occur frequently enough, the cache entries will expire naturally, triggering a blocking refresh when needed.

  3. User configuration awareness The cache respects user-defined settings. If topic.metadata.refresh.interval.ms is set very high, the responsibility for potentially stale data falls on the user. This is an explicit design choice to honor user configuration preferences and align with librdkafka settings.

  4. Process-wide efficiency Since this cache is shared across all Rdkafka producers and consumers within a process, having multiple clients improves overall efficiency. Each client contributes to keeping the cache updated, benefiting all other clients.

  5. Thread-safety approach The implementation uses fine-grained locking with per-topic mutexes to minimize contention in multi-threaded environments while ensuring data consistency.

  6. Topic recreation handling If a topic is deleted and recreated with fewer partitions, the cache keeps reporting the higher count only until the entry's TTL expires. The first refresh after expiry performs an authoritative metadata read and adopts the lower count. Within the TTL window a lower value is still ignored, so a transient or racy lower read cannot clobber a correct higher count.

Instance Method Summary collapse

Methods included from Helpers::Time

#monotonic_now, #monotonic_now_ms

Constructor Details

#initialize(ttl = :not_provided, ttl_ms: :not_provided) ⇒ PartitionsCountCache

Creates a new partition count cache

Parameters:

  • ttl (Integer, nil) (defaults to: :not_provided) —

    DEPRECATED: Use ttl_ms instead. Time-to-live in seconds for cached values. Will be removed in v1.0.0.

  • ttl_ms (Integer, nil) (defaults to: :not_provided) —

    Time-to-live in milliseconds for cached values. Defaults to Defaults::PARTITIONS_COUNT_CACHE_TTL_MS.



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/rdkafka/producer/partitions_count_cache.rb', line 54

def initialize(ttl = :not_provided, ttl_ms: :not_provided)
  @counts = {}
  @mutex_hash = {}
  # Used only for @mutex_hash access to ensure thread-safety when creating new mutexes
  @mutex_for_hash = Mutex.new

  # Determine which TTL value to use
  if ttl != :not_provided && ttl_ms != :not_provided
    warn "DEPRECATION WARNING: Both ttl and ttl_ms were provided to PartitionsCountCache. " \
         "Using ttl_ms. The ttl parameter is deprecated and will be removed in v1.0.0."
    @ttl_ms = ttl_ms
  elsif ttl != :not_provided
    warn "DEPRECATION WARNING: ttl (seconds) parameter for PartitionsCountCache is deprecated. " \
         "Use ttl_ms (milliseconds) instead. This parameter will be removed in v1.0.0."
    @ttl_ms = (ttl * 1000).to_i
  elsif ttl_ms == :not_provided
    @ttl_ms = Defaults::PARTITIONS_COUNT_CACHE_TTL_MS
  else
    @ttl_ms = ttl_ms
  end
end

Instance Method Details

#get(topic) { ... } ⇒ Integer

Note:

The implementation prioritizes read performance over write consistency since partition counts typically only increase during normal operation.

Reads partition count for a topic with automatic refresh when expired

This method will return the cached partition count if available and not expired. If the value is expired or not available, it will execute the provided block to fetch the current value from Kafka.

Parameters:

  • topic (String) —

    Kafka topic name

Yields:

  • Block that returns the current partition count when cache needs refreshing

Yield Returns:

  • (Integer) —

    Current partition count retrieved from Kafka

Returns:

  • (Integer) —

    Partition count for the topic



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/rdkafka/producer/partitions_count_cache.rb', line 89

def get(topic)
  current_info = @counts[topic]

  if current_info.nil? || expired?(current_info[0])
    # The cached entry is missing or expired, so the block performs an authoritative metadata
    # read. We hand it to `set`, which adopts a higher count always and a lower count once the
    # entry has expired (e.g. the topic was recreated with fewer partitions). We then return
    # whatever `set` settled on so a concurrent refresh that wrote a higher value still wins.
    new_count = yield
    set(topic, new_count)

    return @counts[topic][1]
  end

  current_info[1]
end

#set(topic, new_count) ⇒ Object

Note:

We prioritize higher partition counts and only accept them when using a mutex to ensure consistency. This design decision is based on the fact that partition counts in Kafka only increase during normal operation.

Update partition count for a topic when needed

This method updates the partition count for a topic in the cache. It uses a mutex to ensure thread-safety during updates.

Parameters:

  • topic (String) —

    Kafka topic name

  • new_count (Integer) —

    New partition count value



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/rdkafka/producer/partitions_count_cache.rb', line 117

def set(topic, new_count)
  # First check outside mutex to avoid unnecessary locking
  current_info = @counts[topic]

  # Within the TTL window a lower value is treated as a stale/racy read and ignored, since
  # partition counts only grow during normal operation. Once the entry has expired a lower
  # value is an authoritative refresh (e.g. the topic was recreated with fewer partitions),
  # so we fall through and adopt it below.
  if current_info && new_count < current_info[1] && !expired?(current_info[0])
    refresh_timestamp(topic)

    return
  end

  # Only lock the specific topic mutex
  mutex_for(topic).synchronize do
    # Check again inside the lock as another thread might have updated
    current_info = @counts[topic]

    if current_info.nil?
      # Create new entry
      @counts[topic] = [monotonic_now_ms, new_count]
    elsif new_count > current_info[1] || expired?(current_info[0])
      # A higher count always wins; a lower count is accepted only when the existing entry
      # has expired, so a concurrent fresh higher value (which reset the timestamp) is never
      # clobbered by a stale lower one.
      current_info[0] = monotonic_now_ms
      current_info[1] = new_count
    else
      # Same or lower count within the TTL window: refresh the timestamp only
      current_info[0] = monotonic_now_ms
    end
  end
end

#to_h ⇒ Hash

Returns hash with ttls and partitions counts array.

Returns:

  • (Hash) —

    hash with ttls and partitions counts array



153
154
155
 
# File 'lib/rdkafka/producer/partitions_count_cache.rb', line 153

def to_h
  @counts
end