View Source KafkaEx.Cluster.ClusterMetadata (kafka_ex v1.0.0-rc.1)

Encapsulates what we know about the state of a Kafka broker cluster

This module is mainly used internally in New.Client, but some of its functions may be useful for extracting metadata information

Summary

Types

node_select_error()

Possible errors given by select_node/2

t()

Functions

broker_by_node_id(cluster_metadata, node_id)

Returns a t:Broker.t/0 for the given t:KafkaExAPI.node_id/0 or nil if there is no known broker with that node id

brokers(cluster_metadata)

Return a list of the known brokers

known_topics(cluster_metadata)

List names of topics known by the cluster metadata

merge_brokers(old_cluster_metadata, new_cluster_metadata)

Merge two sets of cluster metadata with their brokers. Returns the merged metadata and a list of brokers that should have their connections closed because they are not present in the new metadata

put_consumer_group_coordinator(cluster_metadata, consumer_group, coordinator_node_id)

update a consumer group coordinator node id

remove_topics(cluster_metadata, topics_to_remove)

remove the given topics (e.g., when they are deleted)

select_node(cluster_metadata, node_selector)

Find the node id for a given selector

topics_metadata(cluster_metadata, wanted_topics)

Return the metadata for the given topics

update_brokers(cluster_metadata, cb)

Execute an update callback on each broker

Types

node_select_error()

@type node_select_error() ::
  :no_such_node | :no_such_topic | :no_such_partition | :no_such_consumer_group

Possible errors given by select_node/2

t()

@type t() :: %KafkaEx.Cluster.ClusterMetadata{
  brokers: %{required(KafkaEx.API.node_id()) => KafkaEx.Cluster.Broker.t()},
  consumer_group_coordinators: %{
    required(KafkaEx.API.consumer_group_name()) => KafkaEx.API.node_id()
  },
  controller_id: KafkaEx.API.node_id() | nil,
  topics: %{required(KafkaEx.API.topic_name()) => KafkaEx.Cluster.Topic.t()}
}

Functions

broker_by_node_id(cluster_metadata, node_id)

@spec broker_by_node_id(t(), KafkaEx.API.node_id()) ::
  KafkaEx.Cluster.Broker.t() | nil

Returns a t:Broker.t/0 for the given t:KafkaExAPI.node_id/0 or nil if there is no known broker with that node id

brokers(cluster_metadata)

@spec brokers(t()) :: [KafkaEx.Cluster.Broker.t()]

Return a list of the known brokers

known_topics(cluster_metadata)

@spec known_topics(t()) :: [KafkaEx.API.topic_name()]

List names of topics known by the cluster metadata

Tthis is a subset of the topics in the cluster - it will only contain topics for which we have fetched metadata

merge_brokers(old_cluster_metadata, new_cluster_metadata)

@spec merge_brokers(t(), t()) :: {t(), [KafkaEx.Cluster.Broker.t()]}

Merge two sets of cluster metadata with their brokers. Returns the merged metadata and a list of brokers that should have their connections closed because they are not present in the new metadata

put_consumer_group_coordinator(cluster_metadata, consumer_group, coordinator_node_id)

@spec put_consumer_group_coordinator(
  t(),
  KafkaEx.API.consumer_group_name(),
  KafkaEx.API.node_id()
) ::
  t()

update a consumer group coordinator node id

remove_topics(cluster_metadata, topics_to_remove)

@spec remove_topics(t(), [KafkaEx.API.topic_name()]) :: t()

remove the given topics (e.g., when they are deleted)

select_node(cluster_metadata, node_selector)

@spec select_node(t(), KafkaEx.Client.NodeSelector.t()) ::
  {:ok, KafkaEx.API.node_id()} | {:error, node_select_error()}

Find the node id for a given selector

Note this will not update the metadata, only select a node given the current metadata.

See KafkaEx.Client.NodeSelector.t/0

topics_metadata(cluster_metadata, wanted_topics)

@spec topics_metadata(t(), [KafkaEx.API.topic_name()]) :: [KafkaEx.Cluster.Topic.t()]

Return the metadata for the given topics

update_brokers(cluster_metadata, cb)

@spec update_brokers(t(), (KafkaEx.Cluster.Broker.t() -> KafkaEx.Cluster.Broker.t())) ::
  t()

Execute an update callback on each broker