Phobos issues in 0.5.x of Ruby Kafka

[klippx]

We have an issue with ruby kafka >= 0.5.0 in Phobos that we might need your help to investigate, see phobos/phobos#48

We realize that this is our problem, and we have the workaround to lock < 0.5.0 of Ruby Kafka, but we think you might have seen something similar and can help put us on the right track. I guess the outcome can be either of two things

• There is a bug in ruby kafka that our tests helped expose
• There is a migration path from 0.4.x to 0.5x that we haven't observed

So... Anyway, here is the report and steps to reproduce.

• Version of Ruby: 2.4.1
• Version of Kafka: 0.10.2.0
• Version of ruby-kafka: 0.5.0+

Steps to reproduce

1. Clone https://github.com/klarna/phobos
2. In the gemspec, remove the locking of ruby-kafka to allow 0.5.1 to be installed
3. Run docker-compose up -d --force-recreate kafka zookeeper to start dependencies
4. Run docker-compose build test to build test container
5. Run docker-compose run --rm test bundle exec rspec

Many listeners spec will fail with

     IOError:
       closed stream

Expected outcome

Tests should be green, right now we have to lock < 0.5.0 of Ruby Kafka.

Actual outcome

Tests are red (listener tests only.

[dasch]

Do you have ruby-kafka logs (with DEBUG level) that you can share?

[dasch]

Also, can you give a full stack trace and include information about the API that's being called?

[klippx]

I ran the suite with 0.4.4 and 0.5.1 to try to compare the (normalized) debug log output.

The output is available here:
0.4.4: https://gist.github.com/klippx/a119fcb8225bd9df46f4941562572040
0.5.1: https://gist.github.com/klippx/0ca38936059118ecbd8d701b9d7ddb3d

I need to analyse this more thoroughly tomorrow.

[tom-pang]

@klippx which version of kafka are your tests running against?

[klippx]

We are using jplock/zookeeper:3.4.10 and ches/kafka:0.10.2.1 for the test suite.

[dasch]

It looks like you're returning from the each_batch block – that's not the intended way of terminating the loop, you need to call stop on the Consumer instance.

[klippx]

Actually, I think we are doing it right, give me a chance to explain what is going on in listener_spec.rb:87

https://github.com/klarna/phobos/blob/62c77a3390e873fc2bf29614ab2fc3eca4978388/spec/lib/phobos/listener_spec.rb#L89
First, we start a listener thread in the background and wait for it to signal ready. The listener thread is running a RubyKafka each_batch event loop, until we eventually stop it (consumer.stop) which happens in the final step.

https://github.com/klarna/phobos/blob/62c77a3390e873fc2bf29614ab2fc3eca4978388/spec/lib/phobos/listener_spec.rb#L100-L101
We then produce an event, and wait for the background thread to signal that it has processed it.

https://github.com/klarna/phobos/blob/62c77a3390e873fc2bf29614ab2fc3eca4978388/spec/lib/phobos/listener_spec.rb#L103-L104
Finally, we explicitly stop the listener, and part of this will stop the consumer (consumer.stop). Just to be certain we also return if inside the each_batch event loop, but maybe this is a bit superfluous, as the each_batch event loop would stop after consuming the next batch. I think we might as well remove this line of code.

https://github.com/klarna/phobos/blob/62c77a3390e873fc2bf29614ab2fc3eca4978388/lib/phobos/listener.rb#L65
Not part of the test but part of listener logic is also to stop the consumer in the ensure block, to make sure we always gracefully terminate consumers.

[dasch]

If you return from the block, stop won't have the effect you're looking for. Also, Timeout.timeout is weird and may cause problems.

[klippx]

I tried to remove the return completely, I still face the IOError: closed stream.

I will have a look at ruby-kafka changes in 0.4.4...0.5.1 (or even 0.5.0) and see if I can deduce what was the breaking change and/or how we are using it wrong. Looking at the stacktrace it could be anywhere from our consumer code to ruby-kafka consumer, connection or instrumentation.

Seems like there is a wait_for_response event loop going on inside connection class, which is still waiting for some data but the consumer is stopped => IO error. Which is wrapped in an instrumentation, so maybe we can figure out what is making a send request and catch this gracefully. IDK, need to give it more time.

     IOError:
       closed stream
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/socket_with_timeout.rb:60:in `select'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/socket_with_timeout.rb:60:in `read'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/protocol/decoder.rb:104:in `read'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/protocol/decoder.rb:48:in `int32'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/protocol/decoder.rb:88:in `bytes'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:180:in `read_response'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:199:in `block in wait_for_response'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:198:in `loop'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:198:in `wait_for_response'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:104:in `block in send_request'
     # /usr/local/bundle/gems/activesupport-5.1.4/lib/active_support/notifications.rb:168:in `instrument'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/instrumenter.rb:19:in `instrument'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/connection.rb:94:in `send_request'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/broker.rb:127:in `send_request'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/broker.rb:79:in `commit_offsets'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer_group.rb:73:in `commit_offsets'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/offset_manager.rb:120:in `commit_offsets'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/offset_manager.rb:137:in `commit_offsets_if_necessary'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer.rb:307:in `block (2 levels) in each_batch'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer.rb:279:in `each'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer.rb:279:in `block in each_batch'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer.rb:356:in `consumer_loop'
     # /usr/local/bundle/gems/ruby-kafka-0.5.1/lib/kafka/consumer.rb:271:in `each_batch'
     # ./lib/phobos/listener.rb:80:in `consume_each_batch'
     # ./lib/phobos/listener.rb:48:in `start'
     # ./spec/lib/phobos/listener_spec.rb:45:in `block (3 levels) in <top (required)>'
     # /usr/local/bundle/gems/logging-2.2.2/lib/logging/diagnostic_context.rb:474:in `block in create_with_logging_context'

[klippx]

~  src  phobos  ruby-kafka  ⚓ v0.4.3-2-g6fbaa35  git bisect good
9a5ef827fd4a89222b093c7c425d3e011baebb20 is the first bad commit
commit 9a5ef827fd4a89222b093c7c425d3e011baebb20
Author: Daniel Schierbeck <daniel.schierbeck@gmail.com>
Date:   Fri Oct 20 12:41:07 2017 +0200

   Disconnect the cluster on consumer shutdown

   This should more quickly close any open connections, resulting in faster
   graceful shutdown.

:040000 040000 9cc6ff85fc570d0b5dbefd81bc3a7542b346b676 ba123fccd8d6d8734ea6328882839703039682f9 M	lib
:040000 040000 c741cef0f4c416dd3f4f0d214a38f3af0bdf1527 4de638a38eec08f4b9abbbd27039dbeb500c6f98 M	spec

I did a git bisect and found the bad commit:

9a5ef82

It seems that after we call consumer.stop, we are still in the consumer_loop trying to do stuff. I am seeing errors that arise from various places within the loop:

• ruby-kafka/lib/kafka/consumer.rb

  Line 375 in 9a5ef82

  join_group unless @group.member?

• ruby-kafka/lib/kafka/consumer.rb

  Line 392 in 9a5ef82

  offset = @offset_manager.next_offset_for(topic, partition)

• ruby-kafka/lib/kafka/consumer.rb

  Line 404 in 9a5ef82

  operation.execute

I made a PR and verified this solves our issue.

[klippx]

After working with this issue and trying different solutions, I have come to believe this is more of a test issue than an actual compatibility problem.

When consumer is stopped, the cluster is disconnected, and a number of different things can be going on concurrently in the consumer loop. This can lead to many unexpected behaviors such as IOErrors, becoming unresponsive because of internals might be getting stuck retrying requests, etc. Basically, this can lead to a many different problems. Problems we don't need to deal with in the context of testing, at least.

In Phobos we stop and start consumers for each test, maybe it's simpler just to ignore the cluster disconnect and pretend that piece of code doesn't exist.

[mensfeld]

@klippx Karafka handles that that way. We basically catch it, report, silence and retry. Only heavily failing upon a general crash (but that will cause main thread to exit). Also, maybe this will be helpful for you guys on how to wrap consumer loop with additional logic: https://github.com/karafka/karafka/blob/master/lib/karafka/patches/ruby_kafka.rb