### Full Configuration Example for grpc-kafka

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

A comprehensive JSON5 configuration file example for the grpc-kafka tool, covering settings for all three modes: grpc2kafka, dedup, and kafka2grpc. Supports human-friendly numeric strings.

```json5
// config-kafka.json — full example covering all three modes
{
    // Optional: Prometheus metrics server address
    "prometheus": "127.0.0.1:8873",

    // Global librdkafka settings shared across all modes
    "kafka": {
        "bootstrap.servers": "localhost:29092",
        "statistics.interval.ms": "1000"  // enables per-broker Prometheus metrics
    },

    // --- grpc2kafka mode ---
    "grpc2kafka": {
        "endpoint": "http://127.0.0.1:10000",  // Yellowstone gRPC endpoint
        "x_token": null,                         // optional auth token
        "kafka_topic": "grpc1",
        "kafka_queue_size": "10_000",            // max in-flight Kafka sends
        // Per-mode kafka overrides (merged with top-level "kafka" block)
        "kafka": {},
        // Geyser subscription filter
        "request": {
            "slots": {
                "client": { "filter_by_commitment": null }
            },
            "blocks": {
                "client": {
                    "account_include": [],
                    "include_transactions": false,
                    "include_accounts": false,
                    "include_entries": false
                }
            }
        }
    },

    // --- dedup mode ---
    "dedup": {
        "kafka": {
            "group.id": "dedup",
            "group.instance.id": "dedup"
        },
        "kafka_input": "grpc1",       // topic to consume from
        "kafka_output": "grpc2",      // topic to produce deduplicated messages to
        "kafka_queue_size": "10_000",
        "backend": { "type": "memory" }  // only "memory" supported currently
    },

    // --- kafka2grpc mode ---
    "kafka2grpc": {
        "kafka": {
            "group.id": "kafka2grpc",
            "group.instance.id": "kafka2grpc"
        },
        "kafka_topic": "grpc2",
        "listen": "127.0.0.1:10001",   // gRPC server listen address
        "channel_capacity": 250000      // broadcast channel buffer size per client
    }
}

```

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

### Development Setup for Yellowstone gRPC Kafka Tool

Source: https://github.com/rpcpool/yellowstone-grpc-kafka/blob/master/README.md

These commands outline the steps for setting up a local Kafka environment and interacting with the tool. Ensure Kafka is running before executing other commands.

```bash
# run kafka locally
docker-compose -f docker-kafka.yml up
```

```bash
# create topic
kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server localhost:29092 --create --topic grpc1
```

```bash
# send messages from gRPC to Kafka
cargo run --bin grpc-kafka -- --config config-kafka.json grpc2kafka
```

```bash
# read messages from Kafka
kafka_2.13-3.5.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:29092 --topic grpc1
```

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

### Local Development Setup for Yellowstone gRPC Kafka

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Steps to set up Kafka, create topics, run the pipeline in different modes (ingest, deduplicate, re-expose), and execute tests.

```bash
# 1. Start Kafka (KRaft mode, no ZooKeeper)
docker-compose -f docker-kafka.yml up -d
```

```bash
# 2. Create topics
kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server localhost:29092 --create --topic grpc1
kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server localhost:29092 --create --topic grpc2
```

```bash
# 3. Run the full pipeline (three terminals):
# Terminal 1 — ingest from gRPC
car go run --bin grpc-kafka -- --config config-kafka.json grpc2kafka

# Terminal 2 — deduplicate
car go run --bin grpc-kafka -- --config config-kafka.json dedup

# Terminal 3 — re-expose as gRPC
car go run --bin grpc-kafka -- --config config-kafka.json kafka2grpc
```

```bash
# 4. Run tests
car go test
```

```bash
# 5. Check for dependency vulnerabilities/license issues
car go deny check
```

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

### Start Kafka Locally

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Initializes a local Kafka environment using Docker Compose for development purposes. Ensure this is running before proceeding with Kafka-related operations.

```bash
# Start Kafka locally for development
docker-compose -f docker-kafka.yml up -d
```

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

### Instrument Kafka Clients with StatsContext

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Example of creating instrumented Kafka producer and consumer using StatsContext for Prometheus metrics and error signaling. Ensure librdkafka is configured for statistics.

```rust
// Creating instrumented producer and consumer (from src/kafka/metrics.rs)
use yellowstone_grpc_kafka::kafka::metrics::StatsContext;
use rdkafka::config::ClientConfig;

let mut kafka_config = ClientConfig::new();
kafka_config.set("bootstrap.servers", "localhost:29092");
kafka_config.set("statistics.interval.ms", "1000");

// Producer with error signalling
let (producer, error_rx) = StatsContext::create_future_producer(&kafka_config)
    .expect("failed to create Kafka producer");

// Consumer with error signalling
kafka_config.set("group.id", "my-consumer-group");
let (consumer, error_rx) = StatsContext::create_stream_consumer(&kafka_config)
    .expect("failed to create Kafka consumer");

// In your event loop, watch error_rx to detect fatal Kafka errors:
// tokio::select! {
//     _ = error_rx => { eprintln!("Fatal Kafka error — exiting"); break; }
//     message = consumer.recv() => { /* process message */ }
// }

```

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

### Build and View Help for grpc-kafka

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Build the release version of the tool and display its command-line help information to understand available options and commands.

```bash
cargo build --release
./target/release/grpc-kafka --help
```

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

### Display Help for Yellowstone gRPC Kafka Tool

Source: https://github.com/rpcpool/yellowstone-grpc-kafka/blob/master/README.md

Use this command to view all available options and subcommands for the tool. It helps in understanding the tool's capabilities and usage.

```bash
$ cargo run --bin grpc-kafka -- --help
```

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

### Run kafka2grpc Service and Subscribe to Stream

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Instructions for running the kafka2grpc service and subscribing to its message stream using grpcurl. Ensure the configuration file is correctly set up.

```bash
# Run kafka2grpc: reads from "grpc2", exposes gRPC on port 10001
cargo run --bin grpc-kafka -- --config config-kafka.json kafka2grpc

# Connect to the kafka2grpc server using grpcurl
grpcurl -plaintext localhost:10001 list
# Output: geyser.Geyser, grpc.health.v1.Health

# Subscribe to the stream (receives all messages forwarded from Kafka)
grpcurl -plaintext \
    -d '{"slots": {"client": {}}, "commitment": 1}' \
    localhost:10001 \
    geyser.Geyser/Subscribe

# Health check
grpcurl -plaintext localhost:10001 grpc.health.v1.Health/Check
# Output: { "status": "SERVING" }

# Get version info
grpcurl -plaintext -d '{}' localhost:10001 geyser.Geyser/GetVersion
# Output: { "version": "{\"package\":\"yellowstone-grpc-kafka\",\"version\":\"4.2.1\",...}" }

```

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

### Run grpc2kafka Mode

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Launches the grpc-kafka binary in 'grpc2kafka' mode. This connects to a Yellowstone gRPC endpoint, subscribes using a configuration file, and streams updates to a Kafka topic. Messages are keyed for deduplication.

```bash
# Run grpc2kafka — connects to gRPC and writes to topic "grpc1"
cargo run --bin grpc-kafka -- --config config-kafka.json grpc2kafka
```

```bash
# With Prometheus metrics enabled
cargo run --bin grpc-kafka -- \
    --config config-kafka.json \
    --prometheus 127.0.0.1:9090 \
    grpc2kafka
```

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

### Create Kafka Topic

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Creates a Kafka topic with specified name, partitions, and replication factor. This is a prerequisite for both sending and consuming messages.

```bash
# Create the input topic
kafka_2.13-3.5.0/bin/kafka-topics.sh \
    --bootstrap-server localhost:29092 \
    --create --topic grpc1 \
    --partitions 1 --replication-factor 1
```

```bash
# Create the output (deduplicated) topic
kafka_2.13-3.5.0/bin/kafka-topics.sh \
    --bootstrap-server localhost:29092 \
    --create --topic grpc2 \
    --partitions 1 --replication-factor 1
```

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

### Run grpc-kafka in grpc2kafka Mode

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Execute the grpc-kafka tool in 'grpc2kafka' mode, which subscribes to a Yellowstone gRPC endpoint and forwards data to Kafka. Requires a configuration file.

```bash
./target/release/grpc-kafka --config config-kafka.json grpc2kafka
```

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

### Run grpc-kafka in kafka2grpc Mode

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Execute the grpc-kafka tool in 'kafka2grpc' mode, which reads messages from a Kafka topic and re-exposes them as a gRPC server. Requires a configuration file.

```bash
./target/release/grpc-kafka --config config-kafka.json kafka2grpc
```

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

### Access Prometheus Metrics

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

How to enable and scrape Prometheus metrics from the yellowstone-grpc-kafka service. Metrics cover throughput, deduplication, and Kafka broker statistics.

```bash
# Enable via config
# "prometheus": "127.0.0.1:8873"
# Or override with CLI: --prometheus 0.0.0.0:9090

# Scrape all metrics
curl http://127.0.0.1:8873/metrics

# Key metrics:
# version{buildts,git,package,proto,rustc,solana,version}  — build info (always 1)
# kafka_recv_total          — messages consumed from Kafka input topic
# kafka_sent_total{kind}    — messages produced, labelled by update type
#                             kinds: account, slot, transaction, transactionstatus,
#                                    block, blockmeta, entry, ping, pong, unknown
# kafka_dedup_total         — messages dropped by dedup as duplicates
# kafka_stats{broker,metric} — librdkafka per-broker gauges, including:
#   outbuf_cnt, outbuf_msg_cnt, waitresp_cnt, tx, txerrs, txretries, req_timeouts
#   int_latency.{min,max,avg,p50,p75,p90,p95,p99,p99_99}
#   outbuf_latency.{min,max,avg,p50,p75,p90,p95,p99,p99_99}

# Example Prometheus scrape config (prometheus.yml)
# scrape_configs:
#   - job_name: 'yellowstone-grpc-kafka'
#     static_configs:
#       - targets: ['127.0.0.1:8873']

```

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

### Run dedup Mode

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Launches the grpc-kafka binary in 'dedup' mode. This consumes messages from an input Kafka topic, filters out duplicates using an in-memory store with a sliding window, and forwards unique messages to an output topic.

```bash
# Run dedup: reads from "grpc1", writes unique messages to "grpc2"
cargo run --bin grpc-kafka -- --config config-kafka.json dedup
```

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

### Run grpc-kafka in dedup Mode

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Execute the grpc-kafka tool in 'dedup' mode, which consumes messages from a Kafka topic, deduplicates them, and sends unique messages to another topic. Requires a configuration file.

```bash
./target/release/grpc-kafka --config config-kafka.json dedup
```

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

### Initialize Tracing and Shutdown Handler in Rust

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Initializes tracing with INFO level and respects RUST_LOG. Creates a future that resolves on SIGINT or SIGTERM for graceful shutdown.

```rust
use yellowstone_grpc_kafka::{setup_tracing, create_shutdown};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Initializes tracing-subscriber with INFO default level.
    // Respects RUST_LOG env var, e.g.: RUST_LOG=yellowstone_grpc_kafka=debug
    // Automatically enables ANSI colors when stdout/stderr are a terminal.
    setup_tracing()?;

    // Returns a future that resolves on SIGINT or SIGTERM.
    // Pass to tokio::select! to trigger clean shutdown.
    let shutdown = create_shutdown()?;

    tokio::select! {
        _ = shutdown => {
            tracing::warn!("shutdown signal received");
        }
        // _ = your_main_task => {}
    }
    Ok(())
}
```

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

### Verify Kafka Messages

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Consumes messages from a Kafka topic to verify data flow. This command retrieves a limited number of messages from the beginning of the topic.

```bash
# Verify messages arriving in Kafka (raw binary, but confirms flow)
kafka_2.13-3.5.0/bin/kafka-console-consumer.sh \
    --bootstrap-server localhost:29092 \
    --topic grpc1 \
    --from-beginning \
    --max-messages 5
```

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

### Override Prometheus Address via CLI Flag

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Run the grpc-kafka tool in 'grpc2kafka' mode while overriding the Prometheus metrics server address specified in the configuration file using a command-line flag.

```bash
./target/release/grpc-kafka --config config-kafka.json --prometheus 0.0.0.0:9090 grpc2kafka
```

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

### Check Prometheus Metrics for grpc2kafka

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Retrieves metrics exposed by the grpc-kafka service when Prometheus is enabled. Useful for monitoring the health and throughput of the data streaming process.

```bash
# Check Prometheus metrics
curl http://127.0.0.1:9090/metrics
# Example output:
# kafka_sent_total{kind="slot"} 42300
# kafka_sent_total{kind="transaction"} 8917
# kafka_sent_total{kind="account"} 1203
# kafka_stats{broker="localhost:29092/1",metric="tx"} 52420
```

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

### Configure Yellowstone gRPC SubscribeRequest Filter

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Defines a comprehensive filter for Yellowstone's gRPC SubscribeRequest. Supports filtering by slots, accounts (with various criteria like owner, memcmp, data size, lamports), transactions, blocks, and block metadata. Use this to specify exactly which data streams you need.

```json
{
    "request": {
        "slots": {
            "all": { "filter_by_commitment": true }
        },
        "accounts": {
            "my_account": {
                "account": ["So11111111111111111111111111111111111111112"],
                "owner": [],
                "filters": []
            },
            "token_accounts": {
                "account": [],
                "owner": ["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"],
                "filters": [{ "TokenAccountState": true }],
                "nonempty_txn_signature": true
            },
            "filtered": {
                "account": [],
                "owner": ["MyProgramId111111111111111111111111111111111"],
                "filters": [
                    { "Memcmp": { "offset": 0, "base58": "3Mc6vR" } },
                    { "DataSize": 165 },
                    { "Lamports": { "Gt": 1000000 } }
                ]
            }
        },
        "transactions": {
            "my_txns": {
                "vote": false,
                "failed": false,
                "account_include": ["MyProgramId111111111111111111111111111111111"],
                "account_exclude": [],
                "account_required": []
            }
        },
        "blocks": {
            "all_blocks": {
                "account_include": [],
                "include_transactions": true,
                "include_accounts": false,
                "include_entries": false
            }
        },
        "blocks_meta": ["all"],
        "commitment": "confirmed",
        "accounts_data_slice": [
            { "offset": 0, "length": 32 }
        ],
        "from_slot": 300000000
    }
}
```

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

### Monitor dedup Metrics via Prometheus

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Fetches deduplication metrics from the running dedup service. This helps in monitoring the number of messages processed, duplicates detected, and overall efficiency.

```bash
# Monitor dedup metrics via Prometheus
curl http://127.0.0.1:8873/metrics | grep kafka_dedup
# Example output:
# kafka_recv_total 154823       <- total messages consumed from grpc1
# kafka_dedup_total 3241        <- messages dropped as duplicates
```

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

### Kafka Deduplication Trait Interface

Source: https://context7.com/rpcpool/yellowstone-grpc-kafka/llms.txt

Defines the core deduplication logic for Kafka messages. Use this trait to check if a (slot, hash) pair is new and should be processed.

```rust
// KafkaDedupMemory — core deduplication logic (from src/kafka/dedup.rs)
// The trait interface:
#[async_trait::async_trait]
pub trait KafkaDedup: Clone {
    // Returns true if the (slot, hash) pair is new and should be forwarded.
    // Returns false if it has already been seen (duplicate) or is too old (slot < oldest_tracked_slot).
    async fn allowed(&self, slot: u64, hash: [u8; 32]) -> bool;
}

// Usage pattern in the dedup loop:
// if dedup.allowed(slot, hash_bytes).await {
//     kafka_producer.send(record_to_output_topic).await?;
// }
// // Slots older than (current_slot - 75) are evicted automatically

```

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