Karafka

https://karafka.io/docs
Admin
Karafka is a multi-threaded Kafka processing framework for Ruby and Rails, offering parallel...
Tokens:1,670,726
Snippets:15,715
Trust Score:7
Update:1 week ago
Show doc for...
Tokens:
Raw
### Setup global connection pool examples

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/ConnectionPool.html

Examples for basic and transactional setup of the global connection pool.

```ruby
WaterDrop::ConnectionPool.setup(size: 15) do |config|
  config.kafka = { 'bootstrap.servers': ENV['KAFKA_BROKERS'] }
  config.deliver = true
end
```

```ruby
WaterDrop::ConnectionPool.setup(size: 5) do |config, index|
  config.kafka = {
    'bootstrap.servers': ENV['KAFKA_BROKERS'],
    'transactional.id': "my-app-#{index}"
  }
end
```

--------------------------------

### Implement post_setup for OffsetMetadata

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/Routing/Features/ConsumerGroups/OffsetMetadata.html

The post_setup method installs the necessary listener and initializes the tracker when the application starts.

```ruby
def post_setup(_config)
  Karafka::App.monitor.subscribe("app.running") do
    # Initialize the tracker prior to becoming multi-threaded
    Karafka::Processing::ConsumerGroups::InlineInsights::Tracker.instance

    # Subscribe to the statistics reports and collect them
    Karafka.monitor.subscribe(
      Karafka::Pro::Processing::ConsumerGroups::OffsetMetadata::Listener.new
    )
  end
end
```

--------------------------------

### post_setup method implementation

Source: https://karafka.io/docs/code/karafka/Karafka/Routing/Features/ConsumerGroups/InlineInsights.html

Installs the needed listener and initializes the tracker if required. Called during the routing feature setup phase.

```ruby
def post_setup(_config)
  # If needed installs the needed listener and initializes tracker
end
```

--------------------------------

### setup

Source: https://karafka.io/docs/code/karafka/Karafka/App.html

Setup method included from Setup::Dsl.

```APIDOC
## setup

### Description
Setup method included from Setup::Dsl.
```

--------------------------------

### Karafka::Web.enable! method

Source: https://karafka.io/docs/code/karafka-web/Karafka/Web.html

Activates all needed routing and sets up the listener. Must run after optional configuration. Runs setup if not already configured, validates config, and enables the installer.

```ruby
# File 'lib/karafka/web.rb', line 54

def enable!
  # Run the setup to initialize components if user did not run it prior himself
  setup unless @configured

  # Make sure config is as expected
  # It should be configured before enabling the Web UI
  Contracts::Config.new.validate!(config.to_h)

  Installer.new.enable!
end
```

--------------------------------

### setup

Source: https://karafka.io/docs/code/karafka/Karafka/Setup/Dsl.html

Sets up the whole configuration by delegating to Setup::Config.setup with a block.

```APIDOC
## setup

### Description
Sets up the whole configuration.

### Method
setup

### Parameters
- `&block` (Proc) - Optional - Configuration block

### Source
```ruby
def setup(&)
  Setup::Config.setup(&)
end
```
```

--------------------------------

### start method

Source: https://karafka.io/docs/code/karafka/Karafka/Swarm/Manager.html

Starts all expected nodes for the first time.

```ruby
# File 'lib/karafka/swarm/manager.rb', line 39

def start
  @nodes.each(&:start)
end
```

--------------------------------

### setup ⇒ Object

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Instrumentation/Vendors/Datadog/MetricsListener.html

Alias for configure, consistent with WaterDrop#setup.

```APIDOC
## setup ⇒ Object

### Description
Alias for configure, consistent with WaterDrop#setup.

### Method
setup

### Example
```ruby
listener.setup
```

```

--------------------------------

### Base#setup method

Source: https://karafka.io/docs/code/karafka/Karafka/Instrumentation/Vendors/Appsignal/Base.html

Defines the setup method, which is an alias to be consistent with Karafka#setup. It simply calls configure with the block.

```ruby
# File 'lib/karafka/instrumentation/vendors/appsignal/base.rb', line 22

def setup(&)
  configure(&)
end
```

--------------------------------

### Karafka::Setup::Config#setup method

Source: https://karafka.io/docs/code/karafka/Karafka/Setup/Config.html

Defines the setup method that configures the Karafka framework. It verifies the license, pre-sets routing features and Pro components, wraps the config in a ConfigProxy, validates the config, configures components, installs backwards-compatible forwarders, refreshes cached references, post-sets routing features, subscribes critical error and assignments listeners, and marks the app as initialized. The block is optional and yields the proxy if provided.

```ruby
def setup(&)
  # Will prepare and verify license if present
  Licenser.prepare_and_verify(config.license)

  # Pre-setup configure all routing features that would need this
  Routing::Features::Base.pre_setup_all(config)

  # Will configure all the pro components
  # This needs to happen before end user configuration as the end user may overwrite some
  # of the pro defaults with custom components
  Pro::Loader.pre_setup_all(config) if Karafka.pro?

  # Wrap config in a proxy that intercepts producer block configuration
  proxy = ConfigProxy.new(config)
  # We need to check for the block presence here because user can just run setup without
  # any block given
  configure { yield(proxy) if block_given? }

  Contracts::Config.new.validate!(
    config.to_h,
    scope: %w[config]
  )

  configure_components(proxy)

  # Install backwards-compatible forwarding so that gems (e.g. karafka-testing) that
  # still access config.internal.processing.strategy_selector (etc.) keep working after
  # the move to config.internal.processing.consumer_groups.*
  install_processing_cg_forwarders(config)

  # Refreshes the references that are cached that might have been changed by the config
  Karafka.refresh!

  # Post-setup configure all routing features that would need this
  Routing::Features::Base.post_setup_all(config)

  # Subscribe the critical errors listener so process-critical errors reported anywhere
  # in the framework escalate to a graceful shutdown. Subscribed before any other
  # internal listener (including the Pro post-setup subscriptions) so it reacts first
  config.monitor.subscribe(Instrumentation::CriticalErrorsListener.instance)

  # Runs things that need to be executed after config is defined and all the components
  # are also configured
  Pro::Loader.post_setup_all(config) if Karafka.pro?

  # Subscribe the assignments tracker so we can always query all current assignments
  config.monitor.subscribe(Instrumentation::AssignmentsTracker.instance)

  Karafka::App.initialized!
end
```

--------------------------------

### start method

Source: https://karafka.io/docs/code/karafka/Karafka/Swarm/Manager.html

Starts the manager by creating the specified number of nodes and starting each one. Uses the current process ID as the parent PID.

```ruby
# File 'lib/karafka/swarm/manager.rb', line 39

def start
  parent_pid = ::Process.pid

  @nodes = Array.new(nodes_count) do |i|
    start_one Node.new(i, parent_pid)
  end
end
```

--------------------------------

### #setup ⇒ Object

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Producer.html

Sets up the whole configuration and initializes all that is needed.

```APIDOC
## #setup ⇒ Object

### Description
Sets up the whole configuration and initializes all that is needed.
```

--------------------------------

### Start Karafka server with bundle exec karafka server

Source: https://karafka.io/docs/Basics-Getting-Started

Starts the Karafka server to consume messages. The example outcome shows polling and consuming messages from the 'example' topic.

```shell
bundle exec karafka server

# example outcome
[7616dc24-505a-417f-b87b-6bf8fc2d98c5] Polled 2 messages in 1000ms
[dcf3a8d8-0bd9-433a-8f63-b70a0cdb0732] Consume job for ExampleConsumer on example-0 started
{"ping"=>"pong"}
{"ping"=>"pong"}
[dcf3a8d8-0bd9-433a-8f63-b70a0cdb0732] Consume job for ExampleConsumer on example-0 finished in 0 ms
```

--------------------------------

### Get listener classes

Source: https://karafka.io/docs/code/karafka-core/Karafka/Core/Monitoring/Monitor.html

Example usage to get only the classes of listeners from the listeners hash.

```ruby
monitor.listeners.map(&:class)
```

--------------------------------

### #setup ⇒ Object

Source: https://karafka.io/docs/code/karafka/Karafka/Instrumentation/Vendors/Datadog/LoggerListener.html

Sets up the listener, likely applying configuration.

```APIDOC
## #setup ⇒ Object

### Description
Sets up the listener, likely applying configuration.

### Method
setup

### Returns
Object

```

--------------------------------

### Pipe File Descriptor Calculation Examples

Source: https://karafka.io/docs/Librdkafka-Threads-and-Pipe-Patterns

Examples of calculating pipe file descriptors for a 3-broker cluster and a larger 16-broker cluster setup.

```text
pipe_fds = 4 × 2 = 8
```

```text
pipe_fds ≈ 132 × 17 × 2 = 4,488
```

--------------------------------

### post_setup_all method

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/Loader.html

Runs post-setup operations for each feature, initializes the subscription groups coordinator, and subscribes to lag compensation refresher if enabled.

```ruby
# File 'lib/karafka/pro/loader.rb', line 82

def post_setup_all(config)
  features.each { |feature| feature.post_setup(config) }

  # We initialize it here so we don't initialize it during multi-threading work
  Processing::ConsumerGroups::SubscriptionGroupsCoordinator.instance

  # Subscribe the paused partitions lags refresher only when the feature is enabled.
  # It is checked post setup as the interval is user-configurable during setup
  unless config.internal.statistics.consumer_groups.lag_compensation.interval.zero?
    config.monitor.subscribe(
      Instrumentation::ConsumerGroups::LagCompensation::Refresher.new
    )
  end
end
```

--------------------------------

### Custom Scheduler Configuration

Source: https://karafka.io/docs/Pro-Consumer-Groups-Scheduling-API

Example of assigning a custom scheduler class during Karafka setup.

```APIDOC
## Custom Scheduler Configuration

### Description
Assign a custom scheduler class during Karafka setup. The class is dynamically built, so assign the class, not an instance.

### Configuration
- **config.internal.processing.scheduler_class** (Class) - Required - The custom scheduler class to use.

### Example
```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    # Make sure to assign the class and not the instance as it is dynamically built
    config.internal.processing.scheduler_class = MyCustomScheduler
  end
end
```
```

--------------------------------

### name method examples

Source: https://karafka.io/docs/code/karafka/Karafka/Cli/Base.html

Examples showing the output of the name method for different class names. For Karafka::Cli::Install it returns 'install', and for Karafka::Cli::TestMe it returns 'test_me'.

```ruby
name #=> 'install'
```

```ruby
name => 'test_me'
```

--------------------------------

### setup(...) ⇒ Object

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Producer.html

Sets up the whole configuration and initializes all that is needed. Raises an error if the producer has already been configured.

```APIDOC
## setup(...) ⇒ Object

### Description
Sets up the whole configuration and initializes all that is needed. Raises an error if the producer has already been configured.

### Method
setup

### Endpoint
setup(...)

### Parameters
- **...** (variadic) - Configuration options passed to the underlying Config setup.

### Raises
- **Errors::ProducerAlreadyConfiguredError** - If the producer has already been configured.

### Returns
- **Object** - The result of the setup process.

### Example
```ruby
producer.setup(...)
```
```

--------------------------------

### Get started_at timestamp

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/ScheduledMessages/Tracker.html

Returns the time epoch when this tracker was started. The value is stored in @started_at.

```ruby
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 44

def started_at
  @started_at
end
```

--------------------------------

### .setup

Source: https://karafka.io/docs/code/karafka/Karafka/Setup/Config.html

Configuring method that sets up all framework details. It accepts a block for user configuration, validates the configuration, and initializes all components.

```APIDOC
## .setup

### Description
Configuring method that sets up all framework details. It accepts a block for user configuration, validates the configuration, and initializes all components.

### Method
setup

### Signature
setup(&) ⇒ Object

### Parameters
- **block** (Proc) - Optional - A block that yields a proxy object for configuration.

### Returns
- **Object** - The result of the setup process.

### Example
```ruby
Karafka::Setup::Config.setup do |config|
  # configuration code
end
```

### Notes
- This method prepares and verifies the license if present.
- It pre-sets up routing features and pro components.
- It validates the configuration against a contract.
- It subscribes critical errors listener and assignments tracker.
- It marks the app as initialized.
```

--------------------------------

### Karafka::Swarm::Node#start

Source: https://karafka.io/docs/code/karafka/Karafka/Swarm/Node.html

Starts a new fork and performs setup: stores pid and parent reference, makes sure reader pipe is closed, sets up liveness listener, and recreates producer and web producer.

```APIDOC
## Karafka::Swarm::Node#start

### Description
Starts a new fork and performs setup: stores pid and parent reference, makes sure reader pipe is closed, sets up liveness listener, and recreates producer and web producer.

### Method
start

### Returns
- **Object** - Result of starting the fork.

### Example
```ruby
node.start
```
```

--------------------------------

### call method

Source: https://karafka.io/docs/code/karafka/Karafka/Cli/Install.html

Performs the installation: creates directories, renders templates, and writes files.

```ruby
def call
  INSTALL_DIRS.each do |dir|
    FileUtils.mkdir_p Karafka.root.join(dir)
  end

  puts
  puts "Installing Karafka framework..."
  puts "Ruby on Rails detected..." if rails?
  puts

  INSTALL_FILES_MAP.each do |source, target|
    pathed_target = Karafka.root.join(target)
    FileUtils.mkdir_p File.dirname(pathed_target)

    template = File.read(Karafka.core_root.join("templates/#{source}"))
    render = ERB.new(template, trim_mode: "-").result(binding)

    File.write(pathed_target, render)

    puts "#{green("Created")} #{target}"
  end

  puts
  puts("Installation #{green("completed")}. Have fun!")
  puts
end
```

--------------------------------

### Incorrect consumer setup causing double processing

Source: https://karafka.io/docs/Infrastructure-Debugging

This Ruby example shows a consumer that raises an exception on a specific message, causing the entire batch to be retried and leading to duplicate processing of earlier messages. It is labeled as an incorrect setup.

```ruby
# This example illustrates incorrect setup
class OrdersConsumer < ApplicationConsumer
  def consume
    messages.each do |message|
      # will crash on e.g. 3rd message
      raise 'Boom' if message.payload == 'fail'  # simulate a broken message
    end
  end
end
```

--------------------------------

### Recommended setup location in karafka.rb

Source: https://karafka.io/docs/Monitoring-and-Logging

Place custom uptime listeners in karafka.rb after the configuration block to ensure they are loaded when Karafka starts.

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    # Your configuration here
  end
end

# Subscribe to uptime monitoring events
Karafka.monitor.subscribe 'connection.listener.fetch_loop' do |event|
  listener_id = event[:caller].id
  Rails.logger.info "[UptimeMonitor] #{listener_id} - Karafka is alive and polling..."
end
```

--------------------------------

### ConfigProxy lifecycle diagram

Source: https://karafka.io/docs/code/karafka/Karafka/Setup/ConfigProxy.html

Shows the sequence of steps from Setup.setup through proxy creation, user configuration, interception of producer calls, and final proxy discard.

```text
Setup.setup(&block)
  ↓
proxy = ConfigProxy.new(config)  # Create proxy
  ↓
configure { yield(proxy) }       # User block receives proxy
  ↓
[User calls config methods]      # Most delegate to real config
  ↓
[User calls config.producer {}]  # Intercepted by proxy
  ↓
configure_components(proxy.producer_initialization_block)  # Block retrieved
  ↓
[Block executed with real producer config]
  ↓
proxy = nil                      # Proxy discarded

```

--------------------------------

### Implementation of post_setup method

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/RecurringTasks.html

Performs post-setup configuration validation, event registration, and listener subscription.

```ruby
# File 'lib/karafka/pro/recurring_tasks.rb', line 92

def post_setup(config)
  RecurringTasks::Contracts::Config.new.validate!(
    config.to_h,
    scope: %w[config]
  )

  # Published after task is successfully executed
  Karafka.monitor.notifications_bus.register_event("recurring_tasks.task.executed")

  # Initialize empty dummy schedule, so we always have one and so we do not have to
  # deal with a case where there is no schedule
  RecurringTasks.schedule

  # User can disable logging of executions, in which case we don't track them
  return unless recurring_tasks_logging

  Karafka.monitor.subscribe(Listener.new)
end
```

--------------------------------

### BAD: per-message transactions with find_each

Source: https://karafka.io/docs/WaterDrop-Transactions

BAD example: with a transactional producer, each message production starts and commits its own transaction, adding latency. Avoid this pattern for performance.

```ruby
# This code with a transactional producer will create and commit transaction
# with each outgoing message, slowing things down
Users.find_each do |user|
  producer.produce_async(topic: 'users', payload: user.to_json)
end
```

--------------------------------

### Describe topic configuration

Source: https://karafka.io/docs/code/karafka/Karafka/Admin/Configs.html

Example showing how to describe a topic named 'example' and print its configuration values. The result is always an array of resources.

```ruby
resource = Karafka::Admin::Configs::Resource.new(type: :topic, name: 'example')
results = Karafka::Admin::Configs.describe(resource)
results.first.configs.each do |config|
  puts "#{config.name} - #{config.value}"
end
```

--------------------------------

### Configure Web UI branding

Source: https://karafka.io/docs/Pro-Web-UI-Branding

Example configuration for setting the branding type, notice message, and environment label within the Karafka Web setup block.

```ruby
Karafka::Web.setup do |config|
  config.ui.branding.type = :warning
  config.ui.branding.notice = 'You are working in the production environment - proceed with caution!'
  config.ui.branding.label = 'Production'
end
```

--------------------------------

### setup ⇒ Object

Source: https://karafka.io/docs/code/karafka/Karafka/Instrumentation/Vendors/Datadog/MetricsListener.html

Sets up the listener, likely configuring the Datadog client.

```APIDOC
## setup ⇒ Object

### Description
Sets up the listener, likely configuring the Datadog client.

### Method
setup

### Returns
- **Object** - The result of the setup.
```

--------------------------------

### Misusing Manual Offset Management - Incorrect Setup

Source: https://karafka.io/docs/Infrastructure-Debugging

Shows an incorrect Karafka setup where manual offset management is enabled but mark_as_consumed is never called, causing offsets to never be committed and messages to be redelivered on restart. Use this as a cautionary example; always mark messages as consumed or re-enable auto commits.

```ruby
# This example illustrates incorrect setup
class KarafkaApp < Karafka::App
  routes.draw do
    topic :orders do
      consumer OrdersConsumer
      # Auto marking disabled
      manual_offset_management(true)
    end
  end
end

class OrdersConsumer < ApplicationConsumer
  def consume
    messages.each do |message|
      process(message)
      # mark_as_consumed is missing - offsets will never be committed!
    end
  end
end
```

--------------------------------

### Set initial_offset to 'latest' globally in KarafkaApp setup

Source: https://karafka.io/docs/Basics-Consuming-Messages

Configures Karafka to start consuming from the latest message on a given topic for all topics by default. Overrides the default 'earliest' behavior.

```ruby
# This will make Karafka start consuming from the latest message on a given topic
class KarafkaApp < Karafka::App
  setup do |config|
    config.initial_offset = 'latest'
  end
end
```

--------------------------------

### pre_setup_all method

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/Loader.html

Runs pre-setup operations for each feature, reconfigures, expands, and loads topic features.

```ruby
# File 'lib/karafka/pro/loader.rb', line 70

def pre_setup_all(config)
  features.each { |feature| feature.pre_setup(config) }

  reconfigure(config)
  expand

  load_topic_features
end
```

--------------------------------

### #initialize

Source: https://karafka.io/docs/code/karafka/Karafka/Instrumentation/Notifications.html

Constructor method that returns a monitor instance for system instrumentation.

```APIDOC
## #initialize

### Description
Returns monitor instance for system instrumentation.

### Signature
`#initialize` ⇒ `Karafka::Instrumentation::Monitor`

### Returns
- **Karafka::Instrumentation::Monitor** - The monitor instance.
```

--------------------------------

### Get current offset for pagination

Source: https://karafka.io/docs/code/karafka-web/Karafka/Web/Ui/Controllers/Requests/Params.html

Returns the offset from which to start. -1 indicates the first page discovered based on the high watermark offset. If no offset is provided, it defaults to -1.

```ruby
# File 'lib/karafka/web/ui/controllers/requests/params.rb', line 98

def current_offset
  @current_offset ||= begin
    offset = @request_params.fetch("offset", -1).to_i
    [offset, -1].max
  end
end
```

--------------------------------

### Configure WaterDrop::Polling::Config before creating producers

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Polling/Config.html

Use this setup block to configure the poller settings before creating any producers. The example sets thread_priority to -1 and poll_timeout to 500.

```ruby
WaterDrop::Polling::Config.setup do |config|
  config.thread_priority = -1
  config.poll_timeout = 500
end
```

--------------------------------

### post_setup_all - Configure Pro components and subscribe listeners

Source: https://karafka.io/docs/code/karafka-web/Karafka/Web/Pro/Loader.html

Runs post-setup configuration operations for Commanding, Branding, Policies, and Search, then subscribes each commanding listener to the Karafka monitor.

```ruby
def post_setup_all(config)
  Commanding.post_setup(config)
  Ui::Lib::Branding.post_setup(config)
  Ui::Lib::Policies.post_setup(config)
  Ui::Lib::Search.post_setup(config)

  config.commanding.listeners.each do |listener|
    ::Karafka::App.monitor.subscribe(listener)
  end
end
```

--------------------------------

### Karafka app setup with ConfigProxy

Source: https://karafka.io/docs/code/karafka/Karafka/Setup/ConfigProxy.html

Example of using ConfigProxy in a Karafka app: standard config is delegated to the real config, while the producer block is intercepted and applied to the producer configuration.

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    # Standard config access - delegated to real config
    config.kafka = { 'bootstrap.servers': 'localhost:9092' }
    config.client_id = 'my_app'

    # Special intercepted method - handled by proxy
    config.producer do |producer_config|
      producer_config.kafka['compression.type'] = 'snappy'
      producer_config.kafka['linger.ms'] = 10
      producer_config.max_wait_timeout = 60_000
    end
  end
end

```

--------------------------------

### setup method implementation

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Producer.html

Configures the producer, raising an error if already configured. Sets up configuration, monitor, contract, and default variant. If idle disconnect timeout is zero, instruments configuration and returns; otherwise subscribes an idle disconnector listener.

```ruby
def setup(...)
  raise Errors::ProducerAlreadyConfiguredError, id unless @status.initial?

  @config = Config
    .new
    .setup(...)
    .config

  @id = @config.id
  @monitor = @config.monitor
  @contract = Contracts::Message.new(max_payload_size: @config.max_payload_size)
  @default_variant = Variant.new(self, default: true)

  if @config.idle_disconnect_timeout.zero?
    @status.configured!

    # Instrument producer configuration for global listeners
    class_monitor.instrument(
      "producer.configured",
      producer: self,
      producer_id: @id,
      config: @config
    )

    return
  end

  # Setup idle disconnect listener if configured so we preserve tcp connections on rarely
  # used producers
  disconnector = Instrumentation::IdleDisconnectorListener.new(
    self,
    disconnect_timeout: @config.idle_disconnect_timeout
  )

  @monitor.subscribe(disconnector)

  @status.configured!
end
```

--------------------------------

### Configure enable.partition.eof globally or per-subscription-group

Source: https://karafka.io/docs/Consuming-messages

This example demonstrates how to set the 'enable.partition.eof' option within the Karafka application setup block or specifically within a subscription group's routing configuration.

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    config.client_id = "my_application-#{Process.pid}-#{Socket.gethostname}"

    config.kafka = {
      'bootstrap.servers': '127.0.0.1:9092',
      'enable.partition.eof': true
    }
  end

  # You can also do it per topics in a subscription group
  routes.draw do
    subscription_group :fast do
      topic 'events' do
        consumer EventsConsumer
        kafka(
          'bootstrap.servers': '127.0.0.1:9092',
          'enable.partition.eof': true
        )
      end
    end
  end
end
```

--------------------------------

### Set global initial_offset to earliest

Source: https://karafka.io/docs/Basics-Consuming-Messages

Configure Karafka to start consuming from the earliest available offset for all topics. This is the default behavior. Set `config.initial_offset = 'earliest'` in the setup block of your Karafka application configuration.

```ruby
# This will start from the earliest (default)
class KarafkaApp < Karafka::App
  setup do |config|
    config.initial_offset = 'earliest'
  end
end
```

--------------------------------

### .included(base)

Source: https://karafka.io/docs/code/karafka-core/Karafka/Core/Configurable.html

Sets up all the class and instance methods and inits the core root node. Needs to be used when per instance configuration is needed.

```APIDOC
## .included(base) ⇒ Object

### Description
Sets up all the class and instance methods and inits the core root node. Needs to be used when per instance configuration is needed.

### Method
.included

### Parameters
- **base** (`Class`) - Required - class to which we want to add configuration

### Source
```ruby
def included(base)
  base.include InstanceMethods
  base.extend self
end
```
```

--------------------------------

### Setup global connection pool implementation

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/ConnectionPool.html

Implementation of the setup method for initializing the global connection pool.

```ruby
def setup(size: 5, timeout: 5000, &producer_config)
  ensure_connection_pool_gem!

  @default_pool = new(size: size, timeout: timeout, &producer_config)

  # Emit global event for pool setup
  WaterDrop.instrumentation.instrument(
    "connection_pool.setup",
    pool: @default_pool,
    size: size,
    timeout: timeout
  )

  @default_pool
end
```

--------------------------------

### Read messages from a given time with Karafka::Admin.read_topic

Source: https://karafka.io/docs/Infrastructure-Admin-API

Reads messages from a given time by passing a `Time` object as the fourth argument. The example reads at most 10 messages starting 60 seconds ago.

```ruby
topic = 'my_topic'
partition = 0
how_many = 10
# Read at most 10 messages starting 60 seconds ago
start_at = Time.now - 60

messages = Karafka::Admin.read_topic(topic, partition, how_many, start_at)

messages.each do |message|
  puts message.raw_payload
end

```

--------------------------------

### Enable lag compensation in KarafkaApp setup

Source: https://karafka.io/docs/Monitoring-and-Logging

Configuration example that enables lag compensation by setting the interval to 30,000 ms and pause_age to 30,000 ms. Requires statistics.interval.ms to be enabled. The feature is disabled by default (interval=0).

```ruby
class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka = {
      'bootstrap.servers': 'localhost:9092',
      'statistics.interval.ms': 5_000
    }

    # Refresh the lag of partitions paused for 30s or more, at most every 30s
    config.internal.statistics.consumer_groups.lag_compensation.interval = 30_000
    config.internal.statistics.consumer_groups.lag_compensation.pause_age = 30_000
  end
end
```

--------------------------------

### initialize(topic, partition, offset)

Source: https://karafka.io/docs/code/waterdrop/WaterDrop/Clients/Dummy/Handle.html

Constructor for a new instance of Handle. Stores the topic, partition, and offset for the dummy delivery.

```APIDOC
## initialize(topic, partition, offset)

### Description
Creates a new instance of Handle with the given topic, partition, and offset.

### Method
initialize

### Signature
`def initialize(topic, partition, offset)`

### Parameters
- **topic** (String) - Required - topic where we want to dispatch message
- **partition** (Integer) - Required - target partition
- **offset** (Integer) - Required - offset assigned by our fake "Kafka"

### Returns
- (Handle) - A new instance of Handle.

### Source
```ruby
def initialize(topic, partition, offset)
  @topic = topic
  @partition = partition
  @offset = offset
end
```
```

--------------------------------

### Implementation of pre_setup method

Source: https://karafka.io/docs/code/karafka/Karafka/Pro/RecurringTasks.html

Extends the configuration object with recurring tasks specific settings.

```ruby
# File 'lib/karafka/pro/recurring_tasks.rb', line 84

def pre_setup(config)
  # Expand the config with this feature specific stuff
  config.instance_eval do
    setting(:recurring_tasks, default: Setup::Config.config)
  end
end
```

--------------------------------

### Initialize reporter with error tracking setup

Source: https://karafka.io/docs/code/karafka-web/Karafka/Web/Tracking/Producers/Reporter.html

Sets up the reporter by initializing the tracked_at timestamp to 10 seconds in the past and creating an error contract for validation. This ensures any errors occurring right after sampling starts are dispatched immediately.

```ruby
# File 'lib/karafka/web/tracking/producers/reporter.rb', line 17

def initialize
  super
  # If there are any errors right after we started sampling, dispatch them immediately
  @tracked_at = monotonic_now - 10_000
  @error_contract = Tracking::Contracts::Error.new
end
```

--------------------------------

### Run the swarm supervisor (run method)

Source: https://karafka.io/docs/code/karafka/Karafka/Swarm/Supervisor.html

Validates CLI options, closes the producer, initializes rdkafka, warms up the app, starts the manager, installs signal handlers, and enters the supervision loop. Handles errors by instrumenting and terminating the manager.

```ruby
def run
  # Validate the CLI provided options the same way as we do for the regular server
  cli_contract.validate!(
    activity_manager.to_h,
    scope: %w[swarm cli]
  )

  # Close producer just in case. While it should not be used, we do not want even a
  # theoretical case since librdkafka is not thread-safe.
  # We close it prior to forking just to make sure, there is no issue with initialized
  # producer (should not be initialized but just in case)
  Karafka.producer.close

  # Ensure rdkafka stuff is loaded into memory pre-fork. This will ensure, that we save
  # few MB on forking as this will be already in memory.
  Rdkafka::Bindings.rd_kafka_global_init

  Karafka::App.warmup

  manager.start

  process.on_sigint { stop }
  process.on_sigquit { stop }
  process.on_sigterm { stop }
  process.on_sigtstp { quiet }
  process.on_sigttin { signal("TTIN") }
  # Needed to be registered as we want to unlock on child changes
  process.on_sigchld { nil }
  process.on_any_active { unlock }
  process.supervise

  Karafka::App.supervise!

  loop do
    return if Karafka::App.terminated?

    lock
    control
  end

# If the cli contract validation failed reraise immediately and stop the process
rescue Karafka::Errors::InvalidConfigurationError => e
  raise e
# If anything went wrong during supervision, signal this and die
# Supervisor is meant to be thin and not cause any issues. If you encounter this case
# please report it as it should be considered critical
rescue => e
  monitor.instrument(
    "error.occurred",
    caller: self,
    error: e,
    manager: manager,
    type: "swarm.supervisor.error"
  )

  manager.terminate
  manager.cleanup

  raise e
end
```
Tokens:
Raw