Go SDK

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.9.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.9.0:

go get github.com/apache/iggy/foreign/go@v0.9.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"); 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",
		1,
		iggcon.CompressionAlgorithmNone,
		iggcon.IggyExpiryNeverExpire,
		0,
	); 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 optional trailing iggcon.HeaderEntry values carrying named topic options, for option keys the server catalog gained after the SDK build shipped. DescribeOptions lists the keys a server accepts.

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,
			consumer,
			iggcon.OffsetPollingStrategy(offset),
			10,    // messages per batch
			false, // auto-commit
			&partitionId,
		)
		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 any placements the server reports. A send whose reply is lost to a dropped connection returns ErrDisconnected without a replay: a reconnect registers a fresh client identity, so the server could not deduplicate the replay against a batch that may have already committed. Retrying such a send is the caller's decision and may write the batch twice. Consumers must handle duplicates through idempotent processing or application-level deduplication.

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 the full consumer group API: CreateConsumerGroup, JoinConsumerGroup, LeaveConsumerGroup, GetConsumerGroups, and SyncConsumerGroup (which returns the partitions this client currently owns together with the group generation). To poll as a group member, build the consumer with iggcon.NewGroupConsumer and pass a nil partition id to PollMessages. The client orchestrates the group poll. An empty batch with no assigned partition means nothing to read right now (a member that owns no partition, or a rebalance still settling): poll again. ErrConsumerGroupMemberNotFound means the client is not a member, for example after an explicit LeaveConsumerGroup. JoinConsumerGroup restores membership.

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.