Merge pull request #406 from kafkaex/handle_tcp_close

Handle closed sockets in new client

lib/kafka_ex/new/adapter.ex

@@ -98,7 +98,19 @@ defmodule KafkaEx.New.Adapter do
           }
         ]
       }) do
-    base_offset
+    {:ok, base_offset}
+  end
+
+  def produce_response(%{
+        responses: [
+          %{
+            partition_responses: [
+              %{error_code: error_code}
+            ]
+          }
+        ]
+      }) do
+    {:error, Kayrock.ErrorCode.code_to_atom(error_code)}
   end
 
   def metadata_response(cluster_metadata) do

lib/kafka_ex/new/broker.ex

@@ -9,8 +9,7 @@ defmodule KafkaEx.New.Broker do
             host: nil,
             port: nil,
             socket: nil,
-            rack: nil,
-            socket: nil
+            rack: nil
 
   @type t :: %__MODULE__{}
 

lib/kafka_ex/new/client.ex

@@ -180,6 +180,16 @@ defmodule KafkaEx.New.Client do
     {:noreply, update_metadata(state)}
   end
 
+  def handle_info({:tcp_closed, socket}, state) do
+    state_out = close_broker_by_socket(state, socket)
+    {:noreply, state_out}
+  end
+
+  def handle_info({:ssl_closed, socket}, state) do
+    state_out = close_broker_by_socket(state, socket)
+    {:noreply, state_out}
+  end
+
   @impl true
   def terminate(reason, state) do
     Logger.log(
@@ -222,9 +232,8 @@ defmodule KafkaEx.New.Client do
         for broker <- brokers_to_close do
           Logger.log(
             :debug,
-            "Closing connection to broker #{broker.node_id}: #{
-              inspect(broker.host)
-            } on port #{inspect(broker.port)}"
+            "Closing connection to broker #{broker.node_id}: " <>
+              "#{inspect(broker.host)} on port #{inspect(broker.port)}"
           )
 
           NetworkClient.close_socket(broker.socket)
@@ -688,4 +697,19 @@ defmodule KafkaEx.New.Client do
     {topic_metadata,
      %{updated_state | allow_auto_topic_creation: allow_auto_topic_creation}}
   end
+
+  defp close_broker_by_socket(state, socket) do
+    State.update_brokers(state, fn broker ->
+      if broker.socket.socket == socket do
+        Logger.log(
+          :debug,
+          "Broker #{inspect(broker.host)}:#{inspect(broker.port)} closed connection"
+        )
+
+        Broker.put_socket(broker, nil)
+      else
+        broker
+      end
+    end)
+  end
 end

lib/kafka_ex/new/client_compatibility.ex

@@ -78,7 +78,7 @@ defmodule KafkaEx.New.ClientCompatibility do
         response =
           case response do
             {:ok, :ok} -> {:ok, :ok}
-            {:ok, val} -> {:ok, Adapter.produce_response(val)}
+            {:ok, val} -> Adapter.produce_response(val)
             _ -> response
           end
 

test/integration/new_client_test.exs

@@ -101,4 +101,22 @@ defmodule KafkaEx.New.Client.Test do
     {:ok, offset_after} = KafkaExAPI.latest_offset(client, topic, partition)
     assert offset_after == offset_before + 3
   end
+
+  test "client can receive {:ssl_closed, _}", %{client: client} do
+    send(client, {:ssl_closed, :unused})
+    TestHelper.wait_for(fn ->
+      {:message_queue_len, m} = Process.info(client, :message_queue_len)
+      m == 0
+    end)
+    assert Process.alive?(client)
+  end
+
+  test "client can receive {:tcp_closed, _}", %{client: client} do
+    send(client, {:tcp_closed, :unused})
+    TestHelper.wait_for(fn ->
+      {:message_queue_len, m} = Process.info(client, :message_queue_len)
+      m == 0
+    end)
+    assert Process.alive?(client)
+  end
 end