Class: Rdkafka::Metadata

Inherits:
Object
  • Object
show all
Defined in:
lib/rdkafka/metadata.rb

Overview

Provides cluster metadata information

Defined Under Namespace

Classes: BrokerMetadata, CustomFFIStruct, Metadata, PartitionMetadata, TopicMetadata

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(native_client, topic_name = nil, timeout_ms = Defaults::METADATA_TIMEOUT_MS) ⇒ Metadata

Fetches metadata from the Kafka cluster

Parameters:

  • native_client (FFI::Pointer) —

    pointer to the native Kafka client

  • topic_name (String, nil) (defaults to: nil) —

    specific topic to fetch metadata for, or nil for all topics

  • timeout_ms (Integer) (defaults to: Defaults::METADATA_TIMEOUT_MS) —

    timeout in milliseconds

Raises:



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/rdkafka/metadata.rb', line 25

def initialize(native_client, topic_name = nil, timeout_ms = Defaults::METADATA_TIMEOUT_MS)
  attempt = 0
  deadline = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) +
    Defaults::METADATA_RETRY_BUDGET_MS / 1_000.0

  begin
    attempt += 1
    fetch_metadata(native_client, topic_name, timeout_ms)
  rescue ::Rdkafka::RdkafkaError => e
    raise unless RETRIED_ERRORS.include?(e.code)
    raise if attempt > Defaults::METADATA_MAX_RETRIES

    # Stop once the wall-clock retry budget is spent, but only after at least
    # METADATA_MIN_ATTEMPTS tries so a slow broker (whose requests each consume the full
    # timeout) still gets a few tries rather than being cut off after one or two.
    raise if attempt >= Defaults::METADATA_MIN_ATTEMPTS &&
      ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) >= deadline

    # Exponential backoff between attempts, capped so a long retry sequence cannot block for
    # minutes. The request timeout (`timeout_ms`) is intentionally left unchanged: it used to be
    # overwritten with the backoff value, which shrank the first retries below the configured
    # timeout (near-guaranteeing another timeout) and then inflated later ones to ~100s.
    backoff_ms = [
      (2**attempt) * Defaults::METADATA_RETRY_BACKOFF_BASE_MS,
      Defaults::METADATA_RETRY_BACKOFF_MAX_MS
    ].min

    sleep(backoff_ms / 1_000.0)

    retry
  end
end

Instance Attribute Details

#brokers ⇒ Array<Hash> (readonly)

Returns list of broker metadata.

Returns:

  • (Array<Hash>) —

    list of broker metadata



7
8
9
 
# File 'lib/rdkafka/metadata.rb', line 7

def brokers
  @brokers
end

#topics ⇒ Array<Hash> (readonly)

Returns list of topic metadata.

Returns:

  • (Array<Hash>) —

    list of topic metadata



9
10
11
 
# File 'lib/rdkafka/metadata.rb', line 9

def topics
  @topics
end