Change type of TopicError.error_code from integer to atom
MODIFY
lib/kafka_ex/protocol.ex
lib/kafka_ex/protocol.ex
@@ -106,7 +106,20 @@ defmodule KafkaEx.Protocol do
28 => :invalid_commit_offset,
29 => :topic_authorization_failed,
30 => :group_authorization_failed,
- 31 => :cluster_authorization_failed
+ 31 => :cluster_authorization_failed,
+ 32 => :invalid_timestamp,
+ 33 => :unsupported_sasl_mechanism,
+ 34 => :illegal_sasl_state,
+ 35 => :unsupported_version,
+ 36 => :topic_already_exists,
+ 37 => :invalid_partitions,
+ 38 => :invalid_replication_factor,
+ 39 => :invalid_replica_assignment,
+ 40 => :invalid_config,
+ 41 => :not_controller,
+ 42 => :invalid_request,
+ 43 => :unsupported_for_message_format,
+
}
@spec error(integer) :: atom | integer

MODIFY
lib/kafka_ex/protocol/create_topics.ex
lib/kafka_ex/protocol/create_topics.ex
@@ -52,7 +52,7 @@ defmodule KafkaEx.Protocol.CreateTopics do
defmodule TopicError do
defstruct topic_name: nil, error_code: nil
- @type t :: %TopicError{ topic_name: binary, error_code: integer }
+ @type t :: %TopicError{ topic_name: binary, error_code: atom }
end
defmodule Response do
@@ -137,7 +137,7 @@ defmodule KafkaEx.Protocol.CreateTopics do
@spec parse_topic_errors(integer, binary) :: [TopicError.t]
defp parse_topic_errors(topic_errors_count,
<< topic_name_size :: 16-signed, topic_name :: size(topic_name_size)-binary, error_code :: 16-signed, rest :: binary >>) do
- [%TopicError{topic_name: topic_name, error_code: error_code} | parse_topic_errors(topic_errors_count - 1, rest)]
+ [%TopicError{topic_name: topic_name, error_code: Protocol.error(error_code)} | parse_topic_errors(topic_errors_count - 1, rest)]
end
end

MODIFY
test/integration/server0_p_10_p_1_test.exs
test/integration/server0_p_10_p_1_test.exs
@@ -19,12 +19,10 @@ defmodule KafkaEx.Server0P10P1.Test do
]}
resp = KafkaEx.create_topics([request], timeout: 2000)
- # error = NONE
- assert {0, name} == parse_create_topic_resp(resp)
+ assert {:no_error, name} == parse_create_topic_resp(resp)
resp = KafkaEx.create_topics([request], timeout: 2000)
- # error = TOPIC_ALREADY_EXISTS
- assert {36, name} == parse_create_topic_resp(resp)
+ assert {:topic_already_exists, name} == parse_create_topic_resp(resp)
wait_for(fn ->
topics = KafkaEx.metadata.topic_metadatas |> Enum.map(&(&1.topic))

Change type of TopicError.error_code from integer to atom