Simplify code

This commit is contained in:
2024-03-28 21:23:51 +01:00
parent 6bf68b74ec
commit 08b5923d52
2 changed files with 26 additions and 24 deletions

View File

@@ -1,8 +1,6 @@
defmodule KafkaexLagExporter.ConsumerOffsetFetcher do
@moduledoc "Calculate summarized lag for each consumer group"
require Logger
alias KafkaexLagExporter.KafkaUtils
# TODO fix type
@@ -14,30 +12,25 @@ defmodule KafkaexLagExporter.ConsumerOffsetFetcher do
consumer_group_names = KafkaUtils.get_consumer_group_names(endpoint)
consumer_lags =
KafkaUtils.topic_names_for_consumer_groups(
endpoint,
[],
consumer_group_names
)
|> Enum.map(fn {consumer_group, topics} ->
{consumer_group, get_lag_for_consumer(consumer_group, topics)}
end)
KafkaUtils.topic_names_for_consumer_groups(endpoint, [], consumer_group_names)
|> Enum.flat_map(&get_lag_per_topic(&1))
consumer_lag_sum = get_lag_for_consumer_sum(consumer_lags)
%{lags: consumer_lags, sum: consumer_lag_sum}
end
defp get_lag_for_consumer(consumer_group, topics) do
topics
|> Enum.flat_map(fn topic ->
KafkaUtils.lag(topic, consumer_group, :client1)
defp get_lag_per_topic({consumer_group, topics}) do
Enum.map(topics, fn topic ->
lag = KafkaUtils.lag(topic, consumer_group, :client1)
{consumer_group, topic, lag}
end)
end
defp get_lag_for_consumer_sum(lags_per_consumer_group) do
lags_per_consumer_group
|> Enum.map(fn {topic, lag_per_partition} -> {topic, sum_topic_lag(lag_per_partition, 0)} end)
Enum.map(lags_per_consumer_group, fn {consumer_group, topic, lag_per_partition} ->
{consumer_group, topic, sum_topic_lag(lag_per_partition, 0)}
end)
end
defp sum_topic_lag([], acc), do: acc