Librdkafka: Transient consumer failures with "GroupCoordinator response error: Local: Broker transport failure"

[dblia]

Description

We have an application consisting of a consumer group of 4 consumers, reading from a topic of a PartitionCount of 4 and a ReplicationFactor of 3. The topic receives at around ~700 messages per second, so the consumers are always busy. Our kafka cluster has 4 broker nodes.

After upgrading to librdkafka v0.11.4 we've noticed some transient failures of some of the consumer processes. The same issue persists after upgrading to librdkafka v0.11.5, too.

The error with which the consumers fail is the following:

cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="GroupCoordinator response error: Local: Broker transport failure"}

_Notes:_

• The current application uses the official confluent-kafka-python client but we also seen it occur with a simple custom Ruby client we've implemented here
  
  
  
  • Currently we auto-start the failed consumer process as soon as it fails until we found the cause of those random failures
  
  
  • There are some similar (closed) issues but since they have some differences with our case, I chose to open a new issue
  
  

The respective consumer log of a failed consumer process (named consumer-process-4) where we've set the debug option to all has been pasted in the following Gist, mainly for readability reasons.

AFAICT, it seems that the application connects to a random single broker (kafka-a.example.com for our case) and asks for the group coordinator (every 10mins by default). The broker sends the coordinator query request with CorrId 1639 and then it fails, without getting the response. The
consumer group receives a TERMINATE signal immediately, the consumer that made the query fails and a rebalance operation is triggered. The rest consumers continue to operate normally, and since we auto-restart the failed consumer, it returns in less than a second to its normal operation mode.

An interesting point is that the disconnected kafka-a broker, was disconnected most probably by the idle connection reaper 2 minutes ago:

Sep  5 09:02:12 fqdn consumer-process-3[22622]: %7|1536127332.164|BROKERFAIL|confluent-kafka-python_3#consumer-1| [thrd:kafka-a.example.com:9092/bootstrap]: kafka-a.example.com:9092/1: failed: err: Local: Broker transport failure: (errno: Resource temporarily unavailable)
Sep  5 09:02:12 fqdn consumer-process-3[22622]: %7|1536127332.165|FAIL|confluent-kafka-python_3#consumer-1| [thrd:kafka-a.example.com:9092/bootstrap]: kafka-a.example.com:9092/1: Disconnected
Sep  5 09:02:12 fqdn consumer-process-3[22622]: %7|1536127332.165|STATE|confluent-kafka-python_3#consumer-1| [thrd:kafka-a.example.com:9092/bootstrap]: kafka-a.example.com:9092/1: Broker changed state UP -> DOWN
Sep  5 09:02:12 fqdn consumer-process-3[22622]: %7|1536127332.165|BROADCAST|confluent-kafka-python_3#consumer-1| [thrd:kafka-a.example.com:9092/bootstrap]: Broadcasting state chang

So, IMHO it can't be the idle connection reaper who kills the broker since we use the default value (10 minutes). Also, if it's helpful the group coordinator is kafka-c.example.com.

If you need a bigger part of the log file, I could included it

I'm also including a snippet of the broker's server logs of kafka-a.example.com which seems normal. The log output is the same for the rest brokers, too:

[2018-09-05 09:04:10,534] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2, dir=/srv/kafka/sdc/data] Incrementing log start offset to 140778 (kafka.log.Log)
[2018-09-05 09:04:10,551] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5, dir=/srv/kafka/sdc/data] Incrementing log start offset to 124782 (kafka.log.Log)
[2018-09-05 09:04:10,554] INFO Cleared earliest 0 entries from epoch cache based on passed offset 140778 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:10,572] INFO Cleared earliest 0 entries from epoch cache based on passed offset 124782 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:11,176] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-3, dir=/srv/kafka/sdb/data] Incrementing log start offset to 186306 (kafka.log.Log)
[2018-09-05 09:04:11,191] INFO Cleared earliest 0 entries from epoch cache based on passed offset 186306 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-3 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:11,373] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-4, dir=/srv/kafka/sdc/data] Incrementing log start offset to 145664 (kafka.log.Log)
[2018-09-05 09:04:11,396] INFO Cleared earliest 0 entries from epoch cache based on passed offset 145664 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-4 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:12,196] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-6, dir=/srv/kafka/sdb/data] Incrementing log start offset to 136681 (kafka.log.Log)
[2018-09-05 09:04:12,215] INFO Cleared earliest 0 entries from epoch cache based on passed offset 136681 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-6 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:12,676] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2, dir=/srv/kafka/sdc/data] Incrementing log start offset to 140779 (kafka.log.Log)
[2018-09-05 09:04:12,681] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5, dir=/srv/kafka/sdc/data] Incrementing log start offset to 124783 (kafka.log.Log)
[2018-09-05 09:04:12,699] INFO Cleared earliest 0 entries from epoch cache based on passed offset 140779 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:12,718] INFO Cleared earliest 0 entries from epoch cache based on passed offset 124783 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:13,163] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-1, dir=/srv/kafka/sdb/data] Incrementing log start offset to 140955 (kafka.log.Log)
[2018-09-05 09:04:13,182] INFO Cleared earliest 0 entries from epoch cache based on passed offset 140955 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-1 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:14,247] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-6, dir=/srv/kafka/sdb/data] Incrementing log start offset to 136684 (kafka.log.Log)
[2018-09-05 09:04:14,271] INFO Cleared earliest 0 entries from epoch cache based on passed offset 136684 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-6 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:14,675] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2, dir=/srv/kafka/sdc/data] Incrementing log start offset to 140781 (kafka.log.Log)
[2018-09-05 09:04:14,677] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5, dir=/srv/kafka/sdc/data] Incrementing log start offset to 124785 (kafka.log.Log)
[2018-09-05 09:04:14,707] INFO Cleared earliest 0 entries from epoch cache based on passed offset 124785 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-5 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:14,707] INFO Cleared earliest 0 entries from epoch cache based on passed offset 140781 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-2 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:15,249] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-1, dir=/srv/kafka/sdb/data] Incrementing log start offset to 140957 (kafka.log.Log)
[2018-09-05 09:04:15,272] INFO Cleared earliest 0 entries from epoch cache based on passed offset 140957 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-1 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:15,357] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-3, dir=/srv/kafka/sdb/data] Incrementing log start offset to 186308 (kafka.log.Log)
[2018-09-05 09:04:15,376] INFO Cleared earliest 0 entries from epoch cache based on passed offset 186308 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-3 (kafka.server.epoch.LeaderEpochFileCache)
[2018-09-05 09:04:15,514] INFO [Log partition=_confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-4, dir=/srv/kafka/sdc/data] Incrementing log start offset to 145665 (kafka.log.Log)
[2018-09-05 09:04:15,536] INFO Cleared earliest 0 entries from epoch cache based on passed offset 145665 leaving 1 in EpochFile for partition _confluent-ksql-production_query_CTAS_CLICKS_PER_HOUR_0-KSTREAM-AGGREGATE-STATE-STORE-0000000007-repartition-4 (kafka.server.epoch.LeaderEpochFileCache)

Are there any ideas about this behavior?

How to reproduce

By reducing the broker's connections.max.idle.ms setting in order to make the brokers disconnect more often and also reducing the librdkafka's coordinator.query.interval.ms value to increase the coordinator query count in an experimental cluster, I've managed to see this issue arise more often.

However, I can't say that I've managed to repeatedly reproduce this issue, yet. In our production cluster, the failure occurs randomly but quite often, i.e., 2-3 times a week.

Checklist

• [x] librdkafka version: v0.11.4 and v0.11.5
• [x] Apache Kafka version: 1.1.1
• [x] librdkafka client configuration:

{
  'enable.auto.offset.store': True,
  'retries': 2,
  'default.topic.config': { 'auto.offset.reset': 'earliest' },
  'bootstrap.servers': '....',
  'debug': 'all',
  'log.connection.close': False,
  'group.id': 'test-consumer-group',
  'enable.auto.commit': True,
  'auto.commit.interval.ms': 5000
}

• [x] Operating system: Linux 4.9.0-6-amd64 #1 SMP Debian 4.9.88-1+deb9u1 (2018-05-07) x86_64 GNU/Linux
• [x] Provide logs from librdkafka: debug=all
• [x] Provide broker log excerpts
• [ ] Critical issue

[edenhill]

Great bug report!

There's something in the log I don't really understand:

Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CGRPCOORD|confluent-kafka-python_4#consumer-1| [thrd:main]: kafka-a.example.com:9092/1: Group "test-consumer-group" GroupCoordinator response error: Local: Broker transport failure
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|METADATA|confluent-kafka-python_4#consumer-1| [thrd:kafka-a.example.com:9092/bootstrap]: kafka-c.example.com:9092/3: Request metadata for 1 topic(s): broker down
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CGRPSTATE|confluent-kafka-python_4#consumer-1| [thrd:main]: Group "test-consumer-group" changed state up -> query-coord (v14, join-state started)
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|BROADCAST|confluent-kafka-python_4#consumer-1| [thrd:main]: Broadcasting state change
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|SEND|confluent-kafka-python_4#consumer-1| [thrd:kafka-c.example.com:9092/bootstrap]: kafka-c.example.com:9092/3: Sent MetadataRequest (v2, 59 bytes @ 0, CorrId 622942)
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|RECV|confluent-kafka-python_4#consumer-1| [thrd:kafka-c.example.com:9092/bootstrap]: kafka-c.example.com:9092/3: Received HeartbeatResponse (v0, 2 bytes, CorrId 622941, rtt 0.37ms)
Sep  5 09:04:12 fqdn consumer-process-4[22649]: consumer-process-4[22649]: KafkaError{code=_TRANSPORT,val=-195,str="GroupCoordinator response error: Local: Broker transport failure"}
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CLOSE|confluent-kafka-python_4#consumer-1| [thrd:app]: Closing consumer
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CGRPOP|confluent-kafka-python_4#consumer-1| [thrd:main]: Group "test-consumer-group" received op TERMINATE (v0) in state query-coord (join state started, v14 vs 0)

Right after the GroupCoordinator request fails (likely due to the idle reaper), the consumer group terminates.
But there is no such code in the client itself, it will just reset its cgrp state and try to acquire a new coordinator.
The logical explanation is that it is triggered by the Python application closing the consumer, but the exception is logged a couple of seconds later:

Sep  5 09:04:14 fqdn consumer-process-4[22649]: cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="GroupCoordinator response error: Local: Broker transport failure"}
Sep  5 09:04:15 fqdn systemd[1]: [email protected]: Main process exited, code=exited, status=1/FAILURE

Is there anything in your code that closes the consumer prior to that exception?

[dblia]

Right after the GroupCoordinator request fails (likely due to the idle reaper), the consumer group terminates.
But there is no such code in the client itself, it will just reset its cgrp state and try to acquire a new coordinator.
The logical explanation is that it is triggered by the Python application closing the consumer, but the exception is logged a couple of seconds later:

Apparently you are right. The following snippet is an abstract view of our consumer process:

while True:
    msg = consumer.poll(timeout=timeout)

    if msg is None or msg.value() is None:
        continue
    if msg.error():
        if msg.error().code() == kafka.KafkaError._PARTITION_EOF:
            continue
        logger.error(msg.error())
        raise kafka.KafkaException(msg.error())

    yield d_msg

So from the following output, we can safely assume that the TERMINATE signal is sent by the application which first logs the unknown error:

Sep  5 09:04:12 fqdn consumer-process-4[22649]: consumer-process-4[22649]: KafkaError{code=_TRANSPORT,val=-195,str="GroupCoordinator response error: Local: Broker transport failure"}
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CLOSE|confluent-kafka-python_4#consumer-1| [thrd:app]: Closing consumer
Sep  5 09:04:12 fqdn consumer-process-4[22649]: %7|1536127452.655|CGRPOP|confluent-kafka-python_4#consumer-1| [thrd:main]: Group "test-consumer-group" received op TERMINATE (v0) in state query-coord (join state started, v14 vs 0)

The exception stacktrace is logged a couple of seconds later as you said, which also seems normal to me based on the debug log, since there's a "clean-up" phase in between, resources are freed, etc, etc.

---

However, from the above conversation it seems that there may be a case when the idle connection reaper may "kill" a broker connection at exactly the same time a consumer process makes a GroupCoordinator request to it. In that case the poll method will return the error message to the application (let's say similar to the _PARTITION_EOF_ message).

Firstly, does my explanation above seems reasonable to you? And if so, should such messages reach the application by the poll method and subsequently handled by the application (probably re-poll?), or should be just library-agnostic and never exposed to the apps?

[agis]

@edenhill any updates on this? Any plans for librdkafka to handle those kind of errors transparently? Or we should just ignore them in the application?

[edenhill]

As a general rule for librdkafka-based clients:
given that the cluster and client are correctly configured, all errors can be ignored as they are most likely temporary and librdkafka will attempt to recover automatically.

In this specific case; if a group coordinator request fails it will be retried (using any broker in state Up) within 500ms. The current assignment and group membership will not be affected, if a new coordinator is found before the missing heartbeats times out the membership (session.timeout.ms).
Auto offset commits will be stalled until a new coordinator is found.

In a future version we'll extend the error type to include a severity, allowing applications to happily ignore non-terminal errors. At this time an application should consider all errors informational, and not terminal.

[dblia]

Hello @edenhill. Treating similar errors as informational and not as terminal, as you suggested, let librdkafka client recover automatically without affecting the application. The extension of the error type to include a severity level would be a nice addition, though.

Thank you for you help, I'm closing this one since it has been resolved.