MetadataResponse received by _handle_group_coordinator_response

[dpkp]

I saw this error during local testing. It looks like there is a thread synchronization issue that leads to responses being processed incorrectly (here a MetadataResponse was received by a group coordinator request future).

1514315653.454923 conn.py                 MainThread-heartbeat Connect attempt to <BrokerConnection node_id=0 host=localhost/127.0.0.1 port=65231> returned error 61. Disconnecting.
1514315653.455129 client_async.py         MainThread-heartbeat Node 0 connection failed -- refreshing metadata
1514315653.583284 client_async.py         MainThread-heartbeat Node 1 connection failed -- refreshing metadata
Exception in thread MainThread-heartbeat:
Traceback (most recent call last):
  File "/usr/local/Cellar/python3/3.6.3/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/Users/dpowers/Code/kafka-python/kafka/coordinator/base.py", line 891, in run
    self._run_once()
  File "/Users/dpowers/Code/kafka-python/kafka/coordinator/base.py", line 922, in _run_once
    if not self.coordinator.lookup_coordinator().is_done:
  File "/Users/dpowers/Code/kafka-python/kafka/coordinator/base.py", line 284, in lookup_coordinator
    self._find_coordinator_future = self._send_group_coordinator_request()
  File "/Users/dpowers/Code/kafka-python/kafka/coordinator/base.py", line 650, in _send_group_coordinator_request
    _f.add_callback(self._handle_group_coordinator_response, future)
  File "/Users/dpowers/Code/kafka-python/kafka/future.py", line 52, in add_callback
    self._call_backs('callback', [f], self.value)
  File "/Users/dpowers/Code/kafka-python/kafka/future.py", line 83, in _call_backs
    raise e
  File "/Users/dpowers/Code/kafka-python/kafka/future.py", line 79, in _call_backs
    f(value)
  File "/Users/dpowers/Code/kafka-python/kafka/coordinator/base.py", line 657, in _handle_group_coordinator_response
    error_type = Errors.for_code(response.error_code)
AttributeError: 'MetadataResponse_v0' object has no attribute 'error_code'

[dpkp]

I can only reproduce this error on PR #1332 , so I suspect it is caused by returning multiple responses from conn.recv

[dpkp]

Looking deeper, I think this is a test-only issue that is caused by mocking BrokerConnection in conftest.py:

    conn.send.return_value = Future().success(
        MetadataResponse[0](
            [(0, 'foo', 12), (1, 'bar', 34)],  # brokers
            []))  # topics

Any coordinator / consumer tests that use this mocked connection and do not also mock the new background heartbeat thread will see the mismatched response during background coordinator lookups.

[dpkp]

I can't find any single test that causes this issue. What appears to happen is that an integration test runs with a heartbeat thread and that thread continues to run after the test completes. A later test mocks BrokerConnection.send and this causes the heartbeat thread to receive an unexpected mocked MetadataResponse to a group coordinator lookup. The underlying issue here may be that the heartbeat thread is not properly shutdown via a close() or __del__ destructor.