diff --git a/lib/kafkaex_lag_exporter/kafka_utils.ex b/lib/kafkaex_lag_exporter/kafka_utils.ex index e8be8af..e3fba94 100644 --- a/lib/kafkaex_lag_exporter/kafka_utils.ex +++ b/lib/kafkaex_lag_exporter/kafka_utils.ex @@ -7,8 +7,6 @@ defmodule KafkaexLagExporter.KafkaUtils do alias KafkaexLagExporter.KafkaWrapper.Behaviour, as: KafkaWrapper - require Logger - @default_client :client1 def connection, do: connection(@default_client) @@ -46,10 +44,7 @@ defmodule KafkaexLagExporter.KafkaUtils do group_descriptions |> Enum.flat_map(fn %{group_id: consumer_group, members: members} -> - get_member_info(members) - |> Enum.map(fn {topics, consumer_id, member_host} -> - {consumer_group, topics, consumer_id, member_host} - end) + get_member_info(consumer_group, members) end) end @@ -63,7 +58,7 @@ defmodule KafkaexLagExporter.KafkaUtils do fetch_committed_offsets(topic, consumer_group, client) |> Enum.sort_by(fn {key, _value} -> key end) - for {{part, current}, {_part2, committed}} <- Enum.zip(offsets, committed_offsets) do + for {{part, current}, {_, committed}} <- Enum.zip(offsets, committed_offsets) do {part, current - committed} end end @@ -75,8 +70,7 @@ defmodule KafkaexLagExporter.KafkaUtils do {:ok, partitions_count} = KafkaWrapper.get_partitions_count(client, topic) for i <- Range.new(0, partitions_count - 1), - {:ok, offset} = - KafkaWrapper.resolve_offset(endpoints, topic, i, :latest, sock_opts) do + {:ok, offset} = KafkaWrapper.resolve_offset(endpoints, topic, i, :latest, sock_opts) do {i, offset} end end @@ -88,23 +82,31 @@ defmodule KafkaexLagExporter.KafkaUtils do {:ok, response} = KafkaWrapper.fetch_committed_offsets(endpoints, sock_opts, consumer_group) - for r <- response, - pr <- r[:partitions], - do: {pr[:partition_index], pr[:committed_offset]} + for r <- response, pr <- r[:partitions] do + {pr[:partition_index], pr[:committed_offset]} + end end @spec get_member_info( + binary, list(%{client_host: binary, member_assignment: binary, member_id: binary}) ) :: - list({topic_names :: list(binary), consumer_id :: binary, member_host :: binary}) - defp get_member_info(members) do - Enum.map(members, fn %{ - client_host: member_host, - member_assignment: member_assignment, - member_id: consumer_id - } -> + list( + {consumer_group :: binary, topics :: list(binary), consumer_id :: binary, + member_host :: binary} + ) + defp get_member_info(consumer_group, members) do + members + |> Enum.map(fn %{ + client_host: member_host, + member_assignment: member_assignment, + member_id: consumer_id + } -> topics = KafkaexLagExporter.TopicNameParser.parse_topic_names(member_assignment) {topics, consumer_id, member_host} end) + |> Enum.map(fn {topics, consumer_id, member_host} -> + {consumer_group, topics, consumer_id, member_host} + end) end end