karafka / karafka-testing Goto Github PK

When multiple consumer groups are configured to consume from the same topic, as in the example below, the #karafka_consumer_for method will register its returned consumer only to the consumer group is defined last in the Karafka configuration file.

my_app.rb

class MyApp < Karafka::App
  consumer_groups.draw do
      consumer_group :group_one do
        topic("incoming_events") { consumer EventProcessingConsumer }
      end
  
      consumer_group :group_two do
        topic("incoming_events") { consumer EventReportingConsumer }
      end
    end
end

event_processing_consumer_spec.rb

RSpec.describe EventProcessingConsumer do
  subject(:consumer) { karafka_consumer_for("incoming_events") }

  # consumer is registered to the topic for  consumer group two 
  # since it is the last one defined in the consumer_group.draw block
  consumer.topic.consumer_group # => group_two 
end

I think this issue occurs because of the way the method iterates over the available topic names here, which seems to disregard the consumer_group for which the topic is registered. I would be helpful to be able to optionally pass a target consumer_group_name arg to this method in order to ensure the consumer was registered to the correct group/topic combination.

Affects 1.3, 1.4 and 2.0.

Won't be fixed in 1.3 as it is no longer supported.

Instead of relying on example-apps specs, we should setup it here to tackle edge cases and regressions.

Hi,
First, Thank you for your work for this gem.

in these lines, there is a hard coded App which actually can differ from one application to another.

I just thought of 2 alternatives:

1- Maybe ask the gem user to specify a let(:karafka_app) { MyCustomKarafkaApp } and then use that. (since you already expect the rspec subject to be defined by the user as well)

2- or perhaps there is a way to get this value from the Karafka gem. Actually one way to do so is to do something like this : Karafka::App.descendants.first (or .last) which returns the class that is defined in karafka.rb that inherits from Karafka::App.

Or maybe a hybrid approach. However, I'd go for the Karafka::App.descendants.first

EDIT:

@mensfeld A third idea, I just tried Karafka::App.consumer_groups in console and it worked and listed all my consumer groups. And since all karafka applications inherit from Karafka::App I think this would work.

Consider the following innocent spec:

class Xxx
  def initialize(foo:)
  end
end

RSpec.describe Xxx do
  include Karafka::Testing::RSpec::Helpers

  specify do
    expect { Karafka.producer.produce_sync(topic: 'foo', payload: 'bar') }
      .to change(karafka.produced_messages, :size).by(1)
  end
end

#=>

Failures:

  1) Xxx is expected to change `Array#size` by 1
     Failure/Error:
       def initialize(foo:)
       end
     
     ArgumentError:
       missing keyword: :foo
     # ./spec/xxx_spec.rb:2:in `initialize'
     # …/rspec-core-3.11.0/lib/rspec/core/memoized_helpers.rb:60:in `new'
…
     # …/karafka-testing-2.0.5/lib/karafka/testing/rspec/helpers.rb:91:in `_karafka_add_message_to_consumer_if_needed'

https://github.com/karafka/karafka-testing/blob/master/lib/karafka/testing/rspec/helpers.rb#L87-L94

        def _karafka_add_message_to_consumer_if_needed(message)
          # We're interested in adding message to subject only when it is a consumer
          # Users may want to test other things (models producing messages for example) and in
          # their case subject will not be a consumer
          return unless subject.is_a?(Karafka::BaseConsumer)
          # We target to the consumer only messages that were produced to it, since specs may also
          # produce other messages targeting other topics
          return unless message[:topic] == subject.topic.name

https://github.com/rspec/rspec-core/blob/main/lib/rspec/core/memoized_helpers.rb#L57-L62

      def subject
        __memoized.fetch_or_store(:subject) do
          described = described_class || self.class.metadata.fetch(:description_args).first
          Class === described ? described.new : described
        end
      end

so by calling subject we're trying to initialize the class that requires some arguments.

Thank you for this library, it's been very helpful for testing my topic to consumer mappings and making sure the correct consumer class is invoked and the correct behavior is found.

I'm switching now to using patterns instead of named topics, specifically for the DLQ case as described in the docs.

In my spec I'm using:

karafka.consumer_for("dlq-demo-topic")

And in my consumer mapping I have:

pattern(:dlqs_pattern, /^dlq.*/) do
  consumer DeadLetterQueueConsumer
end

I've tried a few different variations of the regex, but I always get Karafka::Testing::Errors::TopicNotFoundError when running my tests. As I understand, testing pattern matches like this is not currently supported.

I can see this in the 2.0.2 patch notes:

Rename `karafka.publish` to `karafka.produce` to align naming conventions.

But it doesn't call it out as a breaking change and probably shouldn't be in a patch release.

Even though it's a testing library the change broke all of our tests - probably worth noting in the CHANGELOG.md as its a public method.

If you configure a Rails application to use the karafka queue adapter as described in https://karafka.io/docs/Active-Job/, then when writing rspec tests that use the have_enqueued_job matcher, they will fail reporting no jobs were enqueued

Expected behavior

When configuring rails to use the karafka ActiveJob adapter with:

config.active_job.queue_adapter = :karafka

Then if I write ruby code to publish a job:

class SomeJob < ApplicationJob
  def perform
    puts "performed"
  end
end

SomeJob.perform_later

I expect an rspec test like this to pass:

expect do 
  SomeJob.perform_later
end.to have_enqueued_job(SomeJob).once

Actual behavior

The rspec test fails with

expected to enqueue exactly 1 jobs, but enqueued 0

I also observed this happens regardless of whether I use the produce_sync option or not

karafka_options(
  dispatch_method: :produce_sync
)

Steps to reproduce the problem

Create a rails 7 application, and configure it to use the :karafka queue adapter for ActiveJob as per above

Set up the karafka gem

Write a test that enqueues a job

See that it fails to report enqueueing the job

Your setup details

I am using Rails 7.0.4.3 and rspec 3.12.0

Karafka info is below:

$ [bundle exec] karafka info
2023-04-21 09:30:25.307444 I [65649:57120] Rails -- Karafka version: 2.0.38
Ruby version: ruby 3.2.1 (2023-02-08 revision 31819e82c8) +YJIT [arm64-darwin22]
Rdkafka version: 0.12.1
Consumer groups count: 4
Subscription groups count: 4
Workers count: 5
...

re: https://karafka.io/docs/Deserialization/#schema-registry_1

We are deserialising using Avro and Schema Registry. What we found was that the suggested AvroRegistryDeserializer code in the docs wasn't compatible with the Testing recommendations by Avro Turf.

Essentially the web mock stub wasn't able to stub the AvroTurf::Messaging.new client due to the client being intialised in the AVRO constant before the stub. Our workaround was to define a method instead, e.g:
def avro
@avro ||= AvroTurf::Messaging.new(...)
end

This meant the client is initialised after the stub has been called.

We're checking if karafka configuration is properly filled in specs, because we're getting all the configuration via

ENV['something'] = 'foo' # dotenv
MyApp.config.something = ENV['something'] # dry-c w/ contract
KarafkaApp.config.something = MyApp.config.something

and it's good to test if seed brokers etc is properly propagated.

With karafka-1 it was (simplified):
karafka.rb:

require 'my_app/karafka_app'

MyApp::KarafkaApp.boot!

my_app/karafka_app.rb

module MyApp
  # Main MyApp karafka consumer application. Starting via `<APP_ROOT>/karafka.rb`.
  class KarafkaApp < Karafka::App
    class << self
      # Setup application. This method allow us to re-run setup (e.g. for testing purposes)
      def boot!
        setup do |config|
          config.something = MyApp.config.something
        end

        draw_routes

        super

        setup_waterdrop

        config.finalize!
      end
  end
end

spec/my_app/karafka_app_spec.rb

RSpec.describe MyApp::KarafkaApp do
  def setup_karafka_app
    allow(Karafka::Setup::Config).to receive(:config).and_return(described_class.config.pristine)
    allow(WaterDrop::Config).to receive(:config).and_return(WaterDrop.config.pristine)
    allow(DeliveryBoy).to receive(:config).and_return(DeliveryBoy::Config.new(env: ENV))

    described_class.boot!
  end

  it { expect { setup_karafka_app }.to get_config(:something).from_envvar(:something) }
end

and now setup is approx. the same, but it's not clear what should be used instead of #pristine to get a copy of the config in these specs:

(ruby) described_class.config.concurrency
5
(ruby) described_class.config.deep_dup.concurrency
eval error: undefined method `concurrency' for #<Karafka::Core::Configurable::Node:0x00007f0ea0fe8f48 @name=:root,…

described_class.config.deep_dup.concurrency
                               ^^^^^^^^^^^^

So some method to get a config that could be messed with is needed.

This issue lists Renovate updates and detected dependencies. Read the Dependency Dashboard docs to learn more.

This repository currently has no open or pending branches.

Detected dependencies

• Check this box to trigger a request for Renovate to run again on this repository

I'm seeing the following error after upgrading to 2.1.1:

NameError: uninitialized constant WaterDrop::Producer::DummyClient

    class SpecProducerClient < ::WaterDrop::Producer::DummyClient
                                                    ^^^^^^^^^^^^^
from /Users/jordan/.rbenv/versions/3.1.2/lib/ruby/gems/3.1.0/gems/karafka-testing-2.1.1/lib/karafka/testing/spec_producer_client.rb:7:in `<module:Testing>'

It looks like WaterDrop::Producer::DummyClient was renamed to WaterDrop::Clients::Dummy in this PR: karafka/waterdrop#353, but there was no corresponding change in this project to update it.

I see a closed issue here that looks like it may be related, but not 100% sure: #142

It would be good to have a method to test whether error messages in karafka consumer moved to DLQ consumer

After upgrading to karafka-testing 2.0.2 I faced the next error:

NameError:
       undefined local variable or method `consumer'

The reason for this is that I had the subject defined without any name
subject { karafka.consumer_for('topic_name') }
but to fix the error you need to specify the name
subject(:consumer) { karafka.consumer_for('topic_name') }
because there is a dependency on consumer variable inside karafka-testing here
https://github.com/karafka/karafka-testing/blob/master/lib/karafka/testing/rspec/helpers.rb#L90

Just caught this error in our CI when upgrading karafka-testing via Dependabot.

We don't automatically require our gems in the Gemfile and, following the testing docs, we had only this in spec_helper:

require 'karafka/testing/rspec/helpers'

RSpec.configure do |config|
  config.include Karafka::Testing::RSpec::Helpers
end

Simply adding require 'karafka/testing' as well fixed the issue. Not sure if this is a bug or the documentation should be updated.

As reported on Slack

I received a undefined method topic' for #<Model ... error when writing come model specs in my Rails app because my model was producing Kafka events.

The source of the bug seems to be related to this line which forces me to define my test subject as a Karakfa testing consumer instead of an instance on my model class.

But as I do not consume any event from the topic on which I produce the event I am unable to use the consumer_for helper and have to define test double to avoid the bug

@mensfeld wrote this in response to my question

need to decommission the subject from consumer reference and then for consumer specs use the subject but otherwise direct reference

I want to be able test my code that produces messages. For example, if I have the following method

def test_method(val)
  Karafka.producer.produce_async(
      topic: "test",
      payload: {
        val: val
      }.to_json
  )
end

I would ideally like to test this by checking an event was added to the "test" topic with an appropriate payload. Is there a way to achieve this with the karafka-testing gem? Or maybe an alternative recommended way to test producer code?

Hi, thank's for the gem.
Have an issue, may be dependent on using this gem with rails.

described_class.new(selected_topic)
=> wrong number of arguments (given 1, expected 0)

described_class
=> BeaconEventsConsumer

../app/consumers/beacon_events_consumer.rb

class BeaconEventsConsumer < ApplicationConsumer
  # This method just logs parsed data with Karafka logger
  def consume
    Rails.logger.info "Consumed following message: #{params}"

  end
end

../karafka.rb

  consumer_groups.draw do
    consumer_group :batched_group do
      topic :beacon_events do
        consumer BeaconEventsConsumer
        batch_consuming false
      end
    end


  end

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).

I have no name!@33477f4db36a:/app$ ls -larth /app/vendor/bundle/ruby/3.1.0/gems/karafka-testing-2.0.6/certs/cert_chain.pem
-rw-------. 1 root root 1.6K Nov  2 09:20 /app/vendor/bundle/ruby/3.1.0/gems/karafka-testing-2.0.6/certs/cert_chain.pem

in the docker image using karafka-testing and

Run cp -r /app/vendor/bundle ./vendor/
cp: cannot open '/app/vendor/bundle/ruby/3.1.0/gems/karafka-testing-2.0.6/certs/cert_chain.pem' for reading: Permission denied
Error: Process completed with exit code 1.

in CI where gems are installed by root and everything is run by unprivileged user.

certs/cert_chain.pem should be a+r (and it is in the repo), it's not a security issue in this case anyway.

Follow-up of #106

To avoid situations when calling subject was having side effects it has been decided that the gem will use an explicit consumer reference to keep track of messages sent to Kafka.

This has the downside to require developers to declare a consumer object (or a named subject) in their tests instead of relying on implicit subjects.

In the issue comment thread, I thought about looking at the described_class and testing if it inherits from Karafka::BaseConsumer. If that is the case the gem's helper could try to look at the consumer referenced object or call the subject reference.

If the described_class is not a Karafka::BaseConsumer the helper will only look for the consumer object

If you have this route config, two consumer groups with same topics.

consumer_group 'A' do
  topic 'a'
  topic 'b'
end
  
consumer_group 'B' do
  topic 'a'
  topic 'b'
end

In testing helper method consumer_for works incorrectly, it returns consumer with consumer group 'A' even though 'B' was passed as the second parameter.

I believe the error is in helpers.rb, coordinator is set the wrong way.

  def _karafka_build_consumer_for(topic, topics_array)
          coordinators = Karafka::Processing::CoordinatorsBuffer.new(
            Karafka::Routing::Topics.new(topics_array)
          )

          consumer = topic.consumer.new
          consumer.producer = Karafka::App.producer
          # Inject appropriate strategy so needed options and components are available
          strategy = Karafka::App.config.internal.processing.strategy_selector.find(topic)
          consumer.singleton_class.include(strategy)
          consumer.client = _karafka_consumer_client
          consumer.coordinator = coordinators.find_or_create(topic.name, 0)
          consumer.coordinator.seek_offset = 0

          consumer
 end