Merge pull request #310 from klarna/make-spec-backward-compatible

Make spec backward compatible

Makefile

@@ -1,7 +1,7 @@
 KAFKA_VERSION ?= 1.1
 PROJECT = brod
 PROJECT_DESCRIPTION = Kafka client library in Erlang
-PROJECT_VERSION = 3.7.5
+PROJECT_VERSION = 3.7.6
 
 DEPS = supervisor3 kafka_protocol
 

changelog.md

@@ -144,3 +144,5 @@
     `member_assignment` data field in group sync response can either be `null` (kafka 0.10)
     or a struct having empty `topic_partitions` (kafka 0.11 or later). The later case
     was not handled properly in `brod` before this fix.
+* 3.7.6
+  * Fix produce message input type spec to make it backward compatible (bug introduced in 3.7.3)

src/brod.app.src

@@ -1,7 +1,7 @@
 %% -*- mode:erlang -*-
 {application,brod,
  [{description,"Apache Kafka Erlang client library"},
-  {vsn,"3.7.5"},
+  {vsn,"3.7.6"},
   {registered,[]},
   {applications,[kernel,stdlib,kafka_protocol,supervisor3]},
   {env,[]},

src/brod.erl

@@ -168,6 +168,8 @@
 -type value() :: undefined %% no value, transformed to <<>>
                | iodata() %% single value
                | {msg_ts(), binary()} %% one message with timestamp
+               | ?KV(key(), value()) %% backward compatible
+               | ?TKV(msg_ts(), key(), value()) %% backward compatible
                | kpro:msg_input() %% one magic v2 message
                | kpro:batch_input(). %% maybe nested batch
 

src/brod_consumer.erl

@@ -489,7 +489,7 @@ handle_fetch_error(#kafka_fetch_error{error_code = ErrorCode} = Error,
                          } = State) ->
   case err_op(ErrorCode) of
     reset_connection ->
-      error_logger:info_msg("Fetch error ~p-~p: ~p",
+      error_logger:info_msg("Fetch error ~s-~p: ~p",
                             [Topic, Partition, ErrorCode]),
       %% The current connection in use is not connected to the partition leader,
       %% so we dereference and demonitor the connection pid, but leave it alive,
@@ -506,7 +506,7 @@ handle_fetch_error(#kafka_fetch_error{error_code = ErrorCode} = Error,
       {noreply, maybe_send_fetch_request(State)};
     stop ->
       ok = cast_to_subscriber(Subscriber, Error),
-      error_logger:error_msg("Consumer ~p-~p shutdown\nReason: ~p",
+      error_logger:error_msg("Consumer ~s-~p shutdown\nReason: ~p",
                              [Topic, Partition, ErrorCode]),
       {stop, normal, State};
     reset_offset ->

src/brod_group_coordinator.erl

@@ -837,7 +837,7 @@ roundrobin_assign_loop([{Topic, Partition} | Rest] = TopicPartitions,
 -spec get_topic_assignments(state(), binary() | [kpro:struct()]) ->
         brod:received_assignments().
 get_topic_assignments(#state{}, ?kpro_cg_no_assignment) -> [];
-get_topic_assignments(#state{} = State, #{topic_partitions := []}) -> [];
+get_topic_assignments(#state{}, #{topic_partitions := []}) -> [];
 get_topic_assignments(#state{} = State, Assignment) ->
   PartitionAssignments = kpro:find(topic_partitions, Assignment),
   TopicPartitions0 =

src/brod_sup.erl

@@ -87,7 +87,6 @@
 %%    , [ {client_1 %% unique client ID
 %%        , [ {endpoints, [{"localhost", 9092}]}
 %%          , {restart_delay_seconds, 10}
-%%          , {max_metadata_sock_retry, 1}
 %%          , {get_metadata_timeout_seconds, 5}
 %%          , {reconnect_cool_down_seconds, 1}
 %%          , {allow_topic_auto_creation, true}