kafka

type SaramaCachedClient ¶

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

SaramaCachedClient wraps a sarama kafka client, exposing common used methods at sources and targets.

If a method fails it will try to re-connect and try again before failing.

func NewSaramaCachedClient(ctx context.Context, bootstrapServers []string, cfg *sarama.Config, logger *zap.Logger, opts ...SaramaCachedClientOption) (*SaramaCachedClient, error)

func (sc *SaramaCachedClient) Close() error

func (sc *SaramaCachedClient) EnsureTopic(topic string, replicas int16, partitions int32) error

func (sc *SaramaCachedClient) PingTopic(topic string) error

func (sc *SaramaCachedClient) RefreshProducerClients() error

func (sc *SaramaCachedClient) SendMessageSync(message *sarama.ProducerMessage) error

type SaramaCachedClientOption ¶

type SaramaCachedClientOption func(*SaramaCachedClient)

func WithSaramaCachedClientRefresh(d time.Duration) SaramaCachedClientOption