kafka

type AckHandler ¶

type AckHandler func(context.Context, AcknowledgedMessage) error

AckHandler handles a message requiring explicit acknowledgment intent.

type AcknowledgedMessage ¶

type AcknowledgedMessage struct {
	Message
	Ack func(context.Context) error
}

AcknowledgedMessage requires Ack before the handler returns successfully. Ack requests the post-handler offset commit; it is not a broker commit receipt.

type Admin ¶

type Admin interface {
	CreateTopic(context.Context, Topic) error
	DeleteTopic(context.Context, string) error
	ListTopics(context.Context) ([]TopicInfo, error)
	Close() error
}

Admin owns a Kafka administration client.

type Consumer ¶

type Consumer interface {
	Consume(context.Context, Handler) error
	Close() error
}

Consumer owns a Kafka consumer group client.

type CredentialProvider ¶

type CredentialProvider func(context.Context) (Credentials, error)

type Credentials ¶

type Credentials struct{ Username, Password string }

Credentials are fetched at authentication time. Providers must honor cancellation and avoid returning secrets in errors.

type Delivery ¶

type Delivery struct {
	Topic     string
	Partition int32
	Offset    int64
	Timestamp time.Time
}

Delivery contains broker-assigned delivery metadata.

type GroupInfo ¶

type GroupInfo struct {
	Name, State, ProtocolType string
	Members                   []GroupMember
}

GroupInfo describes consumer group state and membership.

type GroupMember ¶

type GroupMember struct{ ID, ClientID, Host string }

GroupMember identifies a current consumer group member.

type Handler ¶

type Handler func(context.Context, Message) error

Handler processes one message with cancellation support.

func DecodeHandler[T any](serializer encodingx.Serializer, handler func(context.Context, Message, T) error) Handler

func Recover(h Handler) Handler

func RequireAck(h AckHandler) Handler

func WithFailure(h Handler, p retry.Policy, producer Producer, dlq string) Handler

type Inspector ¶

type Inspector interface {
	Admin
	DescribeTopic(context.Context, string) (TopicDescription, error)
	ListGroups(context.Context) ([]GroupInfo, error)
	DescribeGroup(context.Context, string) (GroupInfo, error)
	Offsets(context.Context, string) ([]Offset, error)
}

Inspector extends basic administration with topic, group and committed-offset inspection.

type ManualConsumer ¶

type ManualConsumer interface {
	Consumer
	ConsumeAcknowledged(context.Context, AckHandler) error
}

ManualConsumer extends consumption with required explicit acknowledgment before successful return.

type Message ¶

type Message struct {
	Topic      string
	Key, Value []byte
	Headers    map[string][]byte
	Timestamp  time.Time
	Partition  int32
	Offset     int64
}

Message contains transport-independent message data.

type OAuthToken ¶

type OAuthToken struct {
	Token      string
	Extensions map[string]string
	ExpiresAt  time.Time
}

OAuthToken is an opaque SASL bearer token. The issuing authority determines its authorization scope.

type Offset ¶

type Offset struct {
	Topic     string
	Partition int32
	Offset    int64
	Metadata  string
}

Offset is the next offset a consumer group will resume from; -1 indicates no committed offset.

type PartitionInfo ¶

type PartitionInfo struct {
	ID, Leader               int32
	Replicas, InSyncReplicas []int32
}

PartitionInfo describes broker placement at one point in time.

type Producer ¶

type Producer interface {
	Publish(context.Context, Message) error
	PublishBatch(context.Context, []Message) error
	Flush(context.Context) error
	Close() error
}

Producer owns an acknowledged Kafka producer.

type TokenProvider ¶

type TokenProvider func(context.Context) (OAuthToken, error)

type Topic ¶

type Topic struct {
	Name              string
	Partitions        int32
	ReplicationFactor int16
	Config            map[string]*string
}

Topic describes explicit topic creation settings.

type TopicDescription ¶

type TopicDescription struct {
	Name       string
	Internal   bool
	Partitions []PartitionInfo
}

TopicDescription contains topic metadata and broker partition placement.

type TopicInfo ¶

type TopicInfo struct {
	Name       string
	Partitions []int32
	Internal   bool
}

TopicInfo contains basic topic metadata.