kafka

Documentation

Index

• Constants
• type KafkaClient
  • func NewKafkaClient(topic string, readerLastOffset int64, addrs ...string) (*KafkaClient, error)
  • func (k *KafkaClient) Broadcast(ctx context.Context, info *pb.BlockInfo) error
  • func (k *KafkaClient) Fetch(ctx context.Context) (infos []*pb.BlockInfo, err error)
  • func (k *KafkaClient) FetchStart(ctx context.Context, start int64) (infos []*pb.BlockInfo, err error)
  • func (k *KafkaClient) IncrementLastReaderOffset()
  • func (k *KafkaClient) LastReaderOffset() int64
  • func (k *KafkaClient) LastWriterOffset() int64
  • func (k *KafkaClient) RemoteOffset() (firstOffset int64, lastOffset int64, err error)
  • func (k *KafkaClient) ResetLastReaderOffset(offset int64)
  • func (k *KafkaClient) ResetTopic(topic string)

Constants

const (
	KafkaMaxBytes = 1 << 21
	KafkaMaxWait  = 500 * time.Millisecond
)

Types

type KafkaClient

type KafkaClient struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewKafkaClient

func NewKafkaClient(topic string, readerLastOffset int64, addrs ...string) (*KafkaClient, error)

func (*KafkaClient) Broadcast

func (k *KafkaClient) Broadcast(ctx context.Context, info *pb.BlockInfo) error

func (*KafkaClient) Fetch

func (k *KafkaClient) Fetch(ctx context.Context) (infos []*pb.BlockInfo, err error)

func (*KafkaClient) FetchStart

func (k *KafkaClient) FetchStart(ctx context.Context, start int64) (infos []*pb.BlockInfo, err error)

func (*KafkaClient) IncrementLastReaderOffset

func (k *KafkaClient) IncrementLastReaderOffset()

func (*KafkaClient) LastReaderOffset

func (k *KafkaClient) LastReaderOffset() int64

func (*KafkaClient) LastWriterOffset

func (k *KafkaClient) LastWriterOffset() int64

func (*KafkaClient) RemoteOffset

func (k *KafkaClient) RemoteOffset() (firstOffset int64, lastOffset int64, err error)

func (*KafkaClient) ResetLastReaderOffset

func (k *KafkaClient) ResetLastReaderOffset(offset int64)

func (*KafkaClient) ResetTopic

func (k *KafkaClient) ResetTopic(topic string)