Remove function retrieve_metadate_with_version
MODIFY
lib/kafka_ex/protocol/metadata.ex
lib/kafka_ex/protocol/metadata.ex
@@ -100,9 +100,6 @@ defmodule KafkaEx.Protocol.Metadata do
def create_request(correlation_id, client_id, ""), do: create_request(correlation_id, client_id, "", @default_api_version)
def create_request(correlation_id, client_id, topics) when is_list(topics), do: create_request(correlation_id, client_id, topics, @default_api_version)
- def create_request(correlation_id, client_id, "", nil), do: create_request(correlation_id, client_id, "", @default_api_version)
- def create_request(correlation_id, client_id, topics, nil) when is_list(topics), do: create_request(correlation_id, client_id, topics, @default_api_version)
-
def create_request(correlation_id, client_id, "", api_version) do
topic_count = if 0 == api_version, do: 0, else: -1
KafkaEx.Protocol.create_request(:metadata, correlation_id, client_id, api_version) <> << topic_count :: 32-signed >>

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -383,10 +383,10 @@ defmodule KafkaEx.Server do
{:noreply, update_metadata(state)}
end
- def update_metadata(state), do: update_metadata(state, nil)
+ def update_metadata(state), do: update_metadata(state, 0)
def update_metadata(state, api_version) do
- {correlation_id, metadata} = retrieve_metadata_with_version(state.brokers, state.correlation_id, config_sync_timeout(), api_version)
+ {correlation_id, metadata} = retrieve_metadata(state.brokers, state.correlation_id, config_sync_timeout(), [], api_version)
metadata_brokers = metadata.brokers |> Enum.map(&(%{&1 | is_controller: &1.node_id == metadata.controller_id}))
brokers = state.brokers
|> remove_stale_brokers(metadata_brokers)
@@ -395,24 +395,20 @@ defmodule KafkaEx.Server do
end
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata(brokers, correlation_id, sync_timeout, topic \\ []), do: retrieve_metadata(brokers, correlation_id, sync_timeout, topic, @retry_count, 0)
+ def retrieve_metadata(brokers, correlation_id, sync_timeout, topic \\ [], api_version \\ 0) do
+ retrieve_metadata(brokers, correlation_id, sync_timeout, topic, @retry_count, 0, api_version)
+ end
- # credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata_with_version(brokers, correlation_id, sync_timeout, api_version, topic \\ []), do: retrieve_metadata(brokers, correlation_id, sync_timeout, topic, @retry_count, api_version, 0)
+ def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, error_code, api_version \\ 0)
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata(_, correlation_id, _sync_timeout, topic, 0, error_code) do
+ def retrieve_metadata(_, correlation_id, _sync_timeout, topic, 0, error_code, api_version) do
Logger.log(:error, "Metadata request for topic #{inspect topic} failed with error_code #{inspect error_code}")
{correlation_id, %Metadata.Response{}}
end
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, error_code) do
- retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, nil, error_code)
- end
-
- # credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, api_version, _error_code) do
+ def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, _error_code, api_version) do
metadata_request = Metadata.create_request(correlation_id, @client_id, topic, api_version)
data = first_broker_response(metadata_request, brokers, sync_timeout)
if data do
@@ -422,7 +418,7 @@ defmodule KafkaEx.Server do
nil -> {correlation_id + 1, response}
topic_metadata ->
:timer.sleep(300)
- retrieve_metadata(brokers, correlation_id + 1, sync_timeout, topic, retry - 1, topic_metadata.error_code)
+ retrieve_metadata(brokers, correlation_id + 1, sync_timeout, topic, retry - 1, topic_metadata.error_code, api_version)
end
else
message = "Unable to fetch metadata from any brokers. Timeout is #{sync_timeout}."

MODIFY
lib/kafka_ex/server_0_p_10_p_1.ex
lib/kafka_ex/server_0_p_10_p_1.ex
@@ -65,7 +65,7 @@ defmodule KafkaEx.Server0P10P1 do
brokers = Enum.map(uris, fn({host, port}) -> %Broker{host: host, port: port, socket: NetworkClient.create_socket(host, port, ssl_options, use_ssl)} end)
- {correlation_id, metadata} = retrieve_metadata_with_version(brokers, 0, config_sync_timeout(), @metadata_api_version)
+ {correlation_id, metadata} = retrieve_metadata(brokers, 0, config_sync_timeout(), [], @metadata_api_version)
state = %State{metadata: metadata, brokers: brokers, correlation_id: correlation_id, consumer_group: consumer_group, metadata_update_interval: metadata_update_interval, consumer_group_update_interval: consumer_group_update_interval, worker_name: name, ssl_options: ssl_options, use_ssl: use_ssl}
# Get the initial "real" broker list and start a regular refresh cycle.
state = update_metadata(state, @metadata_api_version)
@@ -85,7 +85,7 @@ defmodule KafkaEx.Server0P10P1 do
end
def kafka_server_metadata(topic, state) do
- {correlation_id, metadata} = retrieve_metadata_with_version(state.brokers, state.correlation_id, config_sync_timeout(), @metadata_api_version, topic)
+ {correlation_id, metadata} = retrieve_metadata(state.brokers, state.correlation_id, config_sync_timeout(), topic, @metadata_api_version)
updated_state = %{state | metadata: metadata, correlation_id: correlation_id}
{:reply, metadata, updated_state}
end

Remove function retrieve_metadate_with_version