metadata

package
v1.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 2, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotOwner is returned when a broker tries to write to a partition it does not own.
	ErrNotOwner = errors.New("broker does not own this partition")

	// ErrShuttingDown is returned when a lease acquisition is attempted after
	// ReleaseAll has been called. Callers should treat this the same as
	// ErrNotOwner — the broker is draining and should not accept new writes.
	ErrShuttingDown = errors.New("lease manager is shut down")
)
View Source 
var (
	// ErrTopicExists indicates the topic is already present.
	ErrTopicExists = errors.New("topic already exists")
	// ErrInvalidTopic indicates the topic specification is invalid.
	ErrInvalidTopic = errors.New("invalid topic configuration")
	// ErrUnknownTopic indicates the topic does not exist.
	ErrUnknownTopic = errors.New("unknown topic")
)
View Source 
var (
	// ErrStoreUnavailable is returned when the metadata store cannot be reached.
	ErrStoreUnavailable = errors.New("metadata store unavailable")
)

Functions

func BrokerRegistrationKey 

func BrokerRegistrationKey(brokerID string) string

BrokerRegistrationKey returns the etcd key for broker liveness data.

func ConsumerGroupKey 

func ConsumerGroupKey(groupID string) string

ConsumerGroupKey returns the etcd key for a consumer group metadata blob.

func ConsumerGroupPrefix 

func ConsumerGroupPrefix() string

ConsumerGroupPrefix returns the etcd prefix for consumer groups.

func ConsumerOffsetKey 

func ConsumerOffsetKey(groupID, topic string, partition int32) string

ConsumerOffsetKey returns the etcd key holding the committed offset for a partition.

func DecodeConsumerGroup 

func DecodeConsumerGroup(data []byte) (*metadatapb.ConsumerGroup, error)

DecodeConsumerGroup parses bytes into a ConsumerGroup struct.

func DecodePartitionState 

func DecodePartitionState(data []byte) (*metadatapb.PartitionState, error)

DecodePartitionState parses bytes into a PartitionState.

func DecodeTopicConfig 

func DecodeTopicConfig(data []byte) (*metadatapb.TopicConfig, error)

DecodeTopicConfig parses bytes into a TopicConfig object.

func EncodeConsumerGroup 

func EncodeConsumerGroup(group *metadatapb.ConsumerGroup) ([]byte, error)

EncodeConsumerGroup serializes a ConsumerGroup.

func EncodePartitionState 

func EncodePartitionState(state *metadatapb.PartitionState) ([]byte, error)

EncodePartitionState serializes a PartitionState into bytes.

func EncodeTopicConfig 

func EncodeTopicConfig(cfg *metadatapb.TopicConfig) ([]byte, error)

EncodeTopicConfig serializes a TopicConfig into etcd-ready bytes.

func GroupLeasePrefix added in v1.6.0 

func GroupLeasePrefix() string

GroupLeasePrefix returns the etcd prefix for watching group leases.

func ParseConsumerGroupID 

func ParseConsumerGroupID(key string) (string, bool)

ParseConsumerGroupID extracts a group ID from a metadata key.

func ParseConsumerOffsetKey 

func ParseConsumerOffsetKey(key string) (string, string, int32, bool)

ParseConsumerOffsetKey extracts group, topic, and partition from an offset key.

func PartitionAssignmentKey 

func PartitionAssignmentKey(topic string, partition int32) string

PartitionAssignmentKey returns the etcd key for the current leader assignment.

func PartitionLeasePrefix added in v1.6.0 

func PartitionLeasePrefix() string

PartitionLeasePrefix returns the etcd prefix for watching partition leases.

func PartitionStateKey 

func PartitionStateKey(topic string, partition int32) string

PartitionStateKey returns the etcd key for a partition state object.

func TopicConfigKey 

func TopicConfigKey(topic string) string

TopicConfigKey returns the etcd key for a topic configuration object.

func TopicIDForName 

func TopicIDForName(name string) [16]byte

TopicIDForName returns a stable 16-byte ID derived from a topic name.

Types

type AcquireResult added in v1.6.0 

type AcquireResult struct {
	Partition PartitionID
	Err       error
}

AcquireResult holds the outcome of a single partition lease acquisition.

type ClusterMetadata 

type ClusterMetadata struct {
	Brokers      []protocol.MetadataBroker
	ControllerID int32
	Topics       []protocol.MetadataTopic
	ClusterName  *string
	ClusterID    *string
}

ClusterMetadata describes the Kafka-visible cluster state.

type ConsumerOffset 

type ConsumerOffset struct {
	Group     string
	Topic     string
	Partition int32
	Offset    int64
}

ConsumerOffset captures a committed offset entry.

type EtcdStore 

type EtcdStore struct {
	// contains filtered or unexported fields
}

EtcdStore uses etcd for offset persistence while delegating metadata to an in-memory snapshot.

func NewEtcdStore 

func NewEtcdStore(ctx context.Context, snapshot ClusterMetadata, cfg EtcdStoreConfig) (*EtcdStore, error)

NewEtcdStore initializes a store backed by etcd.

func (*EtcdStore) Available 

func (s *EtcdStore) Available() bool

Available reports whether the most recent etcd operation succeeded.

func (*EtcdStore) CommitConsumerOffset 

func (s *EtcdStore) CommitConsumerOffset(ctx context.Context, group, topic string, partition int32, offset int64, metadata string) error

CommitConsumerOffset implements Store.CommitConsumerOffset.

func (*EtcdStore) CreatePartitions 

func (s *EtcdStore) CreatePartitions(ctx context.Context, topic string, partitionCount int32) error

CreatePartitions expands a topic and writes new partition state entries.

func (*EtcdStore) CreateTopic 

func (s *EtcdStore) CreateTopic(ctx context.Context, spec TopicSpec) (*protocol.MetadataTopic, error)

CreateTopic currently updates only the in-memory snapshot; the operator is still responsible for reconciling durable topic configuration into etcd/S3.

func (*EtcdStore) DeleteConsumerGroup 

func (s *EtcdStore) DeleteConsumerGroup(ctx context.Context, groupID string) error

DeleteConsumerGroup removes persisted consumer group metadata.

func (*EtcdStore) DeleteTopic 

func (s *EtcdStore) DeleteTopic(ctx context.Context, name string) error

DeleteTopic updates the local snapshot so admin APIs behave consistently.

func (*EtcdStore) EtcdClient added in v1.6.0 

func (s *EtcdStore) EtcdClient() *clientv3.Client

func (*EtcdStore) FetchConsumerGroup 

func (s *EtcdStore) FetchConsumerGroup(ctx context.Context, groupID string) (*metadatapb.ConsumerGroup, error)

FetchConsumerGroup loads consumer group metadata from etcd.

func (*EtcdStore) FetchConsumerOffset 

func (s *EtcdStore) FetchConsumerOffset(ctx context.Context, group, topic string, partition int32) (int64, string, error)

FetchConsumerOffset implements Store.FetchConsumerOffset.

func (*EtcdStore) FetchTopicConfig 

func (s *EtcdStore) FetchTopicConfig(ctx context.Context, topic string) (*metadatapb.TopicConfig, error)

FetchTopicConfig loads topic configuration from etcd or falls back to defaults.

func (*EtcdStore) ListConsumerGroups 

func (s *EtcdStore) ListConsumerGroups(ctx context.Context) ([]*metadatapb.ConsumerGroup, error)

ListConsumerGroups returns all persisted consumer groups.

func (*EtcdStore) ListConsumerOffsets 

func (s *EtcdStore) ListConsumerOffsets(ctx context.Context) ([]ConsumerOffset, error)

ListConsumerOffsets returns all committed offsets stored in etcd.

func (*EtcdStore) Metadata 

func (s *EtcdStore) Metadata(ctx context.Context, topics []string) (*ClusterMetadata, error)

Metadata delegates to the snapshot captured at startup (operator keeps it fresh).

func (*EtcdStore) NextOffset 

func (s *EtcdStore) NextOffset(ctx context.Context, topic string, partition int32) (int64, error)

NextOffset reads the last committed offset from etcd and returns the next offset to assign.

func (*EtcdStore) PutConsumerGroup 

func (s *EtcdStore) PutConsumerGroup(ctx context.Context, group *metadatapb.ConsumerGroup) error

PutConsumerGroup persists consumer group metadata in etcd.

func (*EtcdStore) UpdateOffsets 

func (s *EtcdStore) UpdateOffsets(ctx context.Context, topic string, partition int32, lastOffset int64) error

UpdateOffsets stores the next offset (last + 1) so future producers pick up from there.

func (*EtcdStore) UpdateTopicConfig 

func (s *EtcdStore) UpdateTopicConfig(ctx context.Context, cfg *metadatapb.TopicConfig) error

UpdateTopicConfig persists topic configuration into etcd.

type EtcdStoreConfig 

type EtcdStoreConfig struct {
	Endpoints   []string
	Username    string
	Password    string
	DialTimeout time.Duration
}

EtcdStoreConfig defines how we connect to etcd for metadata/offsets.

type GroupLeaseConfig added in v1.6.0 

type GroupLeaseConfig struct {
	BrokerID        string
	LeaseTTLSeconds int
	Logger          *slog.Logger
}

GroupLeaseConfig configures the group lease manager.

type GroupLeaseManager added in v1.6.0 

type GroupLeaseManager struct {
	// contains filtered or unexported fields
}

GroupLeaseManager uses etcd leases to ensure exclusive group coordination ownership. It delegates to the generic LeaseManager.

func NewGroupLeaseManager added in v1.6.0 

func NewGroupLeaseManager(client *clientv3.Client, cfg GroupLeaseConfig) *GroupLeaseManager

NewGroupLeaseManager creates a group lease manager backed by the given etcd client.

func (*GroupLeaseManager) Acquire added in v1.6.0 

func (m *GroupLeaseManager) Acquire(ctx context.Context, groupID string) error

Acquire tries to grab the group coordination lease. If this broker already owns it, it returns nil immediately. If another broker owns it, it returns ErrNotOwner.

func (*GroupLeaseManager) CurrentOwner added in v1.6.0 

func (m *GroupLeaseManager) CurrentOwner(ctx context.Context, groupID string) (string, error)

CurrentOwner queries etcd to find the current owner of a group. Returns the broker ID of the owner, or empty string if unowned.

func (*GroupLeaseManager) EtcdClient added in v1.6.0 

func (m *GroupLeaseManager) EtcdClient() *clientv3.Client

EtcdClient returns the underlying etcd client.

func (*GroupLeaseManager) Owns added in v1.6.0 

func (m *GroupLeaseManager) Owns(groupID string) bool

Owns returns true if this broker currently holds the lease for the group.

func (*GroupLeaseManager) Release added in v1.6.0 

func (m *GroupLeaseManager) Release(groupID string)

Release explicitly gives up ownership of a single group.

func (*GroupLeaseManager) ReleaseAll added in v1.6.0 

func (m *GroupLeaseManager) ReleaseAll()

ReleaseAll releases all group leases. Called during graceful shutdown.

type GroupRoute added in v1.6.0 

type GroupRoute struct {
	GroupID  string
	BrokerID string
}

GroupRoute maps a group ID to the broker that currently coordinates it.

type GroupRouter added in v1.6.0 

type GroupRouter struct {
	// contains filtered or unexported fields
}

GroupRouter watches etcd group lease keys and maintains an in-memory routing table. The proxy uses this to send group coordination requests to the correct broker without round-tripping to etcd on every request.

func NewGroupRouter added in v1.6.0 

func NewGroupRouter(ctx context.Context, client *clientv3.Client, logger *slog.Logger) (*GroupRouter, error)

NewGroupRouter creates a router and starts watching etcd for group lease changes.

func (*GroupRouter) AllRoutes added in v1.6.0 

func (r *GroupRouter) AllRoutes() []GroupRoute

AllRoutes returns a snapshot of the current routing table.

func (*GroupRouter) Invalidate added in v1.6.0 

func (r *GroupRouter) Invalidate(groupID string)

Invalidate removes a group from the routing table. Called when the proxy discovers (via NOT_COORDINATOR) that the cached route is stale. The next watch event will repopulate it with the correct owner.

func (*GroupRouter) LookupOwner added in v1.6.0 

func (r *GroupRouter) LookupOwner(groupID string) string

LookupOwner returns the broker ID that coordinates a group, or "" if unknown/unowned.

func (*GroupRouter) Stop added in v1.6.0 

func (r *GroupRouter) Stop()

Stop terminates the background watcher.

type InMemoryStore 

type InMemoryStore struct {
	// contains filtered or unexported fields
}

InMemoryStore is a simple Store backed by in-process state. Useful for early development and tests.

func NewInMemoryStore 

func NewInMemoryStore(state ClusterMetadata) *InMemoryStore

NewInMemoryStore builds an in-memory metadata store with the provided state.

func (*InMemoryStore) CommitConsumerOffset 

func (s *InMemoryStore) CommitConsumerOffset(ctx context.Context, group, topic string, partition int32, offset int64, metadata string) error

CommitConsumerOffset implements Store.CommitConsumerOffset.

func (*InMemoryStore) CreatePartitions 

func (s *InMemoryStore) CreatePartitions(ctx context.Context, topic string, partitionCount int32) error

CreatePartitions implements Store.CreatePartitions.

func (*InMemoryStore) CreateTopic 

func (s *InMemoryStore) CreateTopic(ctx context.Context, spec TopicSpec) (*protocol.MetadataTopic, error)

CreateTopic implements Store.CreateTopic.

func (*InMemoryStore) DeleteConsumerGroup 

func (s *InMemoryStore) DeleteConsumerGroup(ctx context.Context, groupID string) error

DeleteConsumerGroup implements Store.DeleteConsumerGroup.

func (*InMemoryStore) DeleteTopic 

func (s *InMemoryStore) DeleteTopic(ctx context.Context, name string) error

DeleteTopic implements Store.DeleteTopic.

func (*InMemoryStore) FetchConsumerGroup 

func (s *InMemoryStore) FetchConsumerGroup(ctx context.Context, groupID string) (*metadatapb.ConsumerGroup, error)

FetchConsumerGroup implements Store.FetchConsumerGroup.

func (*InMemoryStore) FetchConsumerOffset 

func (s *InMemoryStore) FetchConsumerOffset(ctx context.Context, group, topic string, partition int32) (int64, string, error)

FetchConsumerOffset implements Store.FetchConsumerOffset.

func (*InMemoryStore) FetchTopicConfig 

func (s *InMemoryStore) FetchTopicConfig(ctx context.Context, topic string) (*metadatapb.TopicConfig, error)

FetchTopicConfig implements Store.FetchTopicConfig.

func (*InMemoryStore) ListConsumerGroups 

func (s *InMemoryStore) ListConsumerGroups(ctx context.Context) ([]*metadatapb.ConsumerGroup, error)

ListConsumerGroups implements Store.ListConsumerGroups.

func (*InMemoryStore) ListConsumerOffsets 

func (s *InMemoryStore) ListConsumerOffsets(ctx context.Context) ([]ConsumerOffset, error)

ListConsumerOffsets implements Store.ListConsumerOffsets.

func (*InMemoryStore) Metadata 

func (s *InMemoryStore) Metadata(ctx context.Context, topics []string) (*ClusterMetadata, error)

Metadata implements Store.

func (*InMemoryStore) NextOffset 

func (s *InMemoryStore) NextOffset(ctx context.Context, topic string, partition int32) (int64, error)

NextOffset implements Store.NextOffset.

func (*InMemoryStore) PutConsumerGroup 

func (s *InMemoryStore) PutConsumerGroup(ctx context.Context, group *metadatapb.ConsumerGroup) error

PutConsumerGroup implements Store.PutConsumerGroup.

func (*InMemoryStore) Update 

func (s *InMemoryStore) Update(state ClusterMetadata)

Update swaps the cluster metadata atomically.

func (*InMemoryStore) UpdateOffsets 

func (s *InMemoryStore) UpdateOffsets(ctx context.Context, topic string, partition int32, lastOffset int64) error

UpdateOffsets implements Store.UpdateOffsets.

func (*InMemoryStore) UpdateTopicConfig 

func (s *InMemoryStore) UpdateTopicConfig(ctx context.Context, cfg *metadatapb.TopicConfig) error

UpdateTopicConfig implements Store.UpdateTopicConfig.

type LeaseManager added in v1.6.0 

type LeaseManager struct {
	// contains filtered or unexported fields
}

LeaseManager uses etcd leases to ensure exclusive ownership of named resources.

All lease keys are attached to a single shared etcd session/lease, so the keepalive cost is O(1) regardless of resource count. When the session dies (broker crash, network partition), etcd expires all keys after the TTL and the manager bulk-clears its local ownership map.

Concurrent Acquire calls for the same resource are deduplicated via singleflight to avoid redundant etcd round-trips and session leaks.

func NewLeaseManager added in v1.6.0 

func NewLeaseManager(client *clientv3.Client, cfg LeaseManagerConfig) *LeaseManager

NewLeaseManager creates a lease manager backed by the given etcd client.

func (*LeaseManager) Acquire added in v1.6.0 

func (m *LeaseManager) Acquire(ctx context.Context, resourceID string) error

Acquire tries to grab the lease for the named resource. If this broker already owns it, it returns nil immediately. If another broker owns it, it returns ErrNotOwner.

func (*LeaseManager) CurrentOwner added in v1.6.0 

func (m *LeaseManager) CurrentOwner(ctx context.Context, resourceID string) (string, error)

CurrentOwner queries etcd to find the current owner of a resource. Returns the broker ID of the owner, or empty string if unowned.

func (*LeaseManager) EtcdClient added in v1.6.0 

func (m *LeaseManager) EtcdClient() *clientv3.Client

EtcdClient returns the underlying etcd client. Used by routers that need to watch the same prefix.

func (*LeaseManager) Owns added in v1.6.0 

func (m *LeaseManager) Owns(resourceID string) bool

Owns returns true if this broker currently holds the lease for the resource.

func (*LeaseManager) Prefix added in v1.6.0 

func (m *LeaseManager) Prefix() string

Prefix returns the etcd prefix for watching leases.

func (*LeaseManager) Release added in v1.6.0 

func (m *LeaseManager) Release(resourceID string)

Release explicitly gives up ownership of a single resource.

func (*LeaseManager) ReleaseAll added in v1.6.0 

func (m *LeaseManager) ReleaseAll()

ReleaseAll releases all leases. Called during graceful shutdown.

type LeaseManagerConfig added in v1.6.0 

type LeaseManagerConfig struct {
	BrokerID        string
	Prefix          string // etcd key prefix, e.g. "/kafscale/partition-leases"
	LeaseTTLSeconds int
	Logger          *slog.Logger
	ResourceKind    string // used in log messages, e.g. "partition" or "group"
}

LeaseManagerConfig configures a generic lease manager.

type PartitionID added in v1.6.0 

type PartitionID struct {
	Topic     string
	Partition int32
}

PartitionID identifies a topic-partition pair.

type PartitionLeaseConfig added in v1.6.0 

type PartitionLeaseConfig struct {
	// BrokerID identifies this broker in lease keys.
	BrokerID string
	// LeaseTTLSeconds controls how long a lease persists after the broker stops refreshing.
	LeaseTTLSeconds int
	// Logger for operational messages.
	Logger *slog.Logger
}

PartitionLeaseConfig configures the lease manager.

type PartitionLeaseManager added in v1.6.0 

type PartitionLeaseManager struct {
	// contains filtered or unexported fields
}

PartitionLeaseManager uses etcd leases to ensure exclusive partition ownership. It delegates to the generic LeaseManager, translating topic+partition pairs into string resource IDs.

func NewPartitionLeaseManager added in v1.6.0 

func NewPartitionLeaseManager(client *clientv3.Client, cfg PartitionLeaseConfig) *PartitionLeaseManager

NewPartitionLeaseManager creates a lease manager backed by the given etcd client.

func (*PartitionLeaseManager) Acquire added in v1.6.0 

func (m *PartitionLeaseManager) Acquire(ctx context.Context, topic string, partition int32) error

Acquire tries to grab the partition lease. If this broker already owns it, it returns nil immediately. If another broker owns it, it returns ErrNotOwner.

func (*PartitionLeaseManager) AcquireAll added in v1.6.0 

func (m *PartitionLeaseManager) AcquireAll(ctx context.Context, partitions []PartitionID) []AcquireResult

AcquireAll attempts to acquire leases for all given partitions concurrently. Partitions already owned by this broker are skipped (no etcd round-trip). Returns a result per partition; callers should check each Err.

func (*PartitionLeaseManager) CurrentOwner added in v1.6.0 

func (m *PartitionLeaseManager) CurrentOwner(ctx context.Context, topic string, partition int32) (string, error)

CurrentOwner queries etcd to find the current owner of a partition. Returns the broker ID of the owner, or empty string if unowned.

func (*PartitionLeaseManager) EtcdClient added in v1.6.0 

func (m *PartitionLeaseManager) EtcdClient() *clientv3.Client

EtcdClient returns the underlying etcd client.

func (*PartitionLeaseManager) Owns added in v1.6.0 

func (m *PartitionLeaseManager) Owns(topic string, partition int32) bool

Owns returns true if this broker currently holds the lease for the partition.

func (*PartitionLeaseManager) Release added in v1.6.0 

func (m *PartitionLeaseManager) Release(topic string, partition int32)

Release explicitly gives up ownership of a single partition.

func (*PartitionLeaseManager) ReleaseAll added in v1.6.0 

func (m *PartitionLeaseManager) ReleaseAll()

ReleaseAll releases all partition leases. Called during graceful shutdown.

type PartitionRoute added in v1.6.0 

type PartitionRoute struct {
	Topic     string
	Partition int32
	BrokerID  string
}

PartitionRoute maps a topic/partition to the broker that currently owns it.

type PartitionRouter added in v1.6.0 

type PartitionRouter struct {
	// contains filtered or unexported fields
}

PartitionRouter watches etcd partition lease keys and maintains an in-memory routing table. The proxy uses this to send produce requests to the correct broker without round-tripping to etcd on every request.

The routing table is best-effort: it can be briefly stale after a lease change. The broker's own ownership check is the authoritative guard.

func NewPartitionRouter added in v1.6.0 

func NewPartitionRouter(ctx context.Context, client *clientv3.Client, logger *slog.Logger) (*PartitionRouter, error)

NewPartitionRouter creates a router and starts watching etcd for lease changes.

func (*PartitionRouter) AllRoutes added in v1.6.0 

func (r *PartitionRouter) AllRoutes() []PartitionRoute

AllRoutes returns a snapshot of the current routing table.

func (*PartitionRouter) Invalidate added in v1.6.0 

func (r *PartitionRouter) Invalidate(topic string, partition int32)

Invalidate removes a partition from the routing table. Called when the proxy discovers (via NOT_LEADER_OR_FOLLOWER) that the cached route is stale. The next watch event will repopulate it with the correct owner.

func (*PartitionRouter) LookupOwner added in v1.6.0 

func (r *PartitionRouter) LookupOwner(topic string, partition int32) string

LookupOwner returns the broker ID that owns a partition, or "" if unknown/unowned.

func (*PartitionRouter) Stop added in v1.6.0 

func (r *PartitionRouter) Stop()

Stop terminates the background watcher.

type Store 

type Store interface {
	// Metadata returns brokers, controller ID, and topics. When topics is non-empty,
	// the implementation should filter to that subset and omit missing topics.
	Metadata(ctx context.Context, topics []string) (*ClusterMetadata, error)
	// NextOffset returns the next offset to assign for a topic/partition.
	NextOffset(ctx context.Context, topic string, partition int32) (int64, error)
	// UpdateOffsets records the last persisted offset so future appends continue from there.
	UpdateOffsets(ctx context.Context, topic string, partition int32, lastOffset int64) error
	// CommitConsumerOffset persists a consumer group offset.
	CommitConsumerOffset(ctx context.Context, group, topic string, partition int32, offset int64, metadata string) error
	// FetchConsumerOffset retrieves the committed offset for a consumer group partition.
	FetchConsumerOffset(ctx context.Context, group, topic string, partition int32) (int64, string, error)
	// ListConsumerOffsets returns all committed consumer offsets.
	ListConsumerOffsets(ctx context.Context) ([]ConsumerOffset, error)
	// PutConsumerGroup persists consumer group metadata.
	PutConsumerGroup(ctx context.Context, group *metadatapb.ConsumerGroup) error
	// FetchConsumerGroup retrieves consumer group metadata.
	FetchConsumerGroup(ctx context.Context, groupID string) (*metadatapb.ConsumerGroup, error)
	// ListConsumerGroups returns all stored consumer group metadata.
	ListConsumerGroups(ctx context.Context) ([]*metadatapb.ConsumerGroup, error)
	// DeleteConsumerGroup removes consumer group metadata.
	DeleteConsumerGroup(ctx context.Context, groupID string) error
	// FetchTopicConfig returns the stored topic configuration.
	FetchTopicConfig(ctx context.Context, topic string) (*metadatapb.TopicConfig, error)
	// UpdateTopicConfig persists topic configuration updates.
	UpdateTopicConfig(ctx context.Context, cfg *metadatapb.TopicConfig) error
	// CreatePartitions expands a topic's partition count.
	CreatePartitions(ctx context.Context, topic string, partitionCount int32) error
	// CreateTopic creates a new topic with the provided specification.
	CreateTopic(ctx context.Context, spec TopicSpec) (*protocol.MetadataTopic, error)
	// DeleteTopic removes a topic and associated offsets.
	DeleteTopic(ctx context.Context, name string) error
}

Store exposes read-only access to cluster metadata used by Kafka protocol handlers.

type TopicSpec 

type TopicSpec struct {
	Name              string
	NumPartitions     int32
	ReplicationFactor int16
}

TopicSpec describes a topic creation request.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.