Streamline Go SDK

Official Go client for Streamline - The Redis of Streaming.

Features

• Idiomatic Go with context support
• Producer with batching and compression
• Consumer with group coordination
• Admin client for topic management
• Query client for SQL analytics
• SASL/SCRAM authentication support
• TLS support

Installation

go get github.com/streamlinelabs/streamline-go-sdk/streamline

OpenTelemetry Tracing

The SDK includes built-in OpenTelemetry tracing via TracingProducer and TracingConsumer wrappers. These wrappers automatically create spans for produce and consume operations and propagate trace context through message headers.

Setup

import "github.com/streamlinelabs/streamline-go-sdk/streamline"

// Wrap an existing producer with tracing
tracingProducer := streamline.NewTracingProducer(client.Producer)

// Wrap an existing consumer with tracing
tracingConsumer := streamline.NewTracingConsumer(consumer)

Producing with Tracing

// Sends a message with automatic span creation and context injection
result, err := tracingProducer.Send(ctx, "orders", []byte("key"), []byte("value"))

Consuming with Tracing

messages, errors := tracingConsumer.Start(ctx)
for msg := range messages {
    // Create a processing span linked to the producer trace
    processCtx, span := tracingConsumer.TraceProcess(ctx, msg)
    processMessage(processCtx, msg)
    span.End()
}

Span Conventions

Attribute	Value
Span name	{topic} {operation} (e.g., "orders produce")
messaging.system	streamline
messaging.destination.name	Topic name
messaging.operation	produce, consume, or process
Span kind	PRODUCER for produce, CONSUMER for consume

Trace context is propagated via W3C TraceContext headers in messages.

Quick Start

package main

import (
    "context"
    "log"

    "github.com/streamlinelabs/streamline-go-sdk/streamline"
)

func main() {
    // Create client
    config := streamline.DefaultConfig()
    config.Brokers = []string{"localhost:9092"}

    client, err := streamline.NewClient(config)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    ctx := context.Background()

    // Produce a message
    result, err := client.Producer.Send(ctx, "my-topic", nil, []byte("Hello, World!"))
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Produced to partition %d at offset %d", result.Partition, result.Offset)
}

Producer

Simple Send

result, err := client.Producer.Send(ctx, "topic", []byte("key"), []byte("value"))

Send with Headers

result, err := client.Producer.SendMessage(ctx, &streamline.Message{
    Topic: "topic",
    Key:   []byte("key"),
    Value: []byte("value"),
    Headers: map[string][]byte{
        "trace-id": []byte("abc123"),
    },
})

Batch Send

messages := []*streamline.Message{
    {Topic: "topic", Value: []byte("msg1")},
    {Topic: "topic", Value: []byte("msg2")},
    {Topic: "topic", Value: []byte("msg3")},
}
results, err := client.Producer.SendBatch(ctx, messages)

Async Send

resultCh := client.Producer.SendAsync(&streamline.Message{
    Topic: "topic",
    Value: []byte("async message"),
})

result := <-resultCh
if result.Err != nil {
    log.Printf("Error: %v", result.Err)
} else {
    log.Printf("Sent to partition %d", result.Partition)
}

Transactions

The Go SDK supports client-side buffered transactions for atomic batch sends:

producer := client.Producer()
producer.BeginTransaction()

producer.SendTransactional(ctx, &streamline.Message{Topic: "orders", Key: []byte("k1"), Value: []byte("v1")})
producer.SendTransactional(ctx, &streamline.Message{Topic: "orders", Key: []byte("k2"), Value: []byte("v2")})

results, err := producer.CommitTransaction(ctx)
if err != nil {
    producer.AbortTransaction()
}

Note: Transactions use client-side buffering — messages are collected locally and sent as a batch on commit. This provides all-or-nothing delivery semantics at the client level. If you need broker-level exactly-once semantics, combine with idempotent producers.

Consumer

Consume Messages

consumer, err := client.NewConsumer(ctx, "my-group", []string{"my-topic"})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

messages, errors := consumer.Start(ctx)

for {
    select {
    case msg := <-messages:
        log.Printf("Received: %s", string(msg.Value))
    case err := <-errors:
        log.Printf("Error: %v", err)
    case <-ctx.Done():
        return
    }
}

Poll Messages

messages, err := consumer.Poll(ctx, 100, 5*time.Second)
for _, msg := range messages {
    log.Printf("Received: %s", string(msg.Value))
}

Admin Client

Create Topic

err := client.Admin.CreateTopic(ctx, streamline.TopicConfig{
    Name:              "my-topic",
    NumPartitions:     3,
    ReplicationFactor: 1,
    Config: map[string]string{
        "retention.ms": "86400000",
    },
})

List Topics

topics, err := client.Admin.ListTopics(ctx)
for _, t := range topics {
    log.Printf("Topic: %s, Partitions: %d", t.Name, t.Partitions)
}

Describe Topic

info, partitions, err := client.Admin.DescribeTopic(ctx, "my-topic")
log.Printf("Topic: %s", info.Name)
for _, p := range partitions {
    log.Printf("  Partition %d: leader=%d, replicas=%v", p.ID, p.Leader, p.Replicas)
}

Consumer Group Management

// List consumer groups
groups, err := client.Admin.ListConsumerGroups(ctx)

// Describe consumer group
info, err := client.Admin.DescribeConsumerGroup(ctx, "my-group")

// Reset offsets to earliest
err := client.Admin.ResetConsumerGroupOffsets(ctx, "my-group", "my-topic", -2)

// Reset offsets to latest
err := client.Admin.ResetConsumerGroupOffsets(ctx, "my-group", "my-topic", -1)

HTTP Admin (Cluster, Lag, Inspect, Metrics)

The HTTPAdmin client communicates with the Streamline HTTP REST API for operations not available via the Kafka wire protocol.

admin := streamline.NewHTTPAdmin("http://localhost:9094")

// Cluster overview
cluster, err := admin.ClusterInfo(ctx)
fmt.Printf("Cluster: %s, Brokers: %d\n", cluster.ClusterID, len(cluster.Brokers))

// Consumer group lag monitoring
lag, err := admin.ConsumerGroupLag(ctx, "my-group")
fmt.Printf("Total lag: %d\n", lag.TotalLag)
for _, p := range lag.Partitions {
    fmt.Printf("  %s:%d lag=%d\n", p.Topic, p.Partition, p.Lag)
}

// Message inspection
messages, err := admin.InspectMessages(ctx, "events", 0, nil, 10)
for _, m := range messages {
    fmt.Printf("offset=%d key=%v value=%s\n", m.Offset, m.Key, m.Value)
}

// Latest messages
latest, err := admin.LatestMessages(ctx, "events", 5)

// Server metrics
metrics, err := admin.MetricsHistory(ctx)
for _, m := range metrics {
    fmt.Printf("%s=%f %v\n", m.Name, m.Value, m.Labels)
}

Query Client

Execute a Query

queryClient := streamline.NewQueryClient("http://localhost:9094")

result, err := queryClient.Query(ctx, "SELECT * FROM topic('events') LIMIT 10")
if err != nil {
    log.Fatal(err)
}

for _, row := range result.Rows {
    fmt.Println(row)
}
fmt.Printf("Scanned %d rows in %dms\n", result.Metadata.RowsScanned, result.Metadata.ExecutionTimeMs)

Query with Options

opts := streamline.QueryOptions{
    TimeoutMs: 5000,
    MaxRows:   100,
}
result, err := queryClient.QueryWithOptions(ctx, "SELECT * FROM topic('events') ORDER BY offset DESC", opts)

Explain Query Plan

plan, err := queryClient.Explain(ctx, "SELECT * FROM topic('events') WHERE key = 'user-123'")
if err != nil {
    log.Fatal(err)
}
fmt.Println(plan)

Configuration

Producer Configuration

config := streamline.DefaultConfig()
config.Producer = streamline.ProducerConfig{
    RequiredAcks:    -1,                      // All replicas
    Compression:     1,                       // gzip
    BatchSize:       16384,                   // 16KB
    BatchTimeout:    10 * time.Millisecond,
    Idempotent:      true,                    // Enable EOS
    Retries:         3,
}

Consumer Configuration

config.Consumer = streamline.ConsumerConfig{
    GroupID:           "my-group",
    AutoOffsetReset:   "earliest",
    SessionTimeout:    30 * time.Second,
    HeartbeatInterval: 3 * time.Second,
    MaxPollRecords:    500,
    IsolationLevel:    1,  // Read committed
}

SASL Authentication

config.SASL = &streamline.SASLConfig{
    Mechanism: "SCRAM-SHA-256",
    Username:  "user",
    Password:  "password",
}

TLS Configuration

config.TLS = &streamline.TLSConfig{
    Enable:   true,
    CertFile: "/path/to/client.crt",
    KeyFile:  "/path/to/client.key",
    CAFile:   "/path/to/ca.crt",
}

Error Handling

All errors returned by the SDK implement the standard error interface. Structured errors are returned as *StreamlineError with error codes, hints, and retryability information.

result, err := client.Producer.Send(ctx, "my-topic", nil, []byte("data"))
if err != nil {
    var se *streamline.StreamlineError
    if errors.As(err, &se) {
        switch se.Code {
        case streamline.ErrTopicNotFound:
            log.Printf("Topic missing: %s (hint: %s)", se.Message, se.Hint)
        case streamline.ErrAuthentication:
            log.Fatalf("Auth failed: %s", se.Message)
        case streamline.ErrTimeout:
            log.Printf("Timeout — retryable: %v", se.Retryable)
        default:
            log.Printf("Error [%s]: %s", se.Code, se.Message)
        }
    } else {
        log.Printf("Unexpected error: %v", err)
    }
}

Error Codes

Code	Constant	Retryable	Description
CONNECTION_ERROR	ErrConnection	✅	Server unreachable or connection dropped
AUTHENTICATION_ERROR	ErrAuthentication	❌	SASL/TLS credentials rejected
AUTHORIZATION_ERROR	ErrAuthorization	❌	ACL denied the operation
TOPIC_NOT_FOUND	ErrTopicNotFound	❌	Topic does not exist
TIMEOUT	ErrTimeout	✅	Operation exceeded deadline
PRODUCER_ERROR	ErrProducer	✅	Send failed (network, batch, etc.)
CONSUMER_ERROR	ErrConsumer	✅	Consume/poll failed
SERIALIZATION_ERROR	ErrSerialization	❌	Message encode/decode failed
CONFIGURATION_ERROR	ErrConfiguration	❌	Invalid client configuration
INTERNAL_ERROR	ErrInternal	❌	Unexpected internal error

Helper Functions

// Check if any error is a StreamlineError
if streamline.IsStreamlineError(err) {
    code := streamline.GetErrorCode(err)
    log.Printf("Streamline error code: %s", code)
}

// Check if an error is safe to retry
if streamline.IsRetryable(err) {
    // Back off and retry the operation
}

// Unwrap to inspect the underlying cause
var se *streamline.StreamlineError
if errors.As(err, &se) && se.Err != nil {
    log.Printf("Caused by: %v", se.Err)
}

Circuit Breaker

The SDK includes a circuit breaker that protects your application from cascading failures. The Producer uses it automatically; you can also use it directly:

import "github.com/streamlinelabs/streamline-go-sdk/streamline"

cb := streamline.NewCircuitBreaker(streamline.CircuitBreakerConfig{
    FailureThreshold: 5,                // Open after 5 consecutive failures
    SuccessThreshold: 2,                // Close after 2 half-open successes
    OpenTimeout:      30 * time.Second, // Probe interval
    OnStateChange: func(from, to streamline.CircuitState) {
        log.Printf("Circuit: %s → %s", from, to)
    },
})

if cb.Allow() {
    err := doSomething()
    if err != nil {
        cb.RecordFailure()
    } else {
        cb.RecordSuccess()
    }
}

When the circuit is open, Allow() returns false and operations are rejected immediately. See the Circuit Breaker guide for details.

API Reference

Client

Method	Description
NewClient(config)	Create a new client
client.Close()	Close the client
client.NewConsumer(ctx, groupID, topics)	Create a consumer

Producer

Method	Description
Send(ctx, topic, key, value)	Send a message synchronously
SendMessage(ctx, msg)	Send a message with full options
SendAsync(msg)	Send a message asynchronously
SendBatch(ctx, messages)	Send multiple messages
Close()	Close the producer

Consumer

Method	Description
Start(ctx)	Start consuming, returns message and error channels
Poll(ctx, maxRecords, timeout)	Poll for messages
Commit()	Commit offsets
Close()	Close the consumer

Admin

Method	Description
CreateTopic(ctx, config)	Create a topic
DeleteTopic(ctx, name)	Delete a topic
ListTopics(ctx)	List all topics
DescribeTopic(ctx, name)	Get topic details
AddPartitions(ctx, name, count)	Add partitions
ListConsumerGroups(ctx)	List consumer groups
DescribeConsumerGroup(ctx, groupID)	Get group details
DeleteConsumerGroup(ctx, groupID)	Delete a consumer group
ResetConsumerGroupOffsets(ctx, groupID, topic, offset)	Reset offsets

Query

Method	Description
NewQueryClient(baseURL)	Create a query client for the HTTP API
Query(ctx, sql)	Execute a SQL query with default options
QueryWithOptions(ctx, sql, opts)	Execute a SQL query with custom options
Explain(ctx, sql)	Get the query execution plan

Requirements

• Go 1.22 or later
• Streamline server 0.2.0 or later

Examples

The examples/ directory contains runnable examples:

Example	Description
Basic Usage	Produce, consume, and admin operations
Query Usage	SQL analytics with the embedded query engine
Schema Registry	Schema registration and validation
Circuit Breaker	Resilient production with circuit breaker
Security	TLS and SASL authentication

Run any example:

go run examples/main.go
go run examples/circuit_breaker/main.go

Moonshot Features

⚠️ Experimental — These features require Streamline server 0.3.0+ with moonshot feature flags enabled.

Semantic Search

Query topics by meaning instead of offset. Requires a topic created with semantic.embed=true.

results, err := consumer.Search(ctx, "logs.app", "payment failure", 10)
if err != nil {
    log.Fatal(err)
}
for _, hit := range results {
    log.Printf("[p%d] offset=%d score=%.2f", hit.Partition, hit.Offset, hit.Score)
}

Attestation Verification

Verify cryptographic provenance attestations attached to records by data contracts.

verifier, err := streamline.NewVerifier(publicKeyBytes)
if err != nil {
    log.Fatal(err)
}
result, err := verifier.Verify(record)
log.Printf("Verified: %v, Producer: %s", result.Verified, result.ProducerID)

Agent Memory (MCP)

Use Streamline as persistent memory for AI agents via the MCP protocol.

memory := streamline.NewMemoryClient("http://localhost:9094/mcp/v1")
err := memory.Remember(ctx, "user prefers dark mode", streamline.MemoryTags("preferences"))
results, err := memory.Recall(ctx, "user preferences", 5)

Branched Streams

Create topic branches for replay, A/B testing, or counterfactual analysis.

branch, err := client.Admin.CreateBranch(ctx, "events", "experiment-v2")
if err != nil {
    log.Fatal(err)
}
consumer, err := client.NewConsumer(ctx, "branch-group", []string{branch.Topic})
messages, errors := consumer.Start(ctx)

Contributing

Contributions are welcome! Please see the organization contributing guide for guidelines.

License

Apache-2.0

Security

To report a security vulnerability, please email security@streamline.dev. Do not open a public issue.

See the Security Policy for details.