gki

Documentation

Index

• Constants
• type AdminClient
• type Config
  • func NewExtractorConfig(spec *entity.Spec, topics []string, topicCreationMutex *sync.Mutex) *Config
  • func NewLoaderConfig(spec *entity.Spec, topic *entity.TopicSpecification, ...) *Config
  • func (c *Config) SetCreateTopics(value bool)
  • func (c *Config) SetKafkaProperty(prop string, value any)
  • func (c *Config) SetPollTimout(timeout int)
  • func (c *Config) SetProps(props ConfigMap)
  • func (c *Config) String() string
• type ConfigMap
• type Consumer
• type ConsumerFactory
• type DefaultAdminClient
  • func (d DefaultAdminClient) CreateTopics(ctx context.Context, topics []kafka.TopicSpecification, ...) ([]kafka.TopicResult, error)
  • func (d DefaultAdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)
• type DefaultConsumerFactory
  • func (d DefaultConsumerFactory) NewAdminClientFromConsumer(c Consumer) (AdminClient, error)
  • func (d DefaultConsumerFactory) NewConsumer(conf *kafka.ConfigMap) (Consumer, error)
• type DefaultProducerFactory
  • func (d DefaultProducerFactory) NewAdminClientFromProducer(p Producer) (AdminClient, error)
  • func (d DefaultProducerFactory) NewProducer(conf *kafka.ConfigMap) (Producer, error)
• type Extractor
  • func NewExtractor(config *Config, id string) (*Extractor, error)
  • func (e *Extractor) Extract(ctx context.Context, query entity.ExtractorQuery, result any) (error, bool)
  • func (e *Extractor) ExtractFromSink(ctx context.Context, query entity.ExtractorQuery, ...) (error, bool)
  • func (e *Extractor) KafkaConfig() (pollTimeoutMs int, cfgMap map[string]any)
  • func (e *Extractor) SendToSource(ctx context.Context, eventData any) (string, error)
  • func (e *Extractor) SetConsumerFactory(cf ConsumerFactory)
  • func (e *Extractor) SetProducerFactory(pf ProducerFactory)
  • func (e *Extractor) StreamExtract(ctx context.Context, reportEvent entity.ProcessEventFunc, err *error, ...)
• type Loader
  • func NewLoader(ctx context.Context, config *Config, id string, pf ProducerFactory) (*Loader, error)
  • func (l *Loader) KafkaConfig() map[string]any
  • func (l *Loader) Shutdown()
  • func (l *Loader) StreamLoad(ctx context.Context, data []*entity.Transformed) (string, error, bool)
• type Producer
• type ProducerFactory

Constants

const (
	PropAutoCommit      = "enable.auto.commit"
	PropAutoOffsetStore = "enable.auto.offset.store"
)

Kafka props which cannot be overridden, required by Extractor

const DefaultPollTimeoutMs = 3000

Types

type AdminClient

type AdminClient interface {
	GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)
	CreateTopics(ctx context.Context, topics []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error)
}

type Config

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

Config is the internal config used by each extractor/loader, combining config from external Config with config from what is inside the stream spec for this specific stream

func NewExtractorConfig

func NewExtractorConfig(
	spec *entity.Spec,
	topics []string,
	topicCreationMutex *sync.Mutex) *Config

func NewLoaderConfig

func NewLoaderConfig(
	spec *entity.Spec,
	topic *entity.TopicSpecification,
	topicCreationMutex *sync.Mutex,
	sync bool) *Config

func (*Config) SetCreateTopics

func (c *Config) SetCreateTopics(value bool)

func (*Config) SetKafkaProperty

func (c *Config) SetKafkaProperty(prop string, value any)

func (*Config) SetPollTimout

func (c *Config) SetPollTimout(timeout int)

func (*Config) SetProps

func (c *Config) SetProps(props ConfigMap)

func (*Config) String

func (c *Config) String() string

type ConfigMap

type ConfigMap map[string]any

type Consumer

type Consumer interface {
	SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) error
	Poll(timeoutMs int) (event kafka.Event)
	StoreOffsets(offsets []kafka.TopicPartition) ([]kafka.TopicPartition, error)
	Commit() ([]kafka.TopicPartition, error)
	CommitMessage(m *kafka.Message) ([]kafka.TopicPartition, error)
	Close() error
}

Consumer interface is used to enable full unit testing

type ConsumerFactory

type ConsumerFactory interface {
	NewConsumer(conf *kafka.ConfigMap) (Consumer, error)
	NewAdminClientFromConsumer(c Consumer) (AdminClient, error)
}

type DefaultAdminClient

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

func (DefaultAdminClient) CreateTopics

func (d DefaultAdminClient) CreateTopics(ctx context.Context, topics []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error)

func (DefaultAdminClient) GetMetadata

func (d DefaultAdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)

type DefaultConsumerFactory

type DefaultConsumerFactory struct{}

func (DefaultConsumerFactory) NewAdminClientFromConsumer

func (d DefaultConsumerFactory) NewAdminClientFromConsumer(c Consumer) (AdminClient, error)

func (DefaultConsumerFactory) NewConsumer

func (d DefaultConsumerFactory) NewConsumer(conf *kafka.ConfigMap) (Consumer, error)

type DefaultProducerFactory

type DefaultProducerFactory struct{}

func (DefaultProducerFactory) NewAdminClientFromProducer

func (d DefaultProducerFactory) NewAdminClientFromProducer(p Producer) (AdminClient, error)

func (DefaultProducerFactory) NewProducer

func (d DefaultProducerFactory) NewProducer(conf *kafka.ConfigMap) (Producer, error)

type Extractor

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

func NewExtractor

func NewExtractor(config *Config, id string) (*Extractor, error)

func (*Extractor) Extract

func (e *Extractor) Extract(
	ctx context.Context,
	query entity.ExtractorQuery,
	result any) (error, bool)

func (*Extractor) ExtractFromSink

func (e *Extractor) ExtractFromSink(
	ctx context.Context,
	query entity.ExtractorQuery,
	result *[]*entity.Transformed) (error, bool)

func (*Extractor) KafkaConfig

func (e *Extractor) KafkaConfig() (pollTimeoutMs int, cfgMap map[string]any)

func (*Extractor) SendToSource

func (e *Extractor) SendToSource(ctx context.Context, eventData any) (string, error)

func (*Extractor) SetConsumerFactory

func (e *Extractor) SetConsumerFactory(cf ConsumerFactory)

func (*Extractor) SetProducerFactory

func (e *Extractor) SetProducerFactory(pf ProducerFactory)

func (*Extractor) StreamExtract

func (e *Extractor) StreamExtract(
	ctx context.Context,
	reportEvent entity.ProcessEventFunc,
	err *error,
	retryable *bool)

type Loader

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

func NewLoader

func NewLoader(ctx context.Context, config *Config, id string, pf ProducerFactory) (*Loader, error)

func (*Loader) KafkaConfig

func (l *Loader) KafkaConfig() map[string]any

func (*Loader) Shutdown

func (l *Loader) Shutdown()

func (*Loader) StreamLoad

func (l *Loader) StreamLoad(ctx context.Context, data []*entity.Transformed) (string, error, bool)

type Producer

type Producer interface {
	Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error
	Events() chan kafka.Event
	Flush(timeoutMs int) int
	Close()
}

Producer interface is used to enable full unit testing

type ProducerFactory

type ProducerFactory interface {
	NewProducer(conf *kafka.ConfigMap) (Producer, error)
	NewAdminClientFromProducer(p Producer) (AdminClient, error)
}