Add network timeout artgument to create_topics
MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -538,10 +538,10 @@ defmodule KafkaEx do
```
"""
@spec create_topics(String.t) :: CreateTopics.Response.t
- def create_topics(topic_name, opts \\ []) do
+ def create_topics(requests, opts \\ []) do
worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
- # topic = Keyword.get(opts, :topic, "")
- Server.call(worker_name, {:create_topics, topic_name})
+ timeout = Keyword.get(opts, :timeout, 4000)
+ Server.call(worker_name, {:create_topics, requests, timeout})
end
#OTP API

MODIFY
lib/kafka_ex/protocol/create_topics.ex
lib/kafka_ex/protocol/create_topics.ex
@@ -31,10 +31,10 @@ defmodule KafkaEx.Protocol.CreateTopics do
defmodule TopicRequest do
defstruct topic: nil,
- num_partitions: nil,
- replication_factor: nil,
- replica_assignment: nil,
- config_entries: nil
+ num_partitions: -1,
+ replication_factor: -1,
+ replica_assignment: [],
+ config_entries: []
@type t :: %TopicRequest{
topic: binary,
num_partitions: integer,

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -260,8 +260,8 @@ defmodule KafkaEx.Server do
kafka_server_heartbeat(request, network_timeout, state)
end
- def handle_call({:create_topics, topic_name}, _from, state) do
- kafka_create_topics(topic_name, state)
+ def handle_call({:create_topics, requests, network_timeout}, _from, state) do
+ kafka_create_topics(requests, network_timeout, state)
end
def handle_info(:update_metadata, state) do

MODIFY
lib/kafka_ex/server_0_p_10_p_1.ex
lib/kafka_ex/server_0_p_10_p_1.ex
@@ -7,17 +7,11 @@ defmodule KafkaEx.Server0P10P1 do
alias KafkaEx.Server0P8P2
alias KafkaEx.Server0P9P0
- # alias KafkaEx.ConsumerGroupRequiredError
alias KafkaEx.InvalidConsumerGroupError
alias KafkaEx.Protocol.ConsumerMetadata
alias KafkaEx.Protocol.ConsumerMetadata.Response, as: ConsumerMetadataResponse
- # alias KafkaEx.Protocol.Heartbeat
- # alias KafkaEx.Protocol.JoinGroup
- # alias KafkaEx.Protocol.LeaveGroup
alias KafkaEx.Protocol.Metadata.Broker
- # alias KafkaEx.Protocol.SyncGroup
alias KafkaEx.Server.State
- # alias KafkaEx.NetworkClient
require Logger
@@ -100,10 +94,10 @@ defmodule KafkaEx.Server0P10P1 do
{:noreply, update_metadata(state, @metadata_api_version)}
end
- def kafka_create_topics(requests, state) do
+ def kafka_create_topics(requests, network_timeout, state) do
create_topics_request = %CreateTopics.Request{
create_topic_requests: requests,
- timeout: 2000
+ timeout: network_timeout
}
mainRequest = CreateTopics.create_request(state.correlation_id, @client_id, create_topics_request)

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -62,7 +62,7 @@ defmodule KafkaEx.Server0P8P0 do
def kafka_server_leave_group(_, _, _state), do: raise "Leave Group is not supported in 0.8.0 version of Kafka"
def kafka_server_heartbeat(_, _, _state), do: raise "Heartbeat is not supported in 0.8.0 version of kafka"
def kafka_server_update_consumer_metadata(_state), do: raise "Consumer Group Metadata is not supported in 0.8.0 version of kafka"
- def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
+ def kafka_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.8.0 version of kafka"
defp fetch(request, state) do
case network_request(request, Fetch, state) do

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -138,7 +138,7 @@ defmodule KafkaEx.Server0P8P2 do
def kafka_server_leave_group(_, _, _state), do: raise "Leave Group is not supported in 0.8.0 version of Kafka"
def kafka_server_heartbeat(_, _, _state), do: raise "Heartbeat is not supported in 0.8.0 version of kafka"
defp update_consumer_metadata(state), do: update_consumer_metadata(state, @retry_count, 0)
- def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
+ def kafka_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.8.0 version of kafka"
defp update_consumer_metadata(%State{consumer_group: consumer_group} = state, 0, error_code) do
Logger.log(:error, "Fetching consumer_group #{consumer_group} metadata failed with error_code #{inspect error_code}")

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
@@ -45,7 +45,7 @@ defmodule KafkaEx.Server0P9P0 do
defdelegate kafka_server_offset_commit(offset_commit_request, state), to: Server0P8P2
defdelegate kafka_server_consumer_group_metadata(state), to: Server0P8P2
defdelegate kafka_server_update_consumer_metadata(state), to: Server0P8P2
- def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
+ def kafka_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
def kafka_server_init([args]) do
kafka_server_init([args, self()])

MODIFY
test/integration/server0_p_10_p_1_test.exs
test/integration/server0_p_10_p_1_test.exs
@@ -5,7 +5,7 @@ defmodule KafkaEx.Server0P10P1.Test do
@tag :createtopic
test "can create a topic" do
- name = "topic3_#{:rand.uniform(2000000)}"
+ name = "create_topic_#{:rand.uniform(2000000)}"
request = %{
topic: name,
@@ -17,11 +17,11 @@ defmodule KafkaEx.Server0P10P1.Test do
%{config_name: "min.compaction.lag.ms", config_value: "0"}
]}
- resp = KafkaEx.create_topics([request])
+ resp = KafkaEx.create_topics([request], :timeout 2000)
# error = NONE
assert {0, name} == parse_create_topic_resp(resp)
- resp = KafkaEx.create_topics([request])
+ resp = KafkaEx.create_topics([request], :timeout 2000)
# error = TOPIC_ALREADY_EXISTS
assert {36, name} == parse_create_topic_resp(resp)

Add network timeout artgument to create_topics