tidb_kafka

Documentation

Index

• Variables
• func KafkaPositionValueDecoder(value string) (interface{}, error)
• func KafkaPositionValueEncoder(v interface{}) (string, error)
• func ParseTimeStamp(tso uint64) uint64
• type BinlogTailer
  • func NewBinlogTailer(pipelineName string, serverID uint32, ...) (*BinlogTailer, error)
  • func (t *BinlogTailer) Close()
  • func (t *BinlogTailer) Start() error
  • func (t *BinlogTailer) Wait()
• type ConsumerGroupOffset
• type KafkaOffsetStoreFactory
  • func NewKafkaOffsetStoreFactory(pipelineName string, positionCache position_store.PositionCacheInterface) *KafkaOffsetStoreFactory
  • func (f *KafkaOffsetStoreFactory) GenOffsetStore(c *sarama_cluster.Consumer) sarama_cluster.OffsetStore
• type KafkaPositionValue
• type OffsetStore
  • func (store *OffsetStore) Close() error
  • func (store *OffsetStore) CommitOffset(req *offsets.OffsetCommitRequest) (*offsets.OffsetCommitResponse, error)
  • func (store *OffsetStore) FetchOffset(req *offsets.OffsetFetchRequest) (*offsets.OffsetFetchResponse, error)
• type TopicOffset

Variables

var (
	BinlogCheckInterval = time.Second
)

Functions

func KafkaPositionValueDecoder

func KafkaPositionValueDecoder(value string) (interface{}, error)

func KafkaPositionValueEncoder

func KafkaPositionValueEncoder(v interface{}) (string, error)

func ParseTimeStamp

func ParseTimeStamp(tso uint64) uint64

Types

type BinlogTailer

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

func NewBinlogTailer

func NewBinlogTailer(
	pipelineName string,
	serverID uint32,
	positionCache position_store.PositionCacheInterface,
	config *gCfg.SourceTiDBConfig,
	emitter core.Emitter,
	binlogChecker binlog_checker.BinlogChecker,
) (*BinlogTailer, error)

func (*BinlogTailer) Close

func (t *BinlogTailer) Close()

func (*BinlogTailer) Start

func (t *BinlogTailer) Start() error

func (*BinlogTailer) Wait

func (t *BinlogTailer) Wait()

type ConsumerGroupOffset

type ConsumerGroupOffset map[string]TopicOffset

type KafkaOffsetStoreFactory

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

func NewKafkaOffsetStoreFactory

func NewKafkaOffsetStoreFactory(pipelineName string, positionCache position_store.PositionCacheInterface) *KafkaOffsetStoreFactory

func (*KafkaOffsetStoreFactory) GenOffsetStore

func (f *KafkaOffsetStoreFactory) GenOffsetStore(c *sarama_cluster.Consumer) sarama_cluster.OffsetStore

type KafkaPositionValue

type KafkaPositionValue struct {
	Offsets map[string]ConsumerGroupOffset `json:"offsets"`
}

type OffsetStore

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

func (*OffsetStore) Close

func (store *OffsetStore) Close() error

func (*OffsetStore) CommitOffset

func (store *OffsetStore) CommitOffset(req *offsets.OffsetCommitRequest) (*offsets.OffsetCommitResponse, error)

func (*OffsetStore) FetchOffset

func (store *OffsetStore) FetchOffset(req *offsets.OffsetFetchRequest) (*offsets.OffsetFetchResponse, error)

type TopicOffset

type TopicOffset map[int32]int64