Close socket on timeout/error

[ukrbublik]

Relates to #346

I was testing kafka_ex on Kafka server with not very good connection, which frequently gives :timeout on network requests (#310) with default params (can be fixed with sync_timeout=3_000).
And during several hours of test I noticed errors in log (see below).
Looks like it tries to parse response on different request (like tries to parse metadata response like it's fetch response etc.)
It can be if socket is not properly closed.
Tested my fix on same environment several hours, no more such errors (just timeouts)

** (RuntimeError) Parse error during KafkaEx.Protocol.Fetch.parse_response. Couldn't parse: <<0, 0, 2, 30, 0, 0, 0, 3, 0, 0, 3, 233, 0, 42, 115, 116, 103, 45, 117, 115, 119, 50, 45, 108, 111, 99, 97, 108, 112, 105, 112, 101, 45, 107, 97, 102, 107, 97, 48, 52, 46, 112, 100, 45, 105, 110, 116, 101, 114, 110, ...>>
    (kafka_ex) lib/kafka_ex/protocol/fetch.ex:86: KafkaEx.Protocol.Fetch.parse_partitions(65601578, <<115, 116, 103, 45, 117, 115, 119, 50, 45, 108, 111, 99, 97, 108, 112, 105, 112, 101, 45, 107, 97, 102, 107, 97, 48, 52, 46, 112, 100, 45, 105, 110, 116, 101, 114, 110, 97, 108, 46, 99, 111, 109, 0, 0, 35, 132, 0, 10, 117, 115, ...>>, [])
    (kafka_ex) lib/kafka_ex/protocol/common.ex:25: KafkaEx.Protocol.Common.parse_topics/3
    (kafka_ex) lib/kafka_ex/server.ex:857: KafkaEx.Server0P8P2.network_request/3
    (kafka_ex) lib/kafka_ex/server_0_p_8_p_2.ex:259: KafkaEx.Server0P8P2.fetch/2
    (kafka_ex) lib/kafka_ex/server_0_p_8_p_2.ex:117: KafkaEx.Server0P8P2.kafka_server_fetch/2
    (stdlib) gen_server.erl:661: :gen_server.try_handle_call/4
    (stdlib) gen_server.erl:690: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3

** (FunctionClauseError) no function clause matching in KafkaEx.Protocol.Metadata.parse_brokers/4
    (kafka_ex) lib/kafka_ex/protocol/metadata.ex:215: KafkaEx.Protocol.Metadata.parse_brokers(1, 1, <<0, 25, 101, 115, 99, 97, 108, 97, 116, 105, 111, 110, 95, 112, 111, 108, 105, 99, 121, 95, 117, 112, 100, 97, 116, 101, 115, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 77, 0, 0, 0, 0>>, [])
    (kafka_ex) lib/kafka_ex/protocol/metadata.ex:194: KafkaEx.Protocol.Metadata.parse_response/2
    (kafka_ex) lib/kafka_ex/server.ex:700: KafkaEx.Server0P10AndLater.retrieve_metadata_with_version/7
    (kafka_ex) lib/kafka_ex/server.ex:590: KafkaEx.Server0P10AndLater.update_metadata/1
    (kafka_ex) lib/kafka_ex/server_0_p_10_and_later.ex:170: KafkaEx.Server0P10AndLater.kafka_server_update_metadata/1
    (stdlib) gen_server.erl:637: :gen_server.try_dispatch/4
    (stdlib) gen_server.erl:711: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3

        ** (FunctionClauseError) no function clause matching in KafkaEx.Protocol.Heartbeat.parse_response/1
            (kafka_ex) lib/kafka_ex/protocol/heartbeat.ex:34: KafkaEx.Protocol.Heartbeat.parse_response(<<0, 0, 0, 19, 0, 0, 0, 1, 0, 0, 0, 0, 0, 9, 49, 50, 55, 46, 48, 46, 48, 46, 49, 0, 0, 35, 132, 255, 255, 0, 0, 0, 0, 0, 0, 0, 10, 0, 0, 0, 4, 116, 101, 115, 116, 0, 0, 0, 0, 1, ...>>)
            (kafka_ex) lib/kafka_ex/server_0_p_9_p_0.ex:231: KafkaEx.Server0P9P0.consumer_group_sync_request/4
            (kafka_ex) lib/kafka_ex/server_0_p_9_p_0.ex:186: KafkaEx.Server0P9P0.kafka_server_heartbeat/3
            (stdlib) gen_server.erl:636: :gen_server.try_handle_call/4
            (stdlib) gen_server.erl:665: :gen_server.handle_msg/6
            (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3

** (FunctionClauseError) no function clause matching in KafkaEx.Protocol.Metadata.parse_topic_metadatas_v1/2
    (kafka_ex) lib/kafka_ex/protocol/metadata.ex:290: KafkaEx.Protocol.Metadata.parse_topic_metadatas_v1(825374510, <<48, 46, 48, 46, 49, 0, 0, 35, 132>>)
    (kafka_ex) lib/kafka_ex/protocol/metadata.ex:201: KafkaEx.Protocol.Metadata.parse_response/2
    (kafka_ex) lib/kafka_ex/server.ex:700: KafkaEx.Server0P10AndLater.retrieve_metadata_with_version/7
    (kafka_ex) lib/kafka_ex/server.ex:590: KafkaEx.Server0P10AndLater.update_metadata/1
    (kafka_ex) lib/kafka_ex/server_0_p_10_and_later.ex:170: KafkaEx.Server0P10AndLater.kafka_server_update_metadata/1
    (stdlib) gen_server.erl:616: :gen_server.try_dispatch/4
    (stdlib) gen_server.erl:686: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3

and :erlang.hd([]) error like in #346

[sourcelevel-bot]

Ebert has finished reviewing this Pull Request and has found:

• 10 fixed issues! 🎉

But beware that this branch is 1 commit behind the kafkaex:master branch, and a review of an up to date branch would produce more accurate results.

You can see more details about this review at https://ebertapp.io/github/kafkaex/kafka_ex/pulls/351.

[joshuawscott]

This looks good to me, thanks!