Parity

rel/kafkerl.app.config

@@ -1,6 +1,6 @@
 [{lager, [{colored, true},
           {handlers, [{lager_console_backend, [debug,true]}]}]},
- {kafkerl, [{gen_server_name, kafkerl_client},
+ {kafkerl, [%{gen_server_name, kafkerl_client},
             {conn_config, [{brokers, [{"localhost", 9092}]},
                            {client_id, kafkerl_client},
                            {flush_every, 60},
@@ -9,5 +9,5 @@
                            {max_metadata_retries, -1},
                            {retry_on_topic_error, true},
                            {metadata_tcp_timeout, 1000}]},
-            {topics, [test1, test2, test3, test4]},
+            {topics, [test1, test2, test3]},
             {buffer, [{max_queue_size, 4}]}]}].

src/kafkerl.erl

@@ -4,7 +4,9 @@
 -export([start/0, start/2]).
 -export([produce/1, produce/2,
          get_partitions/0, get_partitions/1,
-         subscribe/1, subscribe/2]).
+         subscribe/1, subscribe/2,
+         unsubscribe/1, unsubscribe/2,
+         request_metadata/0, request_metadata/1, request_metadata/2]).
 
 -include("kafkerl.hrl").
 -include("kafkerl_consumers.hrl").
@@ -48,4 +50,16 @@ unsubscribe(Callback) ->
   unsubscribe(?MODULE, Callback).
 -spec unsubscribe(atom(), callback()) -> ok.
 unsubscribe(Name, Callback) ->
-  kafkerl_connector:unsubscribe(Name, Callback).
+  kafkerl_connector:unsubscribe(Name, Callback).
+
+-spec request_metadata() -> ok.
+request_metadata() ->
+  request_metadata(?MODULE).
+-spec request_metadata(atom() | [topic()]) -> ok.
+request_metadata(Name) when is_atom(Name) ->
+  kafkerl_connector:request_metadata(Name);
+request_metadata(Topics) ->
+  request_metadata(?MODULE, Topics).
+-spec request_metadata(atom(), [topic()]) -> ok.
+request_metadata(Name, Topics) ->
+  kafkerl_connector:request_metadata(Name, Topics).

src/kafkerl_broker_connection.erl

@@ -4,7 +4,7 @@
 -behaviour(gen_server).
 
 %% API
--export([send/2, flush/1]).
+-export([send/2, flush/1, kill/1]).
 % Only for internal use
 -export([connect/5]).
 % Supervisors
@@ -19,7 +19,8 @@
 -type conn_idx()            :: 0..1023.
 -type start_link_response() :: {ok, atom(), pid()} | ignore | {error, any()}.
 
--record(state, {conn_idx  = undefined :: conn_idx(),
+-record(state, {name      = undefined :: atom(),
+                conn_idx  = undefined :: conn_idx(),
                 client_id = undefined :: binary(),
                 socket    = undefined :: undefined | port(),
                 address   = undefined :: undefined | socket_address(),
@@ -38,8 +39,8 @@
 -spec start_link(conn_idx(), pid(), socket_address(), any()) ->
   start_link_response().
 start_link(Id, Connector, Address, Config) ->
-  Params = [Id, Connector, Address, Config],
   Name = list_to_atom(atom_to_list(?MODULE) ++ "_" ++ integer_to_list(Id)),
+  Params = [Id, Connector, Address, Config, Name],
   case gen_server:start_link({local, Name}, ?MODULE, Params, []) of
     {ok, Pid} ->
       {ok, Name, Pid};
@@ -55,14 +56,22 @@ send(ServerRef, Message) ->
 flush(ServerRef) ->
   gen_server:call(ServerRef, {flush}).
 
+-spec kill(server_ref()) -> ok.
+kill(ServerRef) ->
+  gen_server:call(ServerRef, {kill}).
+
 %%==============================================================================
 %% gen_server callbacks
 %%==============================================================================
 -spec handle_call(any(), any(), state()) -> {reply, ok, state()}.
 handle_call({send, Message}, _From, State) ->
   handle_send(Message, State);
 handle_call({flush}, _From, State) ->
-  handle_flush(State).
+  handle_flush(State);
+handle_call({kill}, _From, State = #state{name = Name}) ->
+  % TODO: handle the potentially buffered messages
+  lager:info("~p stopped by it's parent connector", [Name]),
+  {stop, normal, ok, State}.
 
 -spec handle_info(any(), state()) -> {noreply, state()}.
 handle_info({connected, Socket}, State) ->
@@ -92,17 +101,17 @@ code_change(_OldVsn, State, _Extra) -> {ok, State}.
 %%==============================================================================
 %% Handlers
 %%==============================================================================
-init([Id, Connector, Address, Config]) ->
+init([Id, Connector, Address, Config, Name]) ->
   Schema = [{tcp_options, [any], {default, []}},
             {retry_interval, positive_integer, {default, 1000}},
             {max_retries, positive_integer, {default, 3}},
             {client_id, binary, {default, <<"kafkerl_client">>}}],
   case normalizerl:normalize_proplist(Schema, Config) of
     {ok, [TCPOpts, RetryInterval, MaxRetries, ClientId]} ->
       NewTCPOpts = kafkerl_utils:get_tcp_options(TCPOpts),
-      State = #state{conn_idx = Id, address = Address, connector = Connector,
+      State = #state{conn_idx = Id, tcp_options = NewTCPOpts, address = Address,
                      max_retries = MaxRetries, retry_interval = RetryInterval,
-                     tcp_options = NewTCPOpts, client_id = ClientId},
+                     connector = Connector, client_id = ClientId, name = Name},
       Params = [self(), NewTCPOpts, Address, RetryInterval, MaxRetries],
       _Pid = spawn_link(?MODULE, connect, Params),
       {ok, State};
@@ -164,10 +173,10 @@ handle_tcp_data(Bin, State = #state{connector = Connector}) ->
         {ok, Messages} ->
           Errors = get_errors_from_produce_response(Topics),
           case handle_errors(Errors, Messages) of
-            {ignore, _} ->
+            ignore ->
               State;
             {request_metadata, MessagesToResend} ->
-              Connector ! force_metadata_request,
+              kafkerl_connector:request_metadata(Connector),
               F = fun(M) -> kafkerl_connector:send(Connector, M) end,
               ok = lists:foreach(F, MessagesToResend),
               State
@@ -197,38 +206,51 @@ build_correlation_id(State = #state{request_number = RequestNumber,
 get_errors_from_produce_response(Topics) ->
   get_errors_from_produce_response(Topics, []).
 
+get_errors_from_produce_response([], Acc) ->
+  lists:reverse(Acc);
 get_errors_from_produce_response([{Topic, Partitions} | T], Acc) ->
   Errors = [{Topic, Partition, Error} ||
             {Partition, Error, _} <- Partitions, Error =/= ?NO_ERROR],
   get_errors_from_produce_response(T, Acc ++ Errors).
 
-handle_errors([] = Errors, _Messages) ->
-  {ignore, Errors};
+handle_errors([], _Messages) ->
+  ignore;
 handle_errors(Errors, Messages) ->
-  F = fun(E, Acc) -> handle_error(E, Messages, Acc) end,
-  lists:foreach(F, {ignore, []}, Errors).
+  F = fun(E) -> handle_error(E, Messages) end,
+  case lists:filtermap(F, Errors) of
+    [] -> ignore;
+    L  -> {request_metadata, L}
+  end.
 
-handle_error({Topic, Partition, Error}, Messages, {_Status, Acc} = Status)
+handle_error({Topic, Partition, Error}, Messages)
   when Error =:= ?UNKNOWN_TOPIC_OR_PARTITION orelse
        Error =:= ?LEADER_NOT_AVAILABLE orelse
        Error =:= ?NOT_LEADER_FOR_PARTITION ->
   case get_message_for_error(Topic, Partition, Messages) of
-    undefined -> Status;
-    Message   -> {request_metadata, [Message | Acc]}
+    undefined -> false;
+    Message   -> {true, Message}
   end;
-handle_error({Topic, Partition, Error}, _Messages, Acc) ->
+handle_error({Topic, Partition, Error}, _Messages) ->
   lager:error("Unable to handle ~p error on topic ~p, partition ~p",
               [kafkerl_error:get_error_name(Error), Topic, Partition]),
-  Acc.
-
-get_message_for_error(Topic, Partition, []) ->
-  lager:error("no saved message found for error on topic ~p, partition ~p",
-              [Topic, Partition]),
-  undefined;
-get_message_for_error(Topic, Partition, [{Topic, Partition, _} = H | _T]) ->
-  H;
-get_message_for_error(Topic, Partition, [_Message | T]) ->
-  get_message_for_error(Topic, Partition, T).
+  false.
+
+get_message_for_error(Topic, Partition, SavedMessages) ->
+  case lists:keyfind(Topic, 1, SavedMessages) of
+    false ->
+      lager:error("No saved messages found for topic ~p, partition ~p",
+                  [Topic, Partition]),
+      undefined;
+    {Topic, Partitions} ->
+      case lists:keyfind(Partition, 1, Partitions) of
+        false -> 
+          lager:error("No saved messages found for topic ~p, partition ~p",
+                      [Topic, Partition]),
+          undefined;
+        {Partition, Messages} ->
+          {Topic, Partition, Messages}
+      end
+  end.
 
 connect(Pid, _TCPOpts, {Host, Port} = _Address, _Timeout, 0) ->
   lager:error("Unable to connect to ~p:~p", [Host, Port]),

src/kafkerl_buffer.erl

@@ -132,8 +132,8 @@ handle_get_saved_request(CorrelationId, State) ->
   {reply, Response, State}.
 
 handle_delete_saved_request(CorrelationId,
-                            State = #state{message_buffer = MessageBuffer}) ->
-  case lists:keytake(CorrelationId, 1, MessageBuffer) of
+                            State = #state{saved_requests = Requests}) ->
+  case lists:keytake(CorrelationId, 1, Requests) of
     false ->
       {reply, {error, not_found}, State};
     {value, {CorrelationId, _Bin, Messages}, NewMessageBuffer} ->