### Install ruby-kafka Gem

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Add this line to your application's Gemfile to install the ruby-kafka gem. Then execute `bundle install`.

```ruby
gem 'ruby-kafka'
```

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

### Install ruby-kafka Gem Manually

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Alternatively, install the ruby-kafka gem yourself using the `gem install` command.

```bash
$ gem install ruby-kafka
```

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

### Configure Topic Subscription Start Point

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Control whether to consume messages from the beginning or only new messages upon subscribing to a topic. This setting is only effective when a consumer group initially starts consuming from a topic.

```ruby
# Consume messages from the very beginning of the topic. This is the default.
consumer.subscribe("users", start_from_beginning: true)

# Only consume new messages.
consumer.subscribe("notifications", start_from_beginning: false)
```

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

### Subscribe with Start Position

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Control whether to start consuming from the beginning of a topic or only new messages.

```APIDOC
## Subscribe with Start Position

### Description
Configure the starting position for consuming messages from a topic. This setting applies only when the consumer initially connects to a topic or when no offsets have been previously committed for the consumer group.

### Method
`Consumer#subscribe`

### Endpoint
N/A (Method call)

### Parameters
#### Path Parameters
- **topic_name** (String) - Required - The name of the topic to subscribe to.
#### Query Parameters
- **start_from_beginning** (Boolean) - Optional - If true, consume from the earliest offset; if false, consume only new messages. Defaults to true.

### Request Example
```ruby
# Consume messages from the very beginning of the topic.
consumer.subscribe("users", start_from_beginning: true)

# Only consume new messages.
consumer.subscribe("notifications", start_from_beginning: false)
```
```

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

### Tune Consumer for Throughput with min_bytes and max_wait_time

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Optimize for throughput by configuring `min_bytes` and `max_wait_time` in `each_message`. This example waits up to 5 seconds for data, preferring to fetch at least 5KB at a time.

```ruby
# Waits for data for up to 5 seconds on each broker, preferring to fetch at least 5KB at a time.
# This can wait up to num brokers * 5 seconds.
consumer.each_message(min_bytes: 1024 * 5, max_wait_time: 5) do |message|
  # ...
end
```

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

### Produce JSON Encoded Data

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Serialize event data using JSON and produce it to a Kafka topic. This example demonstrates how to integrate JSON serialization with message production.

```ruby
require "json"

# ...

event = {
  "name" => "pageview",
  "url" => "https://example.com/posts/123",
  # ...
}

data = JSON.dump(event)

producer.produce(data, topic: "events")
```

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

### Initialize Kafka Client with Seed Brokers

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Instantiate a Kafka client by providing a list of seed broker addresses. The `client_id` is recommended for logging and metrics.

```ruby
require "kafka"

# The first argument is a list of "seed brokers" that will be queried for the full
# cluster topology. At least one of these *must* be available. `client_id` is
# used to identify this client in logs and metrics. It's optional but recommended.
kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], client_id: "my-application")
```

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

### Create a topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Creates a new topic with specified configurations for partitions, replication factor, and custom settings.

```APIDOC
## Create a topic

### Description
Creates a new topic. By default, the new topic has 1 partition, replication factor 1 and default configs from the brokers. Those configurations are customizable.

### Method
`kafka.create_topic(topic_name, num_partitions: 1, replication_factor: 1, config: {})`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topic_name** (String) - The name of the topic to create.
- **num_partitions** (Integer) - Optional. The number of partitions for the new topic. Defaults to 1.
- **replication_factor** (Integer) - Optional. The replication factor for the new topic. Defaults to 1.
- **config** (Hash) - Optional. A hash of topic configurations. Example: `{"max.message.bytes" => 100000}`.

### Response
None (Success is indicated by lack of error).

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.create_topic("topic")

kafka.create_topic("topic",
  num_partitions: 3,
  replication_factor: 2,
  config: {
    "max.message.bytes" => 100000
  }
)
```
```

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

### Initialize Kafka Client

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Instantiate the Kafka client with broker addresses and optional SASL OAuth token provider.

```ruby
client = Kafka.new(
  ["kafka1:9092"],
  sasl_oauth_token_provider: TokenProvider.new
)
```

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

### Produce with Custom Partitioning Scheme

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use a custom partitioning scheme by assigning a partition to an event before producing it.

```ruby
partition = PartitioningScheme.assign(partitions, event)

producer.produce(event, topic: "events", partition: partition)
```

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

### Create Kafka Topic with Custom Configuration

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Creates a new Kafka topic with specified number of partitions, replication factor, and custom configurations.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.create_topic("topic",
  num_partitions: 3,
  replication_factor: 2,
  config: {
    "max.message.bytes" => 100000
  }
)
```

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

### Create Kafka Topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Creates a new Kafka topic with default settings (1 partition, replication factor 1).

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.create_topic("topic")
```

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

### Initialize Kafka Client with Hostname Resolution

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Initialize a Kafka client using a hostname for seed brokers and enabling `resolve_seed_brokers`. This is an alternative to providing IP addresses directly.

```ruby
kafka = Kafka.new("seed-brokers:9092", client_id: "my-application", resolve_seed_brokers: true)
```

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

### Set up Asynchronous Kafka Producer

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Configure an asynchronous producer to deliver buffered messages at a specified interval. Ensure the producer is shut down gracefully upon exiting.

```ruby
$kafka_producer = $kafka.async_producer(
  delivery_interval: 10,
)

at_exit { $kafka_producer.shutdown }
```

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

### Configure Ruby Kafka Client with SSL

Source: https://github.com/zendesk/ruby-kafka/wiki/Creating-X509-certificates-from-JKS-format

This Ruby code demonstrates how to initialize the Kafka client with SSL certificates in PEM format for secure communication.

```ruby
kafka = Kafka.new(
  ssl_ca_cert: File.read('ca_cert.pem'),
  ssl_client_cert: File.read('client_cert.pem'),
  ssl_client_cert_key: File.read('client_cert_key.pem'),
  # ...
)
```

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

### Create Async Producer with Delivery Options

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Instantiate an asynchronous producer with specific delivery thresholds and intervals. Ensure messages are delivered within the specified buffer size and time limits.

```ruby
producer = kafka.async_producer(
  # Trigger a delivery once 100 messages have been buffered.
  delivery_threshold: 100,

  # Trigger a delivery every 30 seconds.
  delivery_interval: 30,
)

producer.produce("hello", topic: "greetings")
```

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

### Configure Datadog Reporter Namespace, Host, and Port

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Enable and configure the Datadog reporter by setting its namespace, host, and port. The 'kafka/datadog' library must be required to activate the reporter.

```ruby
# This enables the reporter:
require "kafka/datadog"

# Default is "ruby_kafka".
Kafka::Datadog.namespace = "custom-namespace"

# Default is "127.0.0.1".
Kafka::Datadog.host = "statsd.something.com"

# Default is 8125.
Kafka::Datadog.port = 1234
```

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

### Generate Server Keystore and Certificates

Source: https://github.com/zendesk/ruby-kafka/wiki/Creating-X509-certificates-from-JKS-format

This set of commands is used on each Kafka broker to create a server keystore, generate a server certificate signed by the CA, and import both into the keystore for securing inter-broker communication.

```bash
## 1. Create server keystore
keytool -noprompt -keystore kafka.server.keystore.jks -genkey -alias <hostname> -dname "CN=<hostname>, OU=<unit>, O=<org>, L=<loc>, ST=<state>, C=<country>" -storepass foobar -keypass foobar

## 2. Sign server certificate
keytool -noprompt -keystore kafka.server.keystore.jks -alias <hostname> -certreq -file cert-unsigned -storepass foobar
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-unsigned -out cert-signed -days 365 -CAcreateserial -passin pass:foobar

## 3. Import CA and signed server certificate into server keystore
keytool -noprompt -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert -storepass foobar
keytool -noprompt -keystore kafka.server.keystore.jks -alias <hostname> -import -file cert-signed -storepass foobar
```

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

### List All Kafka Topics

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Fetches and returns an array of all topic names available in the Kafka cluster. Requires a Kafka client instance.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.topics
# => ["topic1", "topic2", "topic3"]
```

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

### Create More Partitions for Kafka Topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Increases the number of partitions for an existing topic. The new partition count must be greater than the current one.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.create_partitions_for("topic", num_partitions: 10)
```

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

### Use Proc for Custom Partitioning Logic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Simplify custom partitioning by using a Proc object that handles the partitioning logic directly.

```ruby
partitioner = -> (partition_count, message) { ... }
Kafka.new(partitioner: partitioner, ...)
```

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

### Reproduce Bug with Kafka Instance

Source: https://github.com/zendesk/ruby-kafka/blob/main/ISSUE_TEMPLATE.md

Instantiate a Kafka client to reproduce the reported issue. Replace '...' with your specific Kafka connection details.

```ruby
kafka = Kafka.new(...)

# Please write an example that reproduces the problem you're describing.
```

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

### Configuring Compression for Kafka Producer

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Enable compression by specifying `compression_codec` and optionally `compression_threshold` when creating the producer. Compression is applied to message sets and improves bandwidth and space usage.

```ruby
producer = kafka.producer(
  compression_codec: :snappy,
  compression_threshold: 10,
)
```

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

### Configure Kafka Client with System CA Certificates

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Enables SSL encryption by using the system's default certificate store for CA certificates. The location of the store can typically be found using OpenSSL::X509::DEFAULT_CERT_FILE.

```ruby
kafka = Kafka.new(["kafka1:9092"], ssl_ca_certs_from_system: true)
```

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

### Configure Statsd Reporter Namespace, Host, and Port

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Customize the Statsd reporter's namespace, host, and port. Ensure the 'kafka/statsd' library is required before configuration.

```ruby
require "kafka/statsd"

# Default is "ruby_kafka".
Kafka::Statsd.namespace = "custom-namespace"

# Default is "127.0.0.1".
Kafka::Statsd.host = "statsd.something.com"

# Default is 8125.
Kafka::Statsd.port = 1234
```

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

### Implement Custom Partitioner Class

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Define a custom partitioner by implementing the `call` method in a class. This class should accept `partition_count` and `message` as arguments.

```ruby
class CustomPartitioner
  def call(partition_count, message)
    ...
  end
end
  
partitioner = CustomPartitioner.new
Kafka.new(partitioner: partitioner, ...)
```

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

### Generate CA and Client Certificates in JKS Format

Source: https://github.com/zendesk/ruby-kafka/wiki/Creating-X509-certificates-from-JKS-format

This sequence of commands generates a certificate authority (CA), a client keystore, signs the client certificate with the CA, and imports both into the client keystore. It also imports the CA into client and server truststores.

```bash
## 1. Create certificate authority (CA)
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365 -passin pass:foobar -passout pass:foobar -subj "/CN=<domain>/OU=<unit>/O=<org>/L=<loc>/ST=<state>/C=<country>"
## 2. Create client keystore
keytool -noprompt -keystore kafka.client.keystore.jks -genkey -alias localhost -dname "CN=<domain>, OU=<unit>, O=<org>, L=<loc>, ST=<state>, C=<country>" -storepass foobar -keypass foobar
## 3. Sign client certificate
keytool -noprompt -keystore kafka.client.keystore.jks -alias localhost -certreq -file cert-unsigned -storepass foobar
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-unsigned -out cert-signed -days 365 -CAcreateserial -passin pass:foobar
## 4. Import CA and signed client certificate into client keystore
keytool -noprompt -keystore kafka.client.keystore.jks -alias CARoot -import -file ca-cert -storepass foobar
keytool -noprompt -keystore kafka.client.keystore.jks -alias localhost -import -file cert-signed -storepass foobar
## 5. Import CA into client truststore (only for debugging with Java consumer)
keytool -noprompt -keystore kafka.client.truststore.jks -alias CARoot -import -file ca-cert -storepass foobar
## 6. Import CA into server truststore
keytool -noprompt -keystore kafka.server.truststore.jks -alias CARoot -import -file ca-cert -storepass foobar
```

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

### Fetch configuration for a topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Retrieves the configuration values for specified keys of a given topic. This is an alpha feature.

```APIDOC
## Fetch configuration for a topic (alpha feature)

### Description
Retrieves the configuration values for specified keys of a given topic.

### Method
`kafka.describe_topic(topic_name, config_keys)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topic_name** (String) - The name of the topic to describe.
- **config_keys** (Array<String>) - An array of configuration keys to fetch.

### Response
#### Success Response (Hash)
- `configurations` (Hash) - A hash where keys are configuration names and values are their corresponding string values.

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.describe_topic("topic", ["max.message.bytes", "retention.ms"])
# => {"max.message.bytes"=>"100000", "retention.ms"=>"604800000"}
```
```

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

### Fetch Kafka Topic Configuration

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Retrieves specific configurations for a given topic. This is an alpha feature.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.describe_topic("topic", ["max.message.bytes", "retention.ms"])
# => {"max.message.bytes"=>"100000", "retention.ms"=>"604800000"}
```

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

### Configure Kafka Client with SSL Client Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Sets up client authentication to the Kafka cluster using a client certificate and key. Hostname validation can be disabled by setting `ssl_verify_hostname: false`. Ensure the broker trusts the provided certificates.

```ruby
kafka = Kafka.new(
  ["kafka1:9092"],
  ssl_ca_cert: File.read('my_ca_cert.pem'),
  ssl_client_cert: File.read('my_client_cert.pem'),
  ssl_client_cert_key: File.read('my_client_cert_key.pem'),
  ssl_client_cert_key_password: 'my_client_cert_key_password',
  ssl_verify_hostname: false,
  # ...
)
```

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

### Consumer Configuration for Checkpointing

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Configure offset commit interval, threshold, and retention time when creating a consumer.

```APIDOC
## Consumer Configuration for Checkpointing

### Description
Configure offset commit interval, threshold, and retention time when creating a consumer.

### Method
`kafka.consumer`

### Parameters
#### Query Parameters
- **offset_commit_interval** (Integer) - Optional - The interval in seconds at which offsets are committed.
- **offset_commit_threshold** (Integer) - Optional - The number of messages processed before committing offsets.
- **offset_retention_time** (Integer) - Optional - The duration in seconds for which committed offsets are retained.

### Request Example
```ruby
consumer = kafka.consumer(
  group_id: "some-group",
  offset_commit_interval: 5,
  offset_commit_threshold: 100,
  offset_retention_time: 7 * 60 * 60
)
```
```

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

### Implement Network Topology Aware Partition Assignment

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

This strategy uses the consumer's IP address as user data and can be configured with a custom protocol name. It demonstrates how to access member metadata and potentially use it for assignment, though the actual assignment logic is omitted (`...`).

```ruby
class NetworkTopologyAssignmentStrategy
  def user_data
    Socket.ip_address_list.find(&:ipv4_private?).ip_address
  end

  def call(cluster:, members:, partitions:)
    # Display the pair of the member ID and IP address
    members.each do |id, metadata|
      puts "#{id}: #{metadata.user_data}"
    end

    # Assign partitions considering the network topology
    ...
  end
end
```

```ruby
class NetworkTopologyAssignmentStrategy
  def protocol_name
    "networktopology"
  end

  def user_data
    Socket.ip_address_list.find(&:ipv4_private?).ip_address
  end

  def call(cluster:, members:, partitions:)
    ...
  end
end
```

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

### Implement Custom Partition Assignment Strategy

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Define a custom strategy by implementing the `#call` method. This is useful for assigning more partitions to specific consumers based on external factors like proximity to a database. The `assignment_strategy` option takes an object that responds to `#call`.

```ruby
class CustomAssignmentStrategy
  def initialize(user_data)
    @user_data = user_data
  end

  # Assign the topic partitions to the group members.
  #
  # @param cluster [Kafka::Cluster]
  # @param members [Hash<String, Kafka::Protocol::JoinGroupResponse::Metadata>] a hash
  #   mapping member ids to metadata
  # @param partitions [Array<Kafka::ConsumerGroup::Assignor::Partition>] a list of
  #   partitions the consumer group processes
  # @return [Hash<String, Array<Kafka::ConsumerGroup::Assignor::Partition>] a hash
  #   mapping member ids to partitions.
  def call(cluster:, members:, partitions:)
    ...
  end
end

strategy = CustomAssignmentStrategy.new("some-host-information")
consumer = kafka.consumer(group_id: "some-group", assignment_strategy: strategy)
```

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

### Configure Consumer Checkpointing Options

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Set consumer options to control offset commit frequency, threshold, and retention time. Useful for managing processing guarantees and data retention.

```ruby
consumer = kafka.consumer(
  group_id: "some-group",

  # Increase offset commit frequency to once every 5 seconds.
  offset_commit_interval: 5,

  # Commit offsets when 100 messages have been processed.
  offset_commit_threshold: 100,

  # Increase the length of time that committed offsets are kept.
  offset_retention_time: 7 * 60 * 60
)
```

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

### Load Balanced Partitioning with Fixed Partition Key

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Ensure all messages share the same partition key to direct them to a single partition. This is useful for optimizing producer performance by enabling per-partition message batching.

```ruby
partition_key = rand(100)

producer.produce(msg1, topic: "messages", partition_key: partition_key)
producer.produce(msg2, topic: "messages", partition_key: partition_key)

# ...
```

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

### Create more partitions for a topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Increases the number of partitions for an existing topic. The new partition count must be greater than the current one.

```APIDOC
## Create more partitions for a topic

### Description
After a topic is created, you can increase the number of partitions for the topic. The new number of partitions must be greater than the current one.

### Method
`kafka.create_partitions_for(topic_name, num_partitions:)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topic_name** (String) - The name of the topic to modify.
- **num_partitions** (Integer) - Required. The new total number of partitions for the topic. Must be greater than the current number of partitions.

### Response
None (Success is indicated by lack of error).

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.create_partitions_for("topic", num_partitions: 10)
```
```

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

### Deliver Message Using Kafka Partition Key

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use `partition_key` to deterministically assign messages to partitions. Messages with the same key will always go to the same partition, useful for grouping related events.

```ruby
# Partition keys assign a partition deterministically.
kafka.deliver_message("Hello, World!", topic: "greetings", partition_key: "hello")
```

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

### Configure Kafka Client with PLAIN Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Authenticates to Kafka using the PLAIN mechanism. Requires a username and password. SSL encryption is recommended and typically enforced unless `sasl_over_ssl: false` is set.

```ruby
kafka = Kafka.new(
  ["kafka1:9092"],
  ssl_ca_cert: File.read('/etc/openssl/cert.pem'),
  sasl_plain_username: 'username',
  sasl_plain_password: 'password'
  # ...
)
```

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

### Rails Application Kafka Producer Initialization

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Configure an asynchronous Kafka producer in a Rails initializer to avoid blocking request/response cycles and handle Kafka outages gracefully. The client is configured with broker hosts and the Rails logger.

```ruby
# config/initializers/kafka_producer.rb
require "kafka"

# Configure the Kafka client with the broker hosts and the Rails
# logger.
$kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], logger: Rails.logger)
```

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

### Configure Kafka Client with GSSAPI Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Authenticates to Kafka using GSSAPI. Requires specifying the principal and optionally the keytab file path. It is highly recommended to use SSL encryption alongside SASL.

```ruby
kafka = Kafka.new(
  ["kafka1:9092"],
  sasl_gssapi_principal: 'kafka/kafka.example.com@EXAMPLE.COM',
  sasl_gssapi_keytab: '/etc/keytabs/kafka.keytab',
  # ...
)
```

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

### Writing Messages with AvroTurf

Source: https://github.com/zendesk/ruby-kafka/wiki/Encoding-messages-with-Avro

Encode a message using AvroTurf. This action also registers the schema with the Schema Registry if it's the first time this schema is used with the AvroTurf instance. Ensure your Schema Registry URL is correctly configured.

```ruby
require "avro_turf"
require "kafka"

# You need to pass the URL of your Schema Registry.
avro = AvroTurf::Messaging.new(registry_url: "http://my-registry:8081/")

# Encoding data has the side effect of registering the schema. This only
# happens the first time a schema is used with the instance of `AvroTurf`.
data = avro.encode({ "title" => "hello, world" }, schema_name: "greeting")

# Assuming you've set up `producer`:
producer.produce(data, topic: "greetings")
```

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

### Configure Kafka Client with a Logger

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

It is recommended to configure the Kafka client with a logger to track important operations and errors. Pass a valid logger object during client instantiation. By default, no logging occurs.

```ruby
logger = Logger.new("log/kafka.log")
kafka = Kafka.new(logger: logger, ...)
```

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

### Tune Consumer for Throughput with max_bytes_per_partition

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Configure `max_bytes_per_partition` during subscription to fetch up to 5MB per partition, enhancing throughput. This setting affects how much data is retrieved in a single fetch request.

```ruby
# Fetches up to 5MB per partition at a time for better throughput.
consumer.subscribe("greetings", max_bytes_per_partition: 5 * 1024 * 1024)

consumer.each_message do |message|
  # ...
end
```

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

### List all topics

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Retrieves a list of all available topic names from the Kafka cluster.

```APIDOC
## List all topics

### Description
Return an array of topic names.

### Method
`kafka.topics`

### Parameters
None

### Response
#### Success Response (Array of Strings)
- `topics` (Array<String>) - A list of topic names.

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.topics
# => ["topic1", "topic2", "topic3"]
```
```

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

### Deliver Message Using Kafka Message Key

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Set a `key` to include a message key alongside the value. If no `partition_key` is specified, the message key is used for partitioning. Message keys can be used for features like Log Compaction.

```ruby
# Set a message key; the key will be used for partitioning since no explicit
# `partition_key` is set.
kafka.deliver_message("Hello, World!", key: "hello", topic: "greetings")
```

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

### Extract X509 Certificates and Key for Ruby Client

Source: https://github.com/zendesk/ruby-kafka/wiki/Creating-X509-certificates-from-JKS-format

These commands extract the signed client certificate, client private key, and CA certificate from the JKS keystore into PEM format, suitable for use with the ruby-kafka client.

```bash
### 7.1 Extract signed client certificate
keytool -noprompt -keystore kafka.client.keystore.jks -exportcert -alias localhost -rfc -storepass foobar -file client_cert.pem
### 7.2 Extract client key
keytool -noprompt -srckeystore kafka.client.keystore.jks -importkeystore -srcalias localhost -destkeystore cert_and_key.p12 -deststoretype PKCS12 -srcstorepass foobar -storepass foobar
openssl pkcs12 -in cert_and_key.p12 -nocerts -nodes -passin pass:foobar -out client_cert_key.pem
### 7.3 Extract CA certificate
keytool -noprompt -keystore kafka.client.keystore.jks -exportcert -alias CARoot -rfc -file ca_cert.pem -storepass foobar
```

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

### Configure Kafka Client with SSL CA Certificate

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use this to enable SSL encryption for communication with Kafka brokers. Requires a valid CA certificate file.

```ruby
kafka = Kafka.new(["kafka1:9092"], ssl_ca_cert: File.read('my_ca_cert.pem'))
```

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

### Implement Random Partition Assignment Strategy

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

A simple strategy that assigns partitions randomly among group members. The `call` method receives cluster, members, and partitions, and must return a hash mapping member IDs to their assigned partitions.

```ruby
class RandomAssignmentStrategy
  def call(cluster:, members:, partitions:)
    member_ids = members.keys
    partitions.each_with_object(Hash.new {|h, k| h[k] = [] }) do |partition, partitions_per_member|
      partitions_per_member[member_ids[rand(member_ids.count)]] << partition
    end
  end
end
```

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

### Configure Murmur2 Hashing for Partitioner

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Utilize the Murmur2 hash function for partitioning, which is compatible with Java-based Kafka producers. Pass `:murmur2` to the `hash_function` option.

```ruby
Kafka.new(partitioner: Kafka::Partitioner.new(hash_function: :murmur2))
```

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

### Asynchronously Produce Messages with Kafka

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Utilize the asynchronous producer (`#async_producer`) to avoid blocking during message delivery. Messages are sent in a background thread, and `#deliver_messages` returns immediately.

```ruby
# `#async_producer` will create a new asynchronous producer.
producer = kafka.async_producer

# The `#produce` API works as normal.
producer.produce("hello", topic: "greetings")

# `#deliver_messages` will return immediately.
producer.deliver_messages

# Make sure to call `#shutdown` on the producer in order to avoid leaking
# resources. `#shutdown` will wait for any pending messages to be delivered
# before returning.
producer.shutdown
```

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

### Deliver a Message to Kafka

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use `#deliver_message` for simple, infrequent writes. This method sends a single message to a random partition in the specified topic.

```ruby
kafka = Kafka.new(...)
kafka.deliver_message("Hello, World!", topic: "greetings")
```

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

### Select Specific Partition for Message

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Manually select a specific partition for a message using the modulo operator to ensure the partition number is within the valid range for the topic. This requires knowledge of the topic's partition count.

```ruby
partitions = kafka.partitions_for("events")

# Make sure that we don't exceed the partition count!
partition = some_number % partitions

producer.produce(event, topic: "events", partition: partition)
```

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

### Configuring Producer Retries and Backoff

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Improve delivery success chances by configuring `max_retries` and `retry_backoff` for the producer. These settings affect the maximum latency of message delivery operations.

```ruby
producer = kafka.producer(
  # The number of retries when attempting to deliver messages. The default is
  # 2, so 3 attempts in total, but you can configure a higher or lower number:
  max_retries: 5,

  # The number of seconds to wait between retries. In order to handle longer
  # periods of Kafka being unavailable, increase this number. The default is
  # 1 second.
  retry_backoff: 5,
)
```

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

### Configure Producer Buffer Size and Bytesize

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Adjust the maximum number of messages and total bytes allowed in the producer's buffer. This impacts resilience and throughput.

```ruby
producer = kafka.producer(
  max_buffer_size: 5_000,           # Allow at most 5K messages to be buffered.
  max_buffer_bytesize: 100_000_000, # Allow at most 100MB to be buffered.
  ...
)
```

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

### Alter Kafka Topic Configuration

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Updates the configurations for a given topic. Use with caution as this is an alpha feature for advanced usage.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.alter_topic("topic", "max.message.bytes" => 100000, "retention.ms" => 604800000)
```

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

### Deliver Message to Specific Kafka Partition

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

To send a message to a specific partition, use the `partition` parameter with `#deliver_message`. This requires knowing the exact partition number.

```ruby
# Will write to partition 42.
kafka.deliver_message("Hello, World!", topic: "greetings", partition: 42)
```

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

### Consume Messages from Kafka Topic (Simple)

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Consume messages from a single Kafka topic using a simple iterator. This method is suitable for basic use cases but has limitations regarding topic subscriptions and partition management.

```ruby
require "kafka"

kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])

kafka.each_message(topic: "greetings") do |message|
  puts message.offset, message.key, message.value
end
```

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

### Configure Kafka Client with OAUTHBEARER Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Authenticates using OAUTHBEARER, supported for Kafka versions 2.0.0 and later. Requires a `TokenProvider` object with a `token` method. For Kafka 2.1.0+, an optional `extensions` method can provide key-value pairs for the initial client response.

```ruby
class TokenProvider
  def token
    "some_id_token"
  end
end
```

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

### Configure Kafka Client with SCRAM Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Authenticates to Kafka using SCRAM (Secure Remote Password) mechanism, supporting SHA-256 and SHA-512. Requires username, password, and the mechanism type. SSL encryption is recommended.

```ruby
kafka = Kafka.new(
  ["kafka1:9092"],
  sasl_scram_username: 'username',
  sasl_scram_password: 'password',
  sasl_scram_mechanism: 'sha256',
  # ...
)
```

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

### Efficiently Produce Batched Messages with Kafka Producer

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use the `Kafka::Producer` API for efficient, batched message delivery. This avoids the overhead of single-message delivery and provides automatic retries and buffering.

```ruby
# Instantiate a new producer.
producer = kafka.producer

# Add a message to the producer buffer.
producer.produce("hello1", topic: "test-messages")

# Deliver the messages to Kafka.
producer.deliver_messages
```

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

### Consume Messages with Consumer Groups

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Utilize the Consumer API with consumer groups for coordinated message consumption across multiple processes. This approach handles partition assignment and failover automatically.

```ruby
require "kafka"

kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])

# Consumers with the same group id will form a Consumer Group together.
consumer = kafka.consumer(group_id: "my-consumer")

# It's possible to subscribe to multiple topics by calling `subscribe`
# repeatedly.
consumer.subscribe("greetings")

# Stop the consumer when the SIGTERM signal is sent to the process.
# It's better to shut down gracefully than to kill the process.
trap("TERM") { consumer.stop }

# This will loop indefinitely, yielding each message in turn.
consumer.each_message do |message|
  puts message.topic, message.partition
  puts message.offset, message.key, message.value
end
```

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

### Configure Kafka Client with AWS MSK IAM Authentication

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Authenticates to an AWS MSK cluster using IAM. Requires AWS access key ID, secret access key, and the AWS region. SSL encryption is enabled by default and `ssl_ca_certs_from_system: true` is used.

```ruby
k = Kafka.new(
  ["kafka1:9092"],
  sasl_aws_msk_iam_access_key_id: 'iam_access_key',
  sasl_aws_msk_iam_secret_key_id: 'iam_secret_key',
  sasl_aws_msk_iam_aws_region: 'us-west-2',
  ssl_ca_certs_from_system: true,
  # ...
)
```

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

### Reading Messages with AvroTurf

Source: https://github.com/zendesk/ruby-kafka/wiki/Encoding-messages-with-Avro

Decode a message received from Kafka using AvroTurf. Specifying `schema_name` provides schema compatibility guarantees against incompatible producer schema changes. Omitting it will decode data without these guarantees.

```ruby
require "avro_turf"
require "kafka"

# You need to pass the URL of your Schema Registry.
avro = AvroTurf::Messaging.new(registry_url: "http://my-registry:8081/")

kafka.each_message(topic: "greetings") do |message|
  # By passing in `schema_name:`, you guard against the producer changing
  # the schema in an incompatible way. You can leave out the argument,
  # in which case you'll just get whatever data the producer encoded out,
  # with no schema compatibility guarantees.
  greeting = avro.decode(message.value, schema_name: "greeting")
  greeting #=> { "title" => "hello, world" }
end
```

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

### Alter a topic configuration

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Updates the configuration values for a given topic. This is an alpha feature and intended for advanced usage.

```APIDOC
## Alter a topic configuration (alpha feature)

### Description
Updates the topic configurations. This feature is for advanced usage.

### Method
`kafka.alter_topic(topic_name, configurations)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topic_name** (String) - The name of the topic to alter.
- **configurations** (Hash) - A hash of configuration key-value pairs to update. Example: `{"max.message.bytes" => 100000, "retention.ms" => 604800000}`.

### Response
None (Success is indicated by lack of error).

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.alter_topic("topic", "max.message.bytes" => 100000, "retention.ms" => 604800000)
```
```

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

### Semantic Partitioning by Session ID

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Assign messages to the same partition based on a common property, such as a session ID. This ensures that all events within a user session are processed together, simplifying downstream consumer logic.

```ruby
# All messages with the same `session_id` will be assigned to the same partition.
producer.produce(event, topic: "user-events", partition_key: session_id)
```

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

### Require Notifications Library

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Ensure Active Support Notifications and the kafka library are required to subscribe to notifications.

```ruby
require "active_support/notifications"
require "kafka"
```

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

### Consume Kafka Messages in Batches

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Process messages in batches for integration with systems that support batch or transactional APIs. The consumer commits offsets only after the entire batch is processed successfully.

```ruby
# A mock search index that we'll be keeping up to date with new Kafka messages.
index = SearchIndex.new

consumer.subscribe("posts")

consumer.each_batch do |batch|
  puts "Received batch: #{batch.topic}/#{batch.partition}"

  transaction = index.transaction

  batch.messages.each do |message|
    # Let's assume that adding a document is idempotent.
    transaction.add(id: message.key, body: message.value)
  end

  # Once this method returns, the messages have been successfully written to the
  # search index. The consumer will only checkpoint a batch *after* the block
  # has completed without an exception.
  transaction.commit!
end
```

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

### Subscribe to All Kafka Notifications

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Subscribe to all notifications sent by the ruby-kafka library using a regular expression. This allows for centralized logging or monitoring of Kafka events.

```ruby
ActiveSupport::Notifications.subscribe(/.*\.kafka$/) do |*args|
  event = ActiveSupport::Notifications::Event.new(*args)
  puts "Received notification `#{event.name}` with payload: #{event.payload.inspect}"
end
```

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

### Manual Offset Commit

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Trigger an immediate, blocking offset commit to ensure progress is saved.

```APIDOC
## Manual Offset Commit

### Description
Trigger an immediate, blocking offset commit to ensure progress is saved. This method blocks until the Kafka cluster acknowledges the commit.

### Method
`Consumer#commit_offsets`

### Endpoint
N/A (Method call)

### Parameters
None

### Request Example
```ruby
consumer.commit_offsets
```
```

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

### Produce Messages in a Controller

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use the configured asynchronous producer within a controller to send event data to a Kafka topic. The event data is serialized to JSON before production.

```ruby
class OrdersController
  def create
    @order = Order.create!(params[:order])

    event = {
      order_id: @order.id,
      amount: @order.amount,
      timestamp: Time.now,
    }

    $kafka_producer.produce(event.to_json, topic: "order_events")
  end
end
```

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

### Gracefully Shut Down Kafka Consumer

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Trap the TERM signal to initiate a clean shutdown of the Kafka consumer. This ensures that any ongoing message processing is handled before the consumer stops.

```ruby
consumer = kafka.consumer(...)

# The consumer can be stopped from the command line by executing
# `kill -s TERM <process-id>`.
trap("TERM") { consumer.stop }

consumer.each_message do |message|
  ...
end
```

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

### Set Message Acknowledgment Level (Required Acks)

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Control message durability by specifying how many replicas must acknowledge a write. Options range from fire-and-forget to waiting for all replicas.

```ruby
# This is the default: all replicas must acknowledge.
producer = kafka.producer(required_acks: :all)

# This is fire-and-forget: messages can easily be lost.
producer = kafka.producer(required_acks: 0)

# This only waits for the leader to acknowledge.
producer = kafka.producer(required_acks: 1)
```

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

### Synchronous Producer Delivery

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Use the synchronous producer for at-least-once delivery guarantees. Ensure messages are delivered to Kafka after `deliver_messages` returns, but be aware of potential Kafka cluster configurations and acknowledgement settings.

```ruby
producer = kafka.producer

producer.produce("hello", topic: "greetings")

# If this line fails with Kafka::DeliveryFailed we *may* have succeeded in delivering
# the message to Kafka but won't know for sure.
producer.deliver_messages

# If we get to this line we can be sure that the message has been delivered to Kafka!
```

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

### Mark Message as Processed

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Manually mark a message and all preceding messages in a partition as processed. Use with caution.

```APIDOC
## Mark Message as Processed

### Description
Manually mark a message and all preceding messages in a partition as processed. This is an advanced API and should only be used when you understand the implications, typically for transactional grouping of messages.

### Method
`Consumer#mark_message_as_processed`

### Endpoint
N/A (Method call)

### Parameters
#### Path Parameters
- **message** (Message) - Required - The message object to mark as processed.

### Request Example
```ruby
consumer.mark_message_as_processed(message)
```
```

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

### Manually Control Message Processing Checkpointing

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Manually mark messages as processed to control checkpointing, especially when buffering messages for transactional processing. Ensure messages are marked to avoid reprocessing upon consumer restart.

```ruby
# Manually controlling checkpointing:

# Typically you want to use this API in order to buffer messages until some
# special "commit" message is received, e.g. in order to group together
# transactions consisting of several items.
buffer = []

# Messages will not be marked as processed automatically. If you shut down the
# consumer without calling `#mark_message_as_processed` first, the consumer will
# not resume where you left off!
consumer.each_message(automatically_mark_as_processed: false) do |message|
  # Our messages are JSON with a `type` field and other stuff.
  event = JSON.parse(message.value)

  case event.fetch("type")
  when "add_to_cart"
    buffer << event
  when "complete_purchase"
    # We've received all the messages we need, time to save the transaction.
    save_transaction(buffer)

    # Now we can set the checkpoint by marking the last message as processed.
    consumer.mark_message_as_processed(message)

    # We can optionally trigger an immediate, blocking offset commit in order
    # to minimize the risk of crashing before the automatic triggers have
    # kicked in.
    consumer.commit_offsets

    # Make the buffer ready for the next transaction.
    buffer.clear
  end
end
```

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

### Delete Kafka Topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Marks a Kafka topic for deletion. The topic is hidden from clients and will be completely removed after a delay.

```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.delete_topic("topic")
```

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

### Delete a topic

Source: https://github.com/zendesk/ruby-kafka/blob/main/README.md

Marks a topic for deletion. Note that Kafka hides topics from clients upon deletion and complete removal may take time.

```APIDOC
## Delete a topic

### Description
Marks a topic for deletion. After a topic is marked as deleted, Kafka only hides it from clients. It would take a while before a topic is completely deleted.

### Method
`kafka.delete_topic(topic_name)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topic_name** (String) - The name of the topic to delete.

### Response
None (Success is indicated by lack of error).

### Request Example
```ruby
kafka = Kafka.new(["kafka:9092"])
kafka.delete_topic("topic")
```
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.