### Production Setup with Multiple Producers and Shared Poller

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Example of setting up multiple producers for different topics in a production environment using FD mode and the singleton poller for efficiency. Producers are configured with specific client IDs.

```ruby
class KafkaProducersService
  def initialize
    # Use FD mode for efficiency in production
    @poller = WaterDrop::Polling::Poller.instance
  end
  
  def initialize_producers(brokers)
    @producers = {
      events: create_producer('events', brokers),
      logs: create_producer('logs', brokers),
      metrics: create_producer('metrics', brokers)
    }
  end
  
  private
  
  def create_producer(name, brokers)
    WaterDrop::Producer.new do |config|
      config.kafka = {
        'bootstrap.servers': brokers,
        'client.id': "app-#{name}"
      }
      config.polling.mode = :fd
      # All producers use @poller (singleton)
    end
  end
end

service = KafkaProducersService.new
service.initialize_producers('localhost:9092,localhost:9093')

# Single background thread manages all three producers
```

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

### Rails Integration with Connection Pool Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Provides an example of configuring the Waterdrop connection pool within a Rails initializer. It demonstrates setting Kafka broker details and delivery options based on the environment.

```ruby
# config/initializers/waterdrop.rb
WaterDrop::ConnectionPool.setup(size: 10) do |config|
  config.kafka = {
    'bootstrap.servers': ENV['KAFKA_BROKERS'],
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'SCRAM-SHA-256',
    'sasl.username': ENV['KAFKA_USERNAME'],
    'sasl.password': ENV['KAFKA_PASSWORD']
  }
  config.deliver = !Rails.env.test?
end

# In models/jobs
class UserSignupJob
  def perform(user_id)
    WaterDrop::ConnectionPool.with do |producer|
      producer.produce_async(
        topic: 'user-events',
        payload: { event: 'signup', user_id: user_id }.to_json,
        key: user_id.to_s
      )
    end
  end
end
```

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

### Async Produce Delivery Handle Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/types.md

Example demonstrating the use of a DeliveryHandle for asynchronous produce operations, including blocking to get the final report.

```ruby
handle = producer.produce_async(topic: 'events', payload: 'data')
report = handle.create_result  # Block until delivered
puts "Offset: #{report.offset}"
```

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

### Poller Initialization Examples

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Demonstrates how to obtain the default singleton `Poller` instance and how to create custom, isolated `Poller` instances.

```ruby
# Singleton instance (default)
default_poller = WaterDrop::Polling::Poller.instance

# Custom isolated instance
custom_poller = WaterDrop::Polling::Poller.new

poller1 = WaterDrop::Polling::Poller.new
poller2 = WaterDrop::Polling::Poller.new
# poller1 and poller2 are independent instances
```

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

### Global Pool Setup: setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Configures a global connection pool as a class-level singleton. It allows specifying the pool size, timeout, and a block for configuring each producer instance.

```APIDOC
## Global Pool Setup: setup

Configures a global connection pool as a class-level singleton.

### Signature:
```ruby
def self.setup(size: 5, timeout: 5000, &producer_config)
```

### Parameters:
#### Query Parameters
- **size** (Integer) - Optional - Default: 5 - Number of producers in the pool
- **timeout** (Integer) - Optional - Default: 5000 - Timeout in ms to wait for available producer
- **producer_config** (Proc) - Required - Block to configure each producer: `\|config, index\|`

### Returns:
`ConnectionPool` instance

### Raises:
RuntimeError if `connection_pool` gem is not available

### Block parameters:
- `config` - Config object for the producer
- `index` - Pool position (0-based)

### Example:
```ruby
# Basic setup
WaterDrop::ConnectionPool.setup(size: 10) do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092'
  }
  config.deliver = true
end

# With transactional producers
WaterDrop::ConnectionPool.setup(size: 5) do |config, index|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': "app-" + Socket.gethostname + "-" + index.to_s
  }
  config.deliver = true
end

# With custom logging
WaterDrop::ConnectionPool.setup(size: 15) do |config, index|
  config.kafka = { 'bootstrap.servers': ENV['KAFKA_BROKERS'] }
  config.logger = Logger.new("producer_" + index.to_s + ".log")
  config.deliver = true
end
```
```

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

### Configure Producer with Setup Block

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Configure an existing WaterDrop::Producer instance using a setup block. This method can only be called once per producer instance.

```ruby
producer.setup do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'request.required.acks': 1
  }
  config.deliver = true
  config.max_payload_size = 1_000_000
end
```

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

### Basic Synchronous Producer Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Initialize a producer for synchronous message delivery. Ensure `config.deliver` is set to `true` for actual Kafka interaction.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092'
  }
  config.deliver = true
end

producer.produce_sync(topic: 'events', payload: 'data')
producer.close
```

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

### Producer Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Configures the producer with a given block. This method can only be called once.

```APIDOC
## Producer Setup

### setup

Configures the producer with the given configuration block. Can only be called once; subsequent calls raise `ProducerAlreadyConfiguredError`.

**Signature:**
```ruby
def setup(...)
```

**Parameters:** Configuration block (see [Configuration](../configuration.md))

**Returns:** nil

**Raises:**
- `Errors::ProducerAlreadyConfiguredError` - if producer is already configured
- `Errors::ConfigurationInvalidError` - if configuration is invalid

**Example:**
```ruby
producer.setup do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'request.required.acks': 1
  }
  config.deliver = true
  config.max_payload_size = 1_000_000
end
```
```

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

### Subscribe to Connection Pool Setup Event

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/instrumentation.md

This snippet shows how to subscribe to the 'connection_pool.setup' event to log the initialized pool size. It also demonstrates how to set up the connection pool.

```ruby
WaterDrop::Instrumentation::Monitor.new.subscribe("connection_pool.setup") do |event|
  puts "Pool initialized with size #{event[:size]}"
end

WaterDrop::ConnectionPool.setup(size: 10) { |config| ... }
```

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

### Instance Method: initialize

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Creates a new, independent connection pool instance. Similar to the global `setup`, it allows configuring pool size, timeout, and producer specifics.

```APIDOC
## Instance Method: initialize

Creates a new connection pool instance.

### Signature:
```ruby
def initialize(size: 5, timeout: 5000, &producer_config)
```

### Parameters:
#### Query Parameters
- **size** (Integer) - Optional - Default: 5 - Number of producers in the pool
- **timeout** (Integer) - Optional - Default: 5000 - Timeout in ms to wait for available producer
- **producer_config** (Proc) - Required - Block to configure each producer: `\|config, index\|`

### Returns:
ConnectionPool instance

### Example:
```ruby
pool = WaterDrop::ConnectionPool.new(size: 8) do |config, index|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': "worker-" + index.to_s
  }
end
```
```

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

### Sync Produce Delivery Report Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/types.md

Example of how to access attributes from a DeliveryReport object obtained after a synchronous produce operation.

```ruby
report = producer.produce_sync(topic: 'events', payload: 'data')
puts "Partition: #{report.partition}, Offset: #{report.offset}"
```

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

### Handling Re-configuration Errors

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/errors.md

Catch `ProducerAlreadyConfiguredError` when attempting to call `setup` on an already configured producer. Create a new producer instance for different configurations instead of re-calling `setup`.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

begin
  producer.setup do |config|  # Error: already configured
    config.kafka = { 'bootstrap.servers': 'other:9092' }
  end
rescue WaterDrop::Errors::ProducerAlreadyConfiguredError => e
  puts "Error: #{e.message}"
end
```

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

### Custom Monitor and Logging Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Integrate a custom monitor and logger for detailed event tracking and logging. Subscribe to specific events using the monitor.

```ruby
monitor = WaterDrop::Instrumentation::Monitor.new

producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.monitor = monitor
  config.logger = Logger.new('waterdrop.log')
  config.deliver = true
end

monitor.subscribe("message.produced_sync") do |event|
  puts "Produced: #{event[:message][:topic]}"
end
```

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

### Global Connection Pool Setup and Usage

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Explains how to set up and use the global singleton connection pool. Multiple calls to `WaterDrop::ConnectionPool.with` will utilize the same pool instance.

```ruby
WaterDrop::ConnectionPool.setup(size: 10) { |config| ... }

# These use the same pool:
WaterDrop::ConnectionPool.with { |p| p.produce_sync(...) }
WaterDrop::ConnectionPool.with { |p| p.produce_sync(...) }

# And this accesses it directly:
pool = WaterDrop::ConnectionPool.default_pool
```

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

### Basic Global Pool Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Configures the global connection pool with a specified size. Each producer in the pool will share the provided Kafka configuration.

```ruby
WaterDrop::ConnectionPool.setup(size: 10) do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092'
  }
  config.deliver = true
end
```

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

### Development Configuration Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Configure the producer for development or testing environments. Set `deliver` to `false` to prevent actual message sending and enable debug logging.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.deliver = false  # Don't actually send
  config.logger = Logger.new(STDOUT, level: Logger::DEBUG)
end
```

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

### Batch/Buffering Production Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Buffer multiple messages to be sent efficiently in a single batch. Use `flush_sync` to send all buffered messages at once.

```ruby
producer.buffer(topic: 'events', payload: 'msg1')
producer.buffer(topic: 'events', payload: 'msg2')
producer.buffer(topic: 'events', payload: 'msg3')
handles = producer.flush_sync  # Send all at once
```

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

### Install WaterDrop Gem

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Add the WaterDrop gem to your Gemfile and run bundle install to install the library.

```ruby
# Gemfile
gem 'waterdrop'
```

```bash
bundle install
```

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

### Asynchronous Production Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Employ asynchronous production when immediate delivery confirmation is not required, such as for logging or analytics. The method returns immediately.

```ruby
handle = producer.produce_async(topic: 'events', payload: 'data')
# Returns immediately
```

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

### Setup Connection Pool for Multi-Threading

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Set up a connection pool for managing Kafka producers across multiple threads. This is the recommended approach for concurrent environments.

```ruby
WaterDrop::ConnectionPool.setup(size: 10) do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.deliver = true
end

# In worker threads
WaterDrop::ConnectionPool.with do |producer|
  producer.produce_sync(topic: 'events', payload: 'data')
end
```

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

### Handling Unconfigured Producer Errors

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/errors.md

Catch `ProducerNotConfiguredError` when attempting to use a producer before calling `setup`. Ensure `setup` is called with valid configuration before any produce operations.

```ruby
producer = WaterDrop::Producer.new  # No configuration
begin
  producer.produce_sync(topic: 'events', payload: 'data')
rescue WaterDrop::Errors::ProducerNotConfiguredError => e
  puts "Error: #{e.message}"
end
```

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

### Producer: Setup Configuration

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/INDEX.md

Configures the producer with specified settings. This method initializes the producer with all necessary options, including Kafka connection details and WaterDrop-specific parameters.

```APIDOC
## setup

### Description
Configures the producer with the provided settings.

### Parameters
#### Request Body
- **config** (Config) - Required - The configuration object for the producer.
```

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

### Development/Testing Producer Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Configure a producer for development or testing by setting `config.deliver` to `false`. This runs validations without sending messages to Kafka.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.deliver = false  # Don't actually send to Kafka
  config.logger = Logger.new(STDOUT, level: Logger::DEBUG)
end

# Runs all validations but doesn't contact Kafka
producer.produce_sync(topic: 'test', payload: 'data')
```

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

### Transactional Producer Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Configure a producer for transactional messaging. Set a `transactional.id` and use the `producer.transaction` block for atomic message delivery.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'my-app-producer-1'
  }
  config.deliver = true
  config.max_wait_timeout = 180_000
end

producer.transaction do
  producer.produce_async(topic: 'events', payload: 'msg1')
  producer.produce_async(topic: 'events', payload: 'msg2')
end

producer.close
```

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

### Idempotent Producer Setup

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Enable idempotence by setting `enable.idempotence` to `true`. Configure retry attempts for fatal errors to ensure messages are not duplicated.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'enable.idempotence': true
  }
  config.reload_on_idempotent_fatal_error = true
  config.max_attempts_on_idempotent_fatal_error = 5
end

producer.produce_sync(topic: 'critical', payload: 'important')
```

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

### Message Contract Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/types.md

Demonstrates the structure of a message hash for produce operations, including topic, payload, key, partition, timestamp, and headers.

```ruby
message = {
  topic: 'users',
  payload: '{"id": 123, "name": "John"}',
  key: 'user_123',
  partition: 0,
  timestamp: Time.now,
  headers: {
    'correlation-id' => 'req_123',
    'tags' => ['important', 'user-update']
  }
}

producer.produce_sync(message)
```

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

### StatisticsNotEnabledError Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/errors.md

Demonstrates attempting to subscribe to statistics after the librdkafka client has been initialized, leading to StatisticsNotEnabledError.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Client is lazy-initialized on first use
producer.produce_sync(topic: 'events', payload: 'data')

# Now too late to subscribe to statistics
begin
  producer.monitor.subscribe('statistics.emitted') do |event|
    puts event
  end
rescue WaterDrop::Errors::StatisticsNotEnabledError => e
  puts "Statistics were not enabled at client build time"
end
```

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

### Middleware Error Handling Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Shows how exceptions raised within middleware steps propagate to the caller, allowing for centralized error handling.

```ruby
middleware.append(->(msg) do
  raise ArgumentError, "Invalid topic" unless valid_topic?(msg[:topic])
  msg
end)

begin
  producer.produce_sync(topic: 'invalid!', payload: 'data')
rescue ArgumentError => e
  puts "Middleware validation failed: #{e.message}"
end
```

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

### PollerError Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/errors.md

Illustrates the scenario where PollerError might be raised when using file descriptor-based polling mode.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.polling.mode = :fd
end

# PollerError would be raised in polling thread
```

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

### Subscribe to Transaction Started Event

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/instrumentation.md

Use this snippet to subscribe to the 'transaction.started' event and log the producer ID when a transaction begins.

```ruby
monitor.subscribe("transaction.started") do |event|
  puts "Transaction started for #{event[:producer_id]}"
end
```

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

### Global Pool Setup with Transactional Producers

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Sets up the global connection pool for transactional producers. Each producer is assigned a unique transactional ID based on hostname and pool index.

```ruby
WaterDrop::ConnectionPool.setup(size: 5) do |config, index|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': "app-#{Socket.gethostname}-#{index}"
  }
  config.deliver = true
end
```

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

### Producer Initialization

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Initializes a new WaterDrop Producer instance. Configuration can be provided immediately via a block or set later using the `setup` method.

```APIDOC
## Producer Initialization

### Constructor

**Signature:**
```ruby
def initialize(&block)
```

**Parameters:**
| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| block | Proc | No | nil | Optional configuration block passed to `setup` |

**Returns:** `Producer` instance

**Behavior:** Creates an uninitialized producer. If a block is provided, the producer is immediately configured via `setup`. Otherwise, you must call `setup` explicitly later.

**Example:**
```ruby
# With immediate configuration
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Without configuration (configure later)
producer = WaterDrop::Producer.new
producer.setup do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end
```
```

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

### Global Pool Setup with Custom Logging

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Configures the global connection pool with custom logging for each producer. Log files are named based on the producer's index in the pool.

```ruby
WaterDrop::ConnectionPool.setup(size: 15) do |config, index|
  config.kafka = { 'bootstrap.servers': ENV['KAFKA_BROKERS'] }
  config.logger = Logger.new("producer_#{index}.log")
  config.deliver = true
end
```

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

### Producer Variant with Timeout

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/types.md

Example of creating a producer variant with a custom `max_wait_timeout` for faster sync operations on a specific topic.

```ruby
# Variant with different timeout
fast_variant = producer.variant(max_wait_timeout: 5000)
fast_variant.produce_sync(topic: 'fast', payload: 'quick')
```

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

### Transactional Production Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Implement transactional production for all-or-nothing message delivery semantics. Messages within a transaction are either committed atomically or rolled back on error.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': 'unique-id'
  }
end

producer.transaction do
  producer.produce_async(topic: 'input', payload: 'msg')
  producer.produce_async(topic: 'output', payload: 'result')
  # Committed atomically or rolled back on error
end
```

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

### Using a Custom Poller Instance

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

This example shows how to provide a custom `Poller` instance to a producer, allowing for isolated polling management instead of using the default singleton.

```APIDOC
## Using Custom Poller Instance

By default, FD mode uses the singleton `Poller.instance`. You can provide a custom poller for isolation:

```ruby
custom_poller = WaterDrop::Polling::Poller.new

producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.polling.mode = :fd
  config.polling.poller = custom_poller
end
```
```

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

### Initialize Producer without Immediate Configuration

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Create a WaterDrop::Producer instance without initial configuration. The setup must be called explicitly later to configure the producer.

```ruby
# With immediate configuration
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Without configuration (configure later)
producer = WaterDrop::Producer.new
producer.setup do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end
```

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

### Synchronous Production Example

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Use synchronous production when you need confirmation that a message has been delivered before proceeding. This is useful for critical messages or maintaining transactional consistency.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

report = producer.produce_sync(topic: 'events', payload: 'data')
# Blocks until delivered or timeout
```

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

### Enable FD Polling Mode

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Configure a producer to use the efficient file-descriptor-based polling mode. This example shows how to enable FD mode and optionally customize its behavior with `max_time` and `periodic_poll_interval`.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  
  # Enable FD-based polling
  config.polling.mode = :fd
  
  # Optional: customize FD polling behavior
  config.polling.fd.max_time = 100              # Max ms per IO.select
  config.polling.fd.periodic_poll_interval = 100  # Periodic poll interval
end
```

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

### Message Transformation with Middleware

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/INDEX.md

Apply custom transformations to messages before they are sent by appending middleware to the producer's configuration. This example adds a timestamp to each message.

```ruby
producer.config.middleware.append(
  ->(msg) { msg.merge(timestamp: Time.now.to_i) }
)
```

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

### Configure Producer for High-Reliability

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Configure a producer for high-reliability by ensuring all replicas acknowledge messages and enabling idempotence and transactions. This setup guarantees message delivery.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'request.required.acks': -1,      # All replicas ack
    'enable.idempotence': true,       # Exactly-once
    'transactional.id': 'unique-id'   # Transactions
  }
  config.reload_on_idempotent_fatal_error = true
end

producer.transaction do
  producer.produce_sync(topic: 'critical', payload: 'important')
end
```

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

### Transaction Monitoring for Producer

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/instrumentation.md

Create a listener to log the start, commit, and abort events of Kafka transactions. Subscribe this listener to the producer's monitor.

```ruby
class TransactionMonitor
  def on_transaction_started(event)
    puts "Transaction started: #{event[:producer_id]}"
  end
  
  def on_transaction_committed(event)
    puts "Transaction committed: #{event[:producer_id]}"
  end
  
  def on_transaction_aborted(event)
    puts "Transaction aborted: #{event[:producer_id]}"
  end
end

producer.monitor.subscribe(TransactionMonitor.new)
```

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

### Configure Producer for High-Throughput

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Configure a producer for high-throughput scenarios by enabling batching, setting a batch size, and using compression. This setup prioritizes sending many messages quickly.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': ENV['KAFKA_BROKERS'],
    'linger.ms': 100,           # Wait for batching
    'batch.size': 32768,
    'compression.type': 'snappy'
  }
  config.polling.mode = :fd  # Single thread, multiple producers
  config.idle_disconnect_timeout = 60_000  # Save TCP connections
end
```

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

### Producer Initialization and Basic Usage

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Demonstrates how to create and configure a WaterDrop producer and send messages synchronously and asynchronously.

```APIDOC
## Producer Initialization and Basic Usage

### Description

This section shows how to initialize a `WaterDrop::Producer` with basic Kafka configuration and demonstrates sending messages using both synchronous (`produce_sync`) and asynchronous (`produce_async`) methods.

### Method

`WaterDrop::Producer.new` for initialization, `produce_sync` and `produce_async` for sending messages.

### Endpoint

N/A (Ruby Gem API)

### Parameters

#### Producer Initialization Block:
- `config` (block) - A block to configure the producer.
  - `config.kafka` (hash) - Required - Kafka broker connection details. Example: `{'bootstrap.servers': 'localhost:9092'}`.
  - `config.deliver` (boolean) - Optional - If `true`, messages are actually sent. Defaults to `true`.

#### `produce_sync` Parameters:
- `topic` (string) - Required - The Kafka topic to send the message to.
- `payload` (string) - Required - The message content.
- `key` (string) - Optional - The message key.

#### `produce_async` Parameters:
- `topic` (string) - Required - The Kafka topic to send the message to.
- `payload` (string) - Required - The message content.

### Request Example

```ruby
# Create and configure a producer
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092'
  }
  config.deliver = true
end

# Synchronous production
report = producer.produce_sync(
  topic: 'events',
  payload: 'user signup event',
  key: 'user_123'
)

puts "Message at partition #{report.partition}, offset #{report.offset}"

# Asynchronous production
handle = producer.produce_async(
  topic: 'events',
  payload: 'user event'
)

# Check delivery later
report = handle.create_result

# Cleanup
producer.close
```

### Response

#### Success Response (`produce_sync`):
- `partition` (integer) - The partition the message was written to.
- `offset` (integer) - The offset of the message within the partition.

#### Success Response (`produce_async`):
- Returns a `handle` object that can be used to retrieve the delivery report later via `handle.create_result`.

#### Response Example (`produce_sync`):

```json
{
  "partition": 0,
  "offset": 123
}
```

#### Response Example (`produce_async`):

```json
{
  "handle": "<WaterDrop::Producer::AsyncHandle>"
}
```
```

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

### Accessing and Providing Custom Monitor

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/instrumentation.md

Shows how to access the producer's monitor and how to provide a custom monitor instance during producer initialization.

```ruby
# Access the monitor
monitor = producer.monitor

# Or provide your own
custom_monitor = WaterDrop::Instrumentation::Monitor.new
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.monitor = custom_monitor
end
```

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

### Get Producer ID

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Retrieves the unique identifier for the producer instance. This UUID can be useful for logging and debugging.

```ruby
puts producer.id  # "waterdrop-abc123def456"
```

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

### Initialize Producer with Default Monitor

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/instrumentation.md

Demonstrates how to create a producer with the default WaterDrop::Instrumentation::Monitor. The monitor is automatically created and available on every producer instance.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.monitor = WaterDrop::Instrumentation::Monitor.new  # Default
end
```

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

### Get Connection Pool Size

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Returns the total number of producers configured in the connection pool. This is set during initialization.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 10) { |config| }
puts pool.size  # => 10
```

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

### Middleware Execution Order

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Demonstrates how to append and prepend middleware steps to control their execution order. Steps are executed in the order they are added, with prepended steps running first.

```ruby
middleware = WaterDrop::Middleware.new

middleware.append(->(msg) { msg.merge(step: 1) })
middleware.append(->(msg) { msg.merge(order: msg[:step]) })  # order will be 1

middleware.prepend(->(msg) { msg.merge(step: 0) })  # Runs first

# Order: prepended step (0), appended step 1, appended step with order
```

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

### Basic Multi-Threaded Usage of Connection Pool

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Demonstrates how to initialize a connection pool and use it across multiple threads. Each thread acquires a producer from the pool to send messages.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 5) do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Spawn worker threads
5.times do |i|
  Thread.new(i) do |thread_id|
    pool.with do |producer|
      producer.produce_sync(
        topic: 'events',
        payload: "message from thread #{thread_id}"
      )
    end
  end
end

Thread.list.each { |t| t.join }
```

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

### Subscribe to Statistics Before Client Initialization

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/errors.md

Shows the correct way to subscribe to statistics by doing so before the producer client is first used.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Subscribe BEFORE first use (before client is initialized)
producer.monitor.subscribe('statistics.emitted') do |event|
  puts "Stats: #{event}"
end

# Now use producer
producer.produce_sync(topic: 'events', payload: 'data')
```

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

### Get Producer Status

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Retrieves the current lifecycle status of the producer. Use the `active?` method to check if the producer is ready for operations.

```ruby
puts producer.status.to_s  # "configured"
puts "Active" if producer.status.active?
```

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

### Basic Producer Usage

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/README.md

Create and configure a producer instance, send messages synchronously and asynchronously, and then close the producer to release resources.

```ruby
# Create and configure a producer
producer = WaterDrop::Producer.new do |config|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092'
  }
  config.deliver = true
end

# Synchronous production
report = producer.produce_sync(
  topic: 'events',
  payload: 'user signup event',
  key: 'user_123'
)

puts "Message at partition #{report.partition}, offset #{report.offset}"

# Asynchronous production
handle = producer.produce_async(
  topic: 'events',
  payload: 'user event'
)

# Check delivery later
report = handle.create_result

# Cleanup
producer.close
```

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

### Connection Pool Configuration and Usage

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Set up a connection pool for managing multiple producer instances. Use `WaterDrop::ConnectionPool.with` to acquire and use a producer from the pool.

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

WaterDrop::ConnectionPool.with do |producer|
  producer.produce_sync(topic: 'events', payload: 'data')
end
```

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

### Get Poller Instance ID

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Retrieve the unique identifier assigned to a `Poller` instance upon initialization. This ID is useful for logging and debugging.

```ruby
poller = WaterDrop::Polling::Poller.new
puts poller.id  # => 0, 1, 2, etc.
```

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

### Initialize Producer with Configuration Block

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Use a configuration block when creating a new WaterDrop producer to set Kafka connection details and other producer options.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.deliver = true
  config.max_payload_size = 1_000_012
end
```

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

### Environment-Based Connection Pool Configuration

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Shows how to configure the connection pool using environment variables. This approach allows for flexible configuration across different deployment environments without code changes.

```ruby
WaterDrop::ConnectionPool.setup(size: ENV.fetch('KAFKA_POOL_SIZE', 5).to_i) do |config|
  config.kafka = {
    'bootstrap.servers': ENV.fetch('KAFKA_BROKERS', 'localhost:9092'),
    'security.protocol': ENV.fetch('KAFKA_SECURITY_PROTOCOL', 'PLAINTEXT'),
    'sasl.mechanism': ENV.fetch('KAFKA_SASL_MECHANISM', 'PLAIN'),
    'sasl.username': ENV.fetch('KAFKA_USERNAME', ''),
    'sasl.password': ENV.fetch('KAFKA_PASSWORD', '')
  }
  config.deliver = !ENV['KAFKA_DISABLED'].present?
  config.logger = Logger.new(STDOUT, level: Logger::WARN)
end
```

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

### Instance Method: with

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Executes a block of code using a producer obtained from this specific connection pool instance. It handles acquiring and releasing producers from the pool.

```APIDOC
## Instance Method: with

Executes a block with a producer from this pool.

### Signature:
```ruby
def with(&block)
```

### Parameters:
#### Request Body
- **block** (Proc) - Required - Block receiving a producer from pool

### Returns:
Result of the block

### Timeout:
Raises error if no producer available within timeout

### Example:
```ruby
pool = WaterDrop::ConnectionPool.new(size: 5) do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

result = pool.with do |producer|
  producer.produce_sync(topic: 'events', payload: 'data')
end

# Nesting works (may block if all producers checked out)
pool.with do |producer1|
  pool.with do |producer2|
    # producer1 and producer2 may be same or different
  end
end
```
```

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

### Count Registered Producers

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Get the number of producers currently registered with a specific `Poller` instance. This count updates as producers are initialized and used with the poller.

```ruby
poller = WaterDrop::Polling::Poller.new

producer1 = WaterDrop::Producer.new do |config|
  config.polling.mode = :fd
  config.polling.poller = poller
end

puts poller.count  # => 0 (not yet used)

producer1.produce_sync(topic: 'events', payload: 'data')

puts poller.count  # => 1 (producer registered)

producer2 = WaterDrop::Producer.new do |config|
  config.polling.mode = :fd
  config.polling.poller = poller
end

producer2.produce_sync(topic: 'events', payload: 'data')

puts poller.count  # => 2
```

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

### Append Various Callable Middleware Steps

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Demonstrates appending different types of callable objects (lambda, Proc, method, custom class instance) as middleware steps.

```ruby
middleware = WaterDrop::Middleware.new

# Lambda
middleware.append(->(msg) { msg.merge(source: 'app') })

# Proc
middleware.append(Proc.new { |msg| msg.merge(version: 1) })

# Method
def add_metadata(msg)
  msg.merge(processed_at: Time.now)
end
middleware.append(method(:add_metadata))

# Custom class with call method
class AddHeaders
  def call(message)
    message.merge(headers: { 'x-source' => 'middleware' })
  end
end
middleware.append(AddHeaders.new)
```

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

### Producer Variant with Compression

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/types.md

Demonstrates creating a producer variant with per-topic configuration for message compression using `topic_config`.

```ruby
# Variant with compression
compressed = producer.variant(
  topic_config: {
    'compression.codec': 'snappy',
    'compression.level': 5
  }
)
compressed.produce_sync(topic: 'events', payload: large_data)
```

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

### Create Multiple Variants for Different Use Cases

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/variant.md

Demonstrates creating several producer variants, each with distinct configurations for timeouts and acknowledgments, all sharing the same underlying Kafka client.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.max_wait_timeout = 180_000
end

fast = producer.variant(max_wait_timeout: 5000)
reliable = producer.variant(
  topic_config: { 'request.required.acks': -1 },
  max_wait_timeout: 60_000
)
compressed = producer.variant(
  topic_config: { 'compression.codec': 'snappy' }
)

# Use them all
fast.produce_sync(topic: 'metrics', payload: data)
reliable.produce_sync(topic: 'critical', payload: data)
compressed.buffer_many(messages)
compressed.flush_sync

# All share the same underlying Kafka client
puts producer.status.connected?  # true for all variants
```

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

### Singleton Poller Instance

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Get the default singleton poller instance when using FD mode. This is automatically shared across all producers configured with FD mode.

```ruby
poller = WaterDrop::Polling::Poller.instance
```

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

### Dynamic Configuration per Pool Member

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Illustrates how to configure individual producers within the connection pool with different settings. This allows for varied producer behaviors, such as different acknowledgment levels or compression types.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 3) do |config, index|
  case index
  when 0
    # Fast, low-ack producer
    config.kafka = {
      'bootstrap.servers': 'localhost:9092',
      'request.required.acks': 1
    }
  when 1
    # Reliable, all-ack producer
    config.kafka = {
      'bootstrap.servers': 'localhost:9092',
      'request.required.acks': -1
    }
  when 2
    # Compressed producer
    config.kafka = {
      'bootstrap.servers': 'localhost:9092',
      'compression.type': 'snappy'
    }
  end
end

# Different pool members have different configs
pool.with do |producer|
  # Might get fast, reliable, or compressed producer
  producer.produce_sync(topic: 'events', payload: 'data')
end
```

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

### Create and Append Middleware Step

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Instantiate the middleware and append a lambda function to add a timestamp to messages.

```ruby
middleware = WaterDrop::Middleware.new
middleware.append(->(msg) { msg.merge(timestamp: Time.now.to_i) })
```

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

### Initialize Middleware

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Middleware is automatically created during producer configuration. You can access and modify it using config.middleware.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
  config.middleware = WaterDrop::Middleware.new  # Default, already created
end

# Access the middleware
producer.config.middleware.append(step)
```

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

### Thread Blocking with Connection Pool

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Demonstrates how calling `with` on a connection pool blocks the current thread until a producer is available. Subsequent threads will wait for producers held by earlier threads.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 1) { |config| ... }

Thread.new do
  pool.with do |producer|
    sleep(10)  # Hold producer for 10 seconds
  end
end

Thread.new do
  sleep(1)
  pool.with do |producer|  # Blocks for ~9 seconds waiting for first thread to finish
    producer.produce_sync(topic: 'events', payload: 'data')
  end
end
```

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

### Get Available Producers in Pool

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Returns the count of producers currently available for use. Producers are available when not checked out by a thread. This count reflects the number of producers that can be acquired via `pool.with`.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 10) { |config| }

puts pool.available  # => 10 (all available)

pool.with do |producer|
  puts pool.available  # => 9 (one checked out)
end

puts pool.available  # => 10 (returned)
```

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

### Create Producer Variant with Topic Configuration

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/producer.md

Creates a producer variant with per-topic configuration, such as compression codec. This allows fine-tuning settings for specific topics.

```ruby
# Variant with topic-specific config
compressed = producer.variant(
  topic_config: { 'compression.codec': 'snappy' }
)
compressed.produce_sync(topic: 'events', payload: 'data')
```

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

### Get Variant's Topic Configuration

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/variant.md

The `topic_config` attribute returns the hash of per-topic librdkafka settings applied to this variant. These settings override the producer's default topic configurations.

```ruby
config = { 'compression.codec': 'snappy', 'compression.level': 5 }
variant = producer.variant(topic_config: config)
puts variant.topic_config  # => { compression.codec: 'snappy', ... }
```

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

### Get Variant's Max Wait Timeout

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/variant.md

The `max_wait_timeout` attribute returns the specific timeout in milliseconds configured for this variant. If not explicitly set, it defaults to the underlying producer's setting.

```ruby
fast = producer.variant(max_wait_timeout: 5000)
puts fast.max_wait_timeout  # => 5000

default = producer.variant
puts default.max_wait_timeout  # => producer.config.max_wait_timeout
```

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

### Multi-Step Pipeline for Message Production

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/middleware.md

Demonstrates a multi-step middleware pipeline that adds metadata, correlation IDs, headers, and serializes JSON payloads sequentially.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

# Step 1: Add metadata
producer.config.middleware.append(->(msg) do
  msg.merge(
    produced_at: Time.now.to_i,
    source_version: '1.0'
  )
end)

# Step 2: Add correlation if not present
producer.config.middleware.append(->(msg) do
  msg.merge(
    correlation_id: msg[:correlation_id] || SecureRandom.uuid
  )
end)

# Step 3: Add headers
producer.config.middleware.append(->(msg) do
  msg.merge(
    headers: (msg[:headers] || {}).merge(
      'x-produced-at' => msg[:produced_at].to_s,
      'x-source-version' => msg[:source_version]
    )
  )
end)

# Step 4: Serialize payload if hash
producer.config.middleware.append(->(msg) do
  if msg[:payload].is_a?(Hash)
    msg.merge(payload: msg[:payload].to_json)
  else
    msg
  end
end)

# Now produce
producer.produce_sync(
  topic: 'events',
  payload: { user_id: 123, action: 'login' }
  # Middleware will:
  # 1. Add timestamps and version
  # 2. Add correlation_id
  # 3. Add headers with metadata
  # 4. Convert payload to JSON
)
```

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

### Initializing a New Connection Pool Instance

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Creates a new, independent connection pool instance with specified size and producer configuration. This is useful for creating isolated pools.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 8) do |config, index|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': "worker-#{index}"
  }
end
```

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

### Transactional Producer Pool Usage

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Shows how to set up a connection pool for transactional producers. Each producer can be configured with a unique transactional ID, suitable for multi-message transactions.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 3) do |config, index|
  config.kafka = {
    'bootstrap.servers': 'localhost:9092',
    'transactional.id': "my-app-#{Socket.gethostname}-#{index}"
  }
  config.deliver = true
end

# Each thread gets its own transactional producer from pool
Thread.new do
  pool.with do |producer|
    producer.transaction do
      producer.produce_async(topic: 'input', payload: 'process')
      producer.produce_async(topic: 'output', payload: 'result')
    end
  end
end
```

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

### Nesting Usage of a Specific Connection Pool Instance

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/connection-pool.md

Demonstrates nesting calls to `with` on a specific connection pool instance. This can lead to blocking if all producers are checked out.

```ruby
pool = WaterDrop::ConnectionPool.new(size: 5) do |config|
  config.kafka = { 'bootstrap.servers': 'localhost:9092' }
end

pool.with do |producer1|
  pool.with do |producer2|
    # producer1 and producer2 may be same or different
  end
end
```

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

### Initial Delay in FD Mode

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/api-reference/polling.md

Illustrates the potential initial delay (up to 1 second) for newly registered producers in FD mode before their first poll cycle. This is due to how the IO list is rebuilt.

```ruby
producer = WaterDrop::Producer.new do |config|
  config.polling.mode = :fd
end

# Might take up to 1 second for first delivery to happen
producer.produce_async(topic: 'events', payload: 'data')
```

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

### Configure Required Kafka Bootstrap Servers

Source: https://github.com/karafka/waterdrop/blob/master/_autodocs/configuration.md

Specify the Kafka cluster's bootstrap servers. This is a mandatory setting for establishing a connection.

```ruby
config.kafka = {
  'bootstrap.servers': 'localhost:9092,localhost:9093'
}
```