kafkabp

type ConsumeErrorFunc func(err error)

type ConsumeMessageFunc func(ctx context.Context, msg *sarama.ConsumerMessage)

type ConsumePartitionFunc func(partitionID int32) bool

type ConsumePartitionFuncProvider func(numPartitions int) ConsumePartitionFunc

type Consumer interface {
	io.Closer

	Consume(ConsumeMessageFunc, ConsumeErrorFunc) error

	// IsHealthy returns false after Consume returns.
	IsHealthy(ctx context.Context) bool
}

type ConsumerConfig struct {
	// Required. Brokers specifies a slice of broker addresses.
	Brokers []string `yaml:"brokers"`

	// Required. Topic is used to specify the topic to consume.
	Topic string `yaml:"topic"`

	// Required. ClientID is used by Kafka broker to track clients' consuming
	// progresses on the topics.
	//
	// In most cases, every instance is expected to have a unique ClientID.
	// The Kubernetes pod ID is usually a good candidate for this unique ID.
	ClientID string `yaml:"clientID"`

	// Optional. When GroupID is non-empty, a new group consumer will be created
	// instead. Messages from the topic will be consumed by one of the consumers
	// in the group (sharing the same GroupID) exactly once. This is the usual use
	// case of streaming consumers.
	//
	// When GroupID is empty, each consumer will have the whole view of the topic
	// (based on Offset), so that is usually for use cases like to deliver
	// configs/data through Kafka brokers.
	//
	// When GroupID is non-empty, Version must be at least "0.10.2.0".
	GroupID string `yaml:"groupID"`

	// Optional. This is only applicable when GroupID is empty string.
	// When GroupID is empty, the configuration enables a TopicConsumer to read
	// from all partitions of the given kafka stream.
	// When ConsumePartitionFuncProvider is also specified, the TopicConsumer will
	// only consume partitions that evaluates to true with the given
	// predicate returned from ConsumePartitionFuncProvider.
	// If ConsumePartitionFuncProvider is specified,
	// it must return a non-nil predicate of type ConsumePartitionFunc,
	// else the topic consumer will return error ErrNilConsumePartitionFunc.
	//
	// This function is called once per reset. The returned ConsumePartitionFunc
	// is also called once per reset per partition.
	// The API is designed to be two-layer so that it's possible to shift all
	// heavylifting to ConsumePartitionFuncProvider to generate a []bool or set of
	// ints, and make the returned ConsumePartitionFunc just do a simple lookup.
	ConsumePartitionFuncProvider ConsumePartitionFuncProvider `yaml:"-"`

	// Optional. The version of the kafka broker this consumer is connected to.
	// In format of "0.10.2.0" or "2.4.0".
	//
	// When omitted, Sarama library would pick the oldest supported version in
	// order to maintain maximum backward compatibility, but some of the newer
	// features might be unavailable. For example, using GroupID requires the
	// version to be at least "0.10.2.0".
	Version string `yaml:"version"`

	// Optional. Defaults to "oldest". Valid values are "oldest" and "newest".
	//
	// Only used when GroupID is empty.
	Offset string `yaml:"offset"`

	// Optional. If non-nil, will be used to log errors. At present, this only
	// pertains to logging errors closing the existing consumer when calling
	// consumer.reset() when GroupID is empty.
	Logger log.Wrapper `yaml:"logger"`

	// Optional. The function to set rack id for this kafka client.
	// It should match rack configured on the broker(s).
	//
	// Currently it defaults to no rack id.
	// In the future the default might be changed to AWSAvailabilityZoneRackID.
	//
	// This feature is currently experimental.
	// It might not make any difference on your client,
	// or it might make things worse.
	// You are advised to test before using non-empty rack id in production.
	RackID RackIDFunc `yaml:"rackID"`
}

type GroupConsumerHandler struct {
	Callback ConsumeMessageFunc
	Topic    string
}

type RackIDFunc func() string

type SaramaConfigOverrider func(*sarama.Config)

type SimpleHTTPRackIDConfig struct {
	// URL to fetch rack id from. Required.
	URL string

	// Limit of how many bytes to read from the response.
	//
	// Optional, default to SimpleHTTPRackIDDefaultLimit.
	Limit int64

	// HTTP client timeout.
	//
	// Optional, default to SimpleHTTPRackIDDefaultTimeout.
	Timeout time.Duration

	// Logger to be used on http errors. Optional.
	Logger log.Wrapper
}