MissingClientError on manual offset commit with rspec

[diogoleitao]

My karafka.rb is as follows

class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka.automatically_mark_as_consumed = false
    config.kafka.seed_brokers = Rails.configuration.kafka_broker_list.split(',').map { |broker| "kafka://#{broker}" }
    config.kafka.start_from_beginning = false
    config.batch_fetching = false
    config.client_id = Rails.configuration.kafka_client_id

    config.logger = Rails.logger if ENV['RAILS_LOG_TO_STDOUT'].present?
  end

  Karafka.monitor.subscribe(WaterDrop::Instrumentation::StdoutListener.new)
  Karafka.monitor.subscribe(Karafka::Instrumentation::StdoutListener.new)
  Karafka.monitor.subscribe(Karafka::Instrumentation::ProctitleListener.new)
  Karafka.monitor.subscribe(Karafka::CodeReloader.new(*Rails.application.reloaders))
  Karafka.monitor.subscribe(RollbarListener)

  consumer_groups.draw do
    ...
  end
end

And on my ApplicationConsumer I have

def consume
    @message = (topic.batch_consuming ? params_batch : params.payload)
    mark_as_consumed!(topic.batch_consuming ? @message.last : @message)

    DATADOG.time('process_message') do # rubocop:disable Style/ExplicitBlockArgument
      yield
    end
rescue StandardException => e
  do_some_stuff
end

When running the specs, every single test runs into this error Karafka::Errors::MissingClientError. I can't figure out what's causing it, and given that the documentation says this error shouldn't be raised at all, I don't know how to approach this issue.

Our setup is a Rails project, karafka (v 1.3.1) and karafka-testing (latest).

[mensfeld]

Hey man, sorry for the delay. I've been traveling. Feq questions:

• are you using most recent karafka-testing of most recent for 1.3?
• did you try to upgrade to 1.4? While I still support 1.3 it's close to EOL.
• do your specs run in multiple threads or anything are that?
• are you using some sort of thread storage that would be cleared with each spec running?

[mib32]

Also have same issue.

• using karafka-testing 1.4.2, karafka 1.4.9, ruby-kafka 1.4.0
• No, very simple setup rspec setup.
• Not sure what this question means... Kinda very plain test. I am using VCR though.

Example of failing test

require_relative '../../karafka_app'

RSpec.describe RuleNotificationsConsumer, vcr: true do
  # This will create a consumer instance with all the settings defined for the given topic
  subject(:consumer) { karafka_consumer_for(:scheduled_rule_events) }

  let(:app) { create(:app) }
  let(:rule1_no_log) { create(:butler_rule, app: app) }
  let(:rule2) { create(:butler_rule, :manual, app: app) }
  let!(:log_rule_2) { create(:butler_rule_log, rule: rule2) }
  let(:rule3) { create(:butler_rule, :manual, app: app) }
  let!(:log_rule_3) { create(:butler_rule_log, rule: rule3) }

  before do
    # Sends messages to Karafka consumer
    publish_for_karafka(encode(rule1_no_log.id, 'opened'))
    publish_for_karafka(encode(rule2.id, 'opened'))
    publish_for_karafka(encode(rule2.id, 'opened'))
    publish_for_karafka(encode(rule2.id, 'errored'))
    publish_for_karafka(encode(rule3.id, 'errored'))
    publish_for_karafka(encode('nonexistent_id', 'errored'))
  end

  def avro
    @avro ||= AvroRegistry.instance
  end

  def encode(rule_id, type)
    avro.encode({
                  'rule_id' => rule_id,
                  'type' => type,
                  'timestamp' => Time.now
                }, subject: 'scheduled_rule_event', version: 1)
  end

  it 'expects to log a proper message' do
    consumer.consume
    expect(log_rule_2.reload.opened_count).to eq 2
    expect(log_rule_2.errored_count).to eq 1
    expect(log_rule_3.reload.opened_count).to eq 0
    expect(log_rule_3.errored_count).to eq 1
  end
end

[mensfeld]

@mib32 would you be willing to narrow it down to a reproduction repo? With this I could fix it in minutes. Otherwise I will have to do my best to replicate your gems setup, etc.

[mensfeld]

@diogoleitao @mib32 I was able to reproduce. I will fix it today.

[mensfeld]

ref #71 and moving to testing lib

[mensfeld]

new karafka-testing has been released. it fixes this problem.