code: def test_recv_disconnected(_socket, conn):
      conn.connect()
      assert conn.connected()
  
      req = MetadataRequest[0]([])
      header = RequestHeader(req, client_id=conn.config["client_id"])
      payload_bytes = len(header.encode()) + len(req.encode())
      _socket.send.side_effect = [4, payload_bytes]
      conn.send(req)
  
      # Empty data on recv means the socket is disconnected
      _socket.recv.return_value = b""
  
      # Attempt to receive should mark connection as disconnected
      assert conn.connected()
      conn.recv()
      assert conn.disconnected()
  
imports: from __future__ import absolute_import
  from errno import EALREADY, EINPROGRESS, EISCONN, ECONNRESET
  import socket
  import mock
  import pytest
  from kafka.conn import BrokerConnection, ConnectionStates, collect_hosts
  from kafka.protocol.api import RequestHeader
  from kafka.protocol.metadata import MetadataRequest
  from kafka.protocol.produce import ProduceRequest
  import kafka.errors
label_desc: Use image node_id 17 for calling a global function with example usage: test_recv_disconnected(_socket, conn) without return types