Current section
17 Versions
Jump to
Current section
17 Versions
Compare versions
4
files changed
+16
additions
-6
deletions
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"kafe">>}. |
| 2 | - {<<"version">>,<<"2.1.3">>}. |
| 2 | + {<<"version">>,<<"2.1.4">>}. |
| 3 3 | {<<"requirements">>, |
| 4 4 | #{<<"bristow">> => #{<<"app">> => <<"bristow">>, |
| 5 5 | <<"optional">> => false, |
| @@ -4,7 +4,7 @@ defmodule Kafe.Mixfile do | |
| 4 4 | def project do |
| 5 5 | [ |
| 6 6 | app: :kafe, |
| 7 | - version: "2.1.3", |
| 7 | + version: "2.1.4", |
| 8 8 | elixir: "~> 1.2", |
| 9 9 | build_embedded: Mix.env == :prod, |
| 10 10 | start_permanent: Mix.env == :prod, |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,kafe, |
| 2 2 | [{description,"A Kafka client for Erlang and Elixir"}, |
| 3 | - {vsn,"2.1.3"}, |
| 3 | + {vsn,"2.1.4"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {modules,[]}, |
| 6 6 | {applications,[kernel,stdlib,syntax_tools,compiler,poolgirl, |
| @@ -170,7 +170,7 @@ fetch(#state{topic = Topic, | |
| 170 170 | [#{error_code := ErrorCode, |
| 171 171 | partition := Partition}]}]}} -> |
| 172 172 | lager:error("Error fetching offset ~p of topic ~s, partition ~p: ~s", [Offset + 1, Topic, Partition, kafe_error:message(ErrorCode)]), |
| 173 | - get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset); |
| 173 | + get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset, GroupID); |
| 174 174 | Error -> |
| 175 175 | lager:error("Error fetching offset ~p of topic ~s, partition ~p: ~p", [Offset + 1, Topic, Partition, Error]), |
| 176 176 | Error |
| @@ -179,14 +179,21 @@ fetch(#state{topic = Topic, | |
| 179 179 | {ok, Offset} |
| 180 180 | end. |
| 181 181 | |
| 182 | - get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset) -> |
| 182 | + get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset, GroupID) -> |
| 183 183 | case maps:get(ErrorCode, ErrorsActions, maps:get('*', ErrorsActions, undefined)) of |
| 184 184 | undefined -> |
| 185 185 | {error, ErrorCode}; |
| 186 186 | error -> |
| 187 187 | {error, ErrorCode}; |
| 188 188 | {call, {Module, Function, Args}} -> |
| 189 | - erlang:apply(Module, Function, Args); |
| 189 | + try |
| 190 | + erlang:apply(Module, Function, [Topic, Partition, Offset, GroupID|Args]) |
| 191 | + catch |
| 192 | + Class:Error -> |
| 193 | + lager:error("Error calling ~s:~s/~p, topic ~s, partition ~p, offset ~p, group ~s, args ~p", |
| 194 | + [Module, Function, Topic, Partition, Offset, GroupID, Args]), |
| 195 | + stop |
| 196 | + end; |
| 190 197 | {retry, Time} -> |
| 191 198 | {wait, Time}; |
| 192 199 | stop -> |
| @@ -196,10 +203,12 @@ get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset) -> | |
| 196 203 | retry -> |
| 197 204 | {ok, Offset}; |
| 198 205 | next -> |
| 206 | + commit(Offset, Topic, Partition, GroupID), |
| 199 207 | {ok, Offset + 1}; |
| 200 208 | {reset, earliest} -> |
| 201 209 | case get_partition_offset(Topic, Partition, -2) of |
| 202 210 | {ok, _} = Offset1 -> |
| 211 | + commit(Offset1 - 1, Topic, Partition, GroupID), |
| 203 212 | Offset1; |
| 204 213 | _ -> |
| 205 214 | {error, internal_error} |
| @@ -207,6 +216,7 @@ get_error_action(ErrorCode, ErrorsActions, Topic, Partition, Offset) -> | |
| 207 216 | {reset, latest} -> |
| 208 217 | case get_partition_offset(Topic, Partition, -1) of |
| 209 218 | {ok, _} = Offset1 -> |
| 219 | + commit(Offset1 - 1, Topic, Partition, GroupID), |
| 210 220 | Offset1; |
| 211 221 | _ -> |
| 212 222 | {error, internal_error} |