The Go SDK, a blocking client speaking the VSR wire protocol over TCP, and which servers it pairs with.

The Iggy Go SDK is a client library for interacting with the Iggy server from Go applications. It is a blocking client that speaks the VSR (Viewstamped Replication) wire protocol over TCP, with or without TLS. The package is available on pkg.go.dev and the source code lives in foreign/go.

Server compatibility

The SDK and the server must speak the same wire protocol. Tagged releases up to v0.8.0 predate VSR (and the current API) and pair only with servers of the same era. The current SDK speaks VSR only. This page targets server 0.10.0. Use a compatible SDK release or edge version. When working from source, build the SDK and the server from the same repository checkout.

Installation

The current source requires Go 1.25 or newer. Run these commands from your application module:

go get github.com/apache/iggy/foreign/go

To pin the release that matches server 0.10.0:

go get github.com/apache/iggy/foreign/go@v0.10.0

For unreleased changes, use the local SDK replacement shown in the repository examples. A published edge tag need not contain every change in your checkout.

Quick start

The snippets below follow the getting-started example. Stream, topic, and partition IDs are 0-based: the first partition of a topic is partition 0.

Run the server from the repository root with disposable development data and no root credential overrides:

cargo run --bin iggy-server -- --fresh --with-default-root-credentials

--fresh wipes this replica's local data directory. IGGY_ROOT_USERNAME and IGGY_ROOT_PASSWORD take precedence over the flag. Bootstrap settings do not replace recovered credentials, including credentials recovered from cluster peers. The snippets use iggy/iggy. Run the producer before the consumer.

Producer

package main

import (
	"context"
	"errors"
	"log"

	"github.com/apache/iggy/foreign/go/client"
	"github.com/apache/iggy/foreign/go/client/tcp"
	iggcon "github.com/apache/iggy/foreign/go/contracts"
	ierror "github.com/apache/iggy/foreign/go/errors"
)

func main() {
	cli, err := client.NewIggyClient(
		client.WithTcp(tcp.WithServerAddress("127.0.0.1:8090")),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := cli.Close(); err != nil {
			log.Printf("Error closing client: %v", err)
		}
	}()

	ctx := context.Background()
	if err := cli.Connect(ctx); err != nil {
		log.Fatal(err)
	}

	if _, err := cli.LoginUser(ctx, "iggy", "iggy"); err != nil {
		log.Fatal(err)
	}

	// Re-running this example is fine: an existing stream or topic is not an error.
	if _, err := cli.CreateStream(ctx, "sample-stream", nil); err != nil &&
		!errors.Is(err, ierror.ErrStreamNameAlreadyExists) {
		log.Fatal(err)
	}

	streamId, err := iggcon.NewIdentifier("sample-stream")
	if err != nil {
		log.Fatal(err)
	}

	if _, err := cli.CreateTopic(
		ctx,
		streamId,
		"sample-topic",
		nil,
	); err != nil && !errors.Is(err, ierror.ErrTopicNameAlreadyExists) {
		log.Fatal(err)
	}

	topicId, err := iggcon.NewIdentifier("sample-topic")
	if err != nil {
		log.Fatal(err)
	}

	message, err := iggcon.NewIggyMessage([]byte("Hello from Go!"))
	if err != nil {
		log.Fatal(err)
	}

	if _, err := cli.SendMessages(
		ctx,
		streamId,
		topicId,
		iggcon.None(),
		[]iggcon.IggyMessage{message},
	); err != nil {
		log.Fatal(err)
	}

	log.Println("Message sent")
}

CreateTopic accepts *iggcon.TopicOptions. A nil value selects server defaults. CreateStream accepts *iggcon.StreamOptions. UpdateTopic, UpdateStream and UpdateUser accept typed patches. DescribeOptions lists the keys and mutability flags.

Consumer

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/apache/iggy/foreign/go/client"
	"github.com/apache/iggy/foreign/go/client/tcp"
	iggcon "github.com/apache/iggy/foreign/go/contracts"
)

func main() {
	cli, err := client.NewIggyClient(
		client.WithTcp(tcp.WithServerAddress("127.0.0.1:8090")),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := cli.Close(); err != nil {
			log.Printf("Error closing client: %v", err)
		}
	}()

	ctx := context.Background()
	if err := cli.Connect(ctx); err != nil {
		log.Fatal(err)
	}

	if _, err := cli.LoginUser(ctx, "iggy", "iggy"); err != nil {
		log.Fatal(err)
	}

	streamId, err := iggcon.NewIdentifier("sample-stream")
	if err != nil {
		log.Fatal(err)
	}
	topicId, err := iggcon.NewIdentifier("sample-topic")
	if err != nil {
		log.Fatal(err)
	}

	offset := uint64(0)
	partitionId := uint32(0)
	consumer := iggcon.DefaultConsumer()

	for {
		polled, err := cli.PollMessages(
			ctx,
			streamId,
			topicId,
			partitionId,
			consumer,
			iggcon.OffsetPollingStrategy(offset),
			iggcon.WaitPollOptions(5*time.Second),
		)
		if err != nil {
			log.Fatal(err)
		}

		if len(polled.Messages) == 0 {
			time.Sleep(500 * time.Millisecond)
			continue
		}

		for _, message := range polled.Messages {
			fmt.Printf("Offset: %d, Payload: %s\n",
				message.Header.Offset, string(message.Payload))
			offset = message.Header.Offset + 1
		}
	}
}

The consumer reads partition 0 independently and keeps its offset only in memory. Restarting reads from offset 0 again, or the earliest retained message if older messages were removed.

Uncompressed payloads and user headers alias the reply buffer, so retaining their slices pins the whole reply. Copy those bytes to retain only the message you need. S2 decompression allocates a separate payload buffer.

TLS

Replace the quick-start client constructor with the following TCP options. Start the TLS server as shown on the examples page. The development CA path below is relative to examples/go:

cli, err := client.NewIggyClient(
	client.WithTcp(
		tcp.WithServerAddress("localhost:8090"),
		tcp.WithTLS(
			tcp.WithTLSCAFile("../../core/certs/iggy_ca_cert.pem"),
			tcp.WithTLSDomain("localhost"),
		),
	),
)

tcp.WithTLSValidateCertificate controls certificate validation. The client also accepts client.WithLogger for a custom *slog.Logger and runs a built-in heartbeat loop (every 5 seconds by default) once connected.

Delivery semantics

SendMessages returns the placements reported by the server. An uncertain mutation retains its frozen request and original identity. Reconnection binds the existing session before attempting a new registration. ResolveOperation retries the retained identity, and AbandonOperation releases local state without undoing the server result. Keep the same client for recovery. A fresh send after losing receipt access can duplicate work.

Crash durability follows the topic's durability policy: replicated confirms replication, while persisted also waits for the required replicas to persist the message data. See Durability. An empty confirmation list is a valid success but does not by itself prove that new messages were appended.

Consumer groups

The client exposes CreateConsumerGroup, JoinConsumerGroup, LeaveConsumerGroup, GetConsumerGroups and SyncConsumerGroup. A one-shot PollMessages reads an explicit partition. Use reader.NewPartitionReader or reader.NewIggyConsumer with iggcon.NewGroupConsumer to read the full assignment.

The reader joins, refreshes assignments and leaves on Close. Its default wait is five seconds. Commit stores a delivered checkpoint after processing. iggcon.AutoCommitPolicy(iggcon.DefaultAutoCommit()) selects automatic commits after processing. See Deferred polling.

Connection strings and metadata

client.ParseConnectionString returns TCP options from a connection string. Pass them to client.WithTcp when constructing the client. It accepts credentials, TLS and reconnect options. Use GetMetadataChanges to read resource or activity changes without listing the entire catalog. Raw requests can target a replica with SendBinaryRequestToReplica. See Metadata changes for restart and retention behavior.

Examples

A runnable getting-started example with producer, consumer, and TLS flags lives in examples/go. See the examples page for how to run it.

Go SDK

On this page