Kafka Streaming

Kafka Advantages

Lower Latency

Shorter data pipeline
Less processing overhead compared to GraphQL subscriptions
No custom database or additional formatting logic needed

Better Reliability

More stable connection protocol than WebSocket
Optimized for persistent connections
Messages can be read from the latest offset

Superior Scalability

Multiple consumers can split the load
Automatic load redistribution
Better handling of high-volume data

Kafka Limitations

Browser Incompatibility

Server-side only implementation
No direct browser consumption

Limited Filtering

No pre-filtering capabilities
Pre-defined schema
Client-side post-processing required

Development Complexity

No IDE support for streams
Debugging limited to the consumer side

Technical Considerations

When using Kafka Streams, consider message retention policies, latency implications, and partitioning strategies. Messages have a default retention of 24 hours for critical data streams.

Implementation Guide

Message Retention

Retention Policies

Protobuf Streams: Retained for 24 hours
DEX Trades (JSON): Retained for 24 hours
Other JSON Streams: Retained for 4 hours

Connection Setup

const sasl_conf = {
  'bootstrap.servers': 'kfk0.chainstream.io:9093,kfk1.chainstream.io:9093,kfk2.chainstream.io:9093',
  'security.protocol': 'SASL_SSL',
  'sasl.mechanism': 'SCRAM-SHA-512',
  'sasl.username': '<YOUR USERNAME>',
  'sasl.password': '<YOUR PASSWORD>',
  'ssl.key.location': 'client.key.pem',
  'ssl.ca.location': 'server.cer.pem',
  'ssl.certificate.location': 'client.cer.pem',
  'ssl.endpoint.identification.algorithm': 'none'
}

Topic Structure

<BLOCKCHAIN_NAME>.<MESSAGE_TYPE>
<BLOCKCHAIN_NAME>.broadcasted.<MESSAGE_TYPE>

Message Types:

dextrades - DEX trades
dexorders - DEX orders
dexpools - DEX pools
transactions - Transactions
transfers - Transfers
instructions - Instructions
raw - Raw data

Best Practices for Kafka Streaming

Parallel Processing for High Throughput

Distribute consumers across multiple partitions to process data in parallel
Allocate one consumer thread per partition for efficient load balancing

Continuous Streaming for Stability

Keep Kafka consumers running to ensure seamless data flow
Process messages asynchronously to prevent bottlenecks
Avoid blocking the main consumption loop to maintain optimal throughput

Optimized Message Handling

Implement batch processing where applicable to reduce overhead
Use worker groups to improve concurrency
Continuously monitor processing latency to ensure real-time performance

Additional Code Examples

Prerequisites

ChainStream Kafka Server Access

Access to ChainStream Kafka brokers

Authentication Credentials

Username and Password for Kafka brokers

Topic Subscription

Topic name(s) to subscribe to

Go Installation

Go Version >= 1.16

Kafka Client

Confluent Kafka Go Client library

Dependencies Setup

go mod init kafka-consumer

Implementation

config := &kafka.ConfigMap{
    "bootstrap.servers":                     "rpk0.chainstream.io:9093,rpk1.chainstream.io:9093,rpk2.chainstream.io:9093",
    "group.id":                              "solanatest2-group-99",
    "session.timeout.ms":                    30000,
    "security.protocol":                     "SASL_SSL",
    "ssl.ca.location":                       "server.cer.pem",
    "ssl.key.location":                      "client.key.pem",
    "ssl.certificate.location":              "client.cer.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "sasl.mechanisms":                       "SCRAM-SHA-512",
    "sasl.username":                         "usernamee",
    "sasl.password":                         "pwww",
    "auto.offset.reset":                     "latest",
}

consumer, err := kafka.NewConsumer(config)
if err != nil {
    fmt.Printf("Failed to create consumer: %s\n", err)
    os.Exit(1)
}

topic := "solana.broadcasted.transactions"

// Subscribe to the topic
err = consumer.SubscribeTopics([]string{topic}, nil)
if err != nil {
    fmt.Printf("Failed to subscribe to topic: %s\n", err)
    os.Exit(1)
}

// Set up a channel to handle shutdown
sigchan := make(chan os.Signal, 1)
signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

// Poll messages and process them
run := true
for run {
    select {
    case sig := <-sigchan:
        fmt.Printf("Caught signal %v: terminating\n", sig)
        run = false
    default:
        ev := consumer.Poll(100)
        if ev == nil {
            continue
        }

        switch e := ev.(type) {
        case *kafka.Message:
            processMessage(e)

        case kafka.Error:
            fmt.Printf("Error: %v\n", e)
            if e.Code() == kafka.ErrAllBrokersDown {
                run = false
            }
        default:
            fmt.Printf("Ignored %v\n", e)
        }
    }
}

// Close down consumer
consumer.Close()

func processMessage(msg *kafka.Message) {
    fmt.Printf("Received message on topic %s [%d] at offset %v:\n",
        *msg.TopicPartition.Topic, msg.TopicPartition.Partition, msg.TopicPartition.Offset)

    // Try to parse the message value as JSON
    var jsonData interface{}
    err := json.Unmarshal(msg.Value, &jsonData)
    if err != nil {
        fmt.Printf("Error parsing JSON: %v\n", err)
        fmt.Printf("Raw message: %s\n", string(msg.Value))
        return
    }

    // Pretty print the JSON
    prettyJSON, err := json.MarshalIndent(jsonData, "", "  ")
    if err != nil {
        fmt.Printf("Error prettifying JSON: %v\n", err)
        return
    }

    fmt.Printf("Parsed JSON:\n%s\n", string(prettyJSON))

    // Log message data
    logEntry := map[string]interface{}{
        "topic":     *msg.TopicPartition.Topic,
        "partition": msg.TopicPartition.Partition,
        "offset":    msg.TopicPartition.Offset,
        "key":       string(msg.Key),
        "value":     string(prettyJSON),
    }
    fmt.Printf("Log entry: %+v\n", logEntry)
    fmt.Println("----------------------------------------")
}

Running the Application

go mod init kafka-consumer

Execution Workflow

Kafka Client Initialization

Initialize with SSL and SASL configurations
Set unique group.id for independent message consumption

Consumer Setup

Create Kafka consumer
Subscribe to specified topic

Message Processing

Poll for new messages every 100ms
Parse JSON messages
Log processed data

Graceful Shutdown

Handle termination signals
Close consumer cleanly

Get Started

ChainStream Products

Solutions

ChainStream Data Services (CDS)

Documentation

Legal & Policies

Kafka Advantages

Lower Latency

Better Reliability

Superior Scalability

Kafka Limitations

Browser Incompatibility

Limited Filtering

Development Complexity

Technical Considerations

Implementation Guide

Message Retention

Retention Policies

Connection Setup

Topic Structure

Best Practices for Kafka Streaming

Parallel Processing for High Throughput

Continuous Streaming for Stability

Optimized Message Handling

Additional Code Examples

Prerequisites

ChainStream Kafka Server Access

Authentication Credentials

Topic Subscription

Go Installation

Kafka Client

Dependencies Setup

Implementation

Running the Application

Execution Workflow

Kafka Client Initialization

Consumer Setup

Message Processing

Graceful Shutdown

Get Started

ChainStream Products

Solutions

ChainStream Data Services (CDS)

Documentation

Legal & Policies

​Kafka Advantages

Lower Latency

Better Reliability

Superior Scalability

​Kafka Limitations

Browser Incompatibility

Limited Filtering

Development Complexity

​Technical Considerations

​Implementation Guide

​Message Retention

Retention Policies

​Connection Setup

​Topic Structure

​Best Practices for Kafka Streaming

Parallel Processing for High Throughput

Continuous Streaming for Stability

Optimized Message Handling

​Additional Code Examples

​Prerequisites

ChainStream Kafka Server Access

Authentication Credentials

Topic Subscription

Go Installation

Kafka Client

​Dependencies Setup

​Implementation

​Running the Application

​Execution Workflow

Kafka Client Initialization

Consumer Setup

Message Processing

Graceful Shutdown

Kafka Advantages

Kafka Limitations

Technical Considerations

Implementation Guide

Message Retention

Connection Setup

Topic Structure

Best Practices for Kafka Streaming

Additional Code Examples

Prerequisites

Dependencies Setup

Implementation

Running the Application

Execution Workflow