kafka

Documentation

Index

• Constants
• type Reader
  • func NewReader(cfg ReaderConfig) Reader
  • func (r Reader) Read(ctx context.Context, task streams.ReadTask) (err error)
• type ReaderConfig
• type Writer
  • func NewWriter(kafkaWriter *kafka.Writer) Writer
  • func (w Writer) Write(ctx context.Context, msgBatch []streams.Message) error

Constants

const (
	// HeaderClientID is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the connection identifier from the underlying Apache Kafka client.
	HeaderClientID = "streams-kafka-client-id"
	// HeaderGroupID is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the consumer group identifier the reader instance is in.
	HeaderGroupID = "streams-kafka-group-id"
	// HeaderPartitionID is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the partition identifier the reader instance is reading from.
	HeaderPartitionID = "streams-kafka-partition-id"
	// HeaderInitialOffset is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the offset a reader instance started to read messages from the topic's partition append log.
	HeaderInitialOffset = "streams-kafka-init-offset"
	// HeaderCurrentOffset is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the offset of a message a reader instance is currently reading from.
	HeaderCurrentOffset = "streams-kafka-current-offset"
	// HeaderHighWaterMarkOffset is a header key passed to reader handlers -through streams.Message's Headers field-.
	// This key represents the next offset available from a topic partition's append log.
	HeaderHighWaterMarkOffset = "streams-kafka-hwm-offset"
)

const (
	// ReaderTaskGroupIDKey is the argument key to set up a reader to be placed in an Apache Kafka consumer group.
	ReaderTaskGroupIDKey string = "kafka-group-id"
	// ReaderTaskPartitionIDKey is the argument key to set up a reader to read from a specific partition of an Apache Kafka topic.
	ReaderTaskPartitionIDKey string = "kafka-partition-id"
	// ReaderTaskInitialOffsetKey is the argument key to set up a reader to start reading messages
	// from a specific offset from the topic's partition append log.
	ReaderTaskInitialOffsetKey string = "kafka-init-offset"
)

Types

type Reader

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

A Reader type is the concrete implementation of streams.Reader using Apache Kafka.

func NewReader

func NewReader(cfg ReaderConfig) Reader

NewReader allocates a Reader instance.

func (Reader) Read

func (r Reader) Read(ctx context.Context, task streams.ReadTask) (err error)

type ReaderConfig

type ReaderConfig struct {
	kafka.ReaderConfig
	HandlerTimeout time.Duration // Maximum duration for message handler processes (streams.ReaderHandleFunc).
}

ReaderConfig is the configuration for Reader.

type Writer

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

A Writer type is the concrete implementation of streams.Writer using Apache Kafka.

func NewWriter

func NewWriter(kafkaWriter *kafka.Writer) Writer

NewWriter allocates a Writer instance.

func (Writer) Write

func (w Writer) Write(ctx context.Context, msgBatch []streams.Message) error