karafka / karafka Goto Github PK

Example of my app:

# Karafka application class
class App < Karafka::App
  Dir[App.root + 'lib/**/*.rb'].each { |file| require file }

  setup do |config|
    config.kafka_hosts = System::Settings.kafka.hosts
    config.zookeeper_hosts = System::Settings.zookeeper.hosts
    config.worker_timeout = System::Settings.sidekiq.timeout
    config.redis = System::Settings.sidekiq.redis
    config.concurrency = System::Settings.karafka.concurrency
    config.name = System::Settings.karafka.application_name
  end
end

# Load all application internal code (app/, lib/, etc)
Karafka::Loader.new.load(App.root)

I have to load /lib to get System::Settings before initializing App before running `Karafka::Loader.new.load. And doing 2 loading feels like duplicating. I think loading files and configuration should be 2 separate phases.

Hello,

after an incident that our Kafka producer generated a huge backlog, Karafka Sidekiq used huge ammounts of Ram while queueing everything in Redis. So the question is, is there a way to set a backoff configuration, so that the consumer stops fetching messages if the Sidekiq queue is maxed out?

Best
Giorgos

Now Karafka assumes that messages are in JSON format. I think there should an ability to hook up a custom parser.

Karafka looks great. But I want to use this is my app which has Sidekiq 3.4.1 and Celluloid 0.16.

Dependency on celluloid should be >= 0.17

Exceptions during load time should not be catched

Unless I has config/sidekiq.yml file, be rake karakfa:sidekiq starts sidekiq with default settings. You can track it by number of threads: it's always 25.

It'd be great to be able to set monitor in config:

class App < Karafka::App
  ...
  setup do |config|
    config.monitor = AppMonitor.new
    ...
  end
end

In default sidekiq config there's a possibility to specify redis namespace:

  config.redis = { url: 'redis://redis.example.com:7372/12', namespace: 'mynamespace' }

And karafka configuration lack this parameter.

I'm not sure how it should be resolved, since it really adds maintainability cost to support every sidekiq's option. That's why I think that explicit workers are great choice.

There should be an abillity to use explicit sidekiq workers, since current implicit creates more problems that it solves:

• it's impossible to use sidekiq plugins
• what if I want to use enterprise version of sidekiq
• there's not option to prioritise sidekiq workers

and so on. It'd be 100 times better to be able to use user-defined workers.

We should have an option to tune this parameter and based on that we should also tune Celluloid shutdown_timeout

I think we should notify developers about such difference with warning/logging/failing.

I see it's not easy to monitor karafka since it's not a rack-based app. It'd be good to add some monitoring options. For instance, it can be done for new relic with method tracers https://docs.newrelic.com/docs/agents/ruby-agent/features/ruby-custom-instrumentation#method_tracers

https://github.com/karafka/karafka/blob/master/lib/karafka/params/params.rb#L64

fetch method accidentally overrides Hash#fetch

Topics are stored in Zookeeper so we can get them from it.

Would be nice to have karafka routes to list routing that we have

If I define my own worker, perform method should not be needed.

class LandingInvitationController < Karafka::BaseController
  self.topic = 'mail.landing_invitation'
  self.worker = LandingInvitationJob

  # def perform
  # end
end

class LandingInvitationJob
  include Sidekiq::Worker

  sidekiq_options :queue => :landing_invitation

  def perform(payload)
    Karafka.logger.info(payload)
  end
end

echo "Cg5mb29Aa3VlbmRlLmNvbRIGYmFyYmF6" | kafkacat -b localhost:9092 -t mail.landing_invitation -P

Karafka::Errors::PerformMethodNotDefined (Karafka::Errors::PerformMethodNotDefined)

If I uncomment the perform method, it is not called anyway. So why should I add the empty method if worker is called instead ?

We should get back to enqueuing raw data and reparsing it twice if needed (or fix it somehow)
#30

When we trigger graceful shutdown, we should not process the whole bulk, instead we should stop processing after current message and commit the offset in this place

We need to update readme for all karafka apps to link to gitter:
https://gitter.im/karafka/karafka:

for consistency with base controller and future inheritance

We should have a base application controller and application worker by default

Worker should have worker glass options in it so they can be disabled.

This is especially related with #67 by @Geokoumpa that will require this in order to make it workable. Backoff might be implemented in the user code namespace.

Hello,

Does consumer support chroot for zookeeper connections?
I am asking because on a multi-environment setup and each Kafka broker should running under a different "/" node when all Kafka clusters are using the same Zookeeper cluster.

I've tried adding the chroot in config.zookeeper_hosts array but Karafka seems to ignore the setting.
config.zookeeper_hosts = %w( 10.0.0.1:2181/kafka_prd 10.0.0.2:2181/kafka_prd 10.0.0.3:2181/kafka_prd )

Thank you.

I'd like to use app.rb for sinatra app and leave karafka_app.rb for karafka app. It's not possible now and it forces me to keep all setting together in one file

Controller code:

# app/controllers/channels/playlist_items/uploaded_controller.rb
module Channels
  module PlaylistItems
    class UploadedController < Karafka::BaseController

Exception:

    from /Users/yukke/.rvm/gems/ruby-2.2.3@playlist-processor/bundler/gems/karafka-c43362cd860f/lib/karafka/workers/builder.rb:33:in `build'
    from /Users/yukke/.rvm/gems/ruby-2.2.3@playlist-processor/bundler/gems/karafka-c43362cd860f/lib/karafka/base_controller.rb:131:in `worker'
    from /Users/yukke/.rvm/gems/ruby-2.2.3@playlist-processor/bundler/gems/karafka-c43362cd860f/lib/karafka/routing/mapper.rb:15:in `map'
    from /Users/yukke/.rvm/gems/ruby-2.2.3@playlist-processor/bundler/gems/karafka-c43362cd860f/lib/karafka/routing/mapper.rb:15:in `workers'
    from /Users/yukke/.rvm/gems/ruby-2.2.3@playlist-processor/bundler/gems/karafka-c43362cd860f/lib/karafka/app.rb:19:in `bootstrap'
    from /Users/yukke/apps/StrikeSocial/PlaylistProcessor/app.rb:27:in `<top (required)>'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb/init.rb:280:in `require'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb/init.rb:280:in `block in load_modules'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb/init.rb:278:in `each'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb/init.rb:278:in `load_modules'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb/init.rb:20:in `setup'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/lib/ruby/2.2.0/irb.rb:378:in `start'
    from /Users/yukke/.rvm/rubies/ruby-2.2.3/bin/irb:15:in `<main>'

Hello,

I was not able to use routing because I get the error

undefined method 'routes' for App:Class (NoMethodError)

Where are routes supposed to be defined? I added mine inside app.rb, under App class definition.

Any ideas about what I might be doing wrong?

Also I had to use Karafka::BaseController insead of ApplicationController because the latter was undefined. I am using v 0.4.

thanks

shoould log to stdout and file in dev mode

NewRelic monitor have stopped to work after recent update of karafka.

Hi,

I'm working with protobuf messages and for each message I have a generated class:

class Proto::MyClass
  include Beefcake::Message
  required :email, :string, 1
  required :token, :string, 2
end

Is it possible to support lambda as well for parser class ?

class MyController < Karafka::BaseController
     self.parser = ParserProxy.proxy(Proto::MyClass)
end

class ParserProxy
  class ParserError < StandardError; end

  def self.proxy(klass)
    lambda do |msg|
      begin
        klass.decode(msg)
      rescue => e
        MyNotifier.capture_error(e, :extra => { :consumer => klass.name.to_s, :message => Base64.encode64(msg) })
        raise ParserError
      end
    end
  end
end

We should take the poseidon cluster logic and incorporate it in Karafka environment. It is really small and would be better to keep it as a core feature since we will work on clusterization, etc

I believe that we should have more strict routing engine that will allow use to do following things:

1. Validate topics names (so invalid won't hit Kafka)
2. Should allow catching typos and other programmers bugs
3. Should allow control over outgoing messages/events so we can design a nice flow that will be strict
4. Should be independent from the message source as long as it responds to .topic
5. Should take over the responsibility of managing topics from controllers and groups

They should also inherit from a Karafka::Errors::BaseError < Standard error so we cold catch them all at once if we need to

Channels::PlaylistItemsController => 'channels_playlists_items'

should be:

Channels::PlaylistItemsController => 'channels/playlists_items'

We could add an optional support for specifying an action that should be taken on a single topic, so we would could easier build a more restful controllers