go_kafka_client

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

type BlueGreenDeployment struct {
	// Comma separated list of topics to consume from
	Topics string
	// Either black_list, white_list or static
	Pattern string
	//Consumer group to switch to
	Group string
}

type BrokerInfo struct {
	Version int16
	Id      int32
	Host    string
	Port    uint32
}

type ByteDecoder struct{}

type ByteEncoder struct{}

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

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

type ConsumerConfig struct {
	/* A string that uniquely identifies a set of consumers within the same consumer group */
	Groupid string

	/* A string that uniquely identifies a consumer within a group. Generated automatically if not set.
	Set this explicitly for only testing purpose. */
	Consumerid string

	/* The socket timeout for network requests. Its value should be at least FetchWaitMaxMs. */
	SocketTimeout time.Duration

	/* The maximum number of bytes to attempt to fetch */
	FetchMessageMaxBytes int32

	/* The number of goroutines used to fetch data */
	NumConsumerFetchers int

	/* Max number of message batches buffered for consumption, each batch can be up to FetchBatchSize */
	QueuedMaxMessages int32

	/* Max number of retries during rebalance */
	RebalanceMaxRetries int32

	/* The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will block */
	FetchMinBytes int32

	/* The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy FetchMinBytes */
	FetchWaitMaxMs int32

	/* Backoff time between retries during rebalance */
	RebalanceBackoff time.Duration

	/* Backoff time to refresh the leader of a partition after it loses the current leader */
	RefreshLeaderBackoff time.Duration

	/* Retry the offset commit up to this many times on failure. */
	OffsetsCommitMaxRetries int

	/* Try to commit offset every OffsetCommitInterval. If previous offset commit for a partition is still in progress updates the next offset to commit and continues.
	This way it does not commit all the offset history if the coordinator is slow, but only the highest offsets. */
	OffsetCommitInterval time.Duration

	/* What to do if an offset is out of range.
	SmallestOffset : automatically reset the offset to the smallest offset.
	LargestOffset : automatically reset the offset to the largest offset.
	Defaults to LargestOffset. */
	AutoOffsetReset string

	/* Client id is specified by the kafka consumer client, used to distinguish different clients. */
	Clientid string

	/* Whether messages from internal topics (such as offsets) should be exposed to the consumer. */
	ExcludeInternalTopics bool

	/* Select a strategy for assigning partitions to consumer streams. Possible values: RangeStrategy, RoundRobinStrategy */
	PartitionAssignmentStrategy string

	/* Amount of workers per partition to process consumed messages. */
	NumWorkers int

	/* Times to retry processing a failed message by a worker. */
	MaxWorkerRetries int

	/* Worker retry threshold. Increments each time a worker fails to process a message within MaxWorkerRetries.
	When this threshold is hit within a WorkerThresholdTimeWindow, WorkerFailureCallback is called letting the user to decide whether the consumer should stop. */
	WorkerRetryThreshold int32

	/* Resets WorkerRetryThreshold if it isn't hit within this period. */
	WorkerThresholdTimeWindow time.Duration

	/* Callback executed when WorkerRetryThreshold exceeded within WorkerThresholdTimeWindow */
	WorkerFailureCallback FailedCallback

	/* Callback executed when Worker failed to process the message after MaxWorkerRetries and WorkerRetryThreshold is not hit */
	WorkerFailedAttemptCallback FailedAttemptCallback

	/* Worker timeout to process a single message. */
	WorkerTaskTimeout time.Duration

	/* Backoff between worker attempts to process a single message. */
	WorkerBackoff time.Duration

	/* Maximum wait time to gracefully stop a worker manager */
	WorkerManagersStopTimeout time.Duration

	/* A function which defines a user-specified action on a single message. This function is responsible for actual message processing.
	Consumer panics if Strategy is not set. */
	Strategy WorkerStrategy

	/* Number of messages to accumulate before flushing them to workers */
	FetchBatchSize int

	/* Timeout to accumulate messages. Flushes accumulated batch to workers even if it is not yet full.
	Resets after each flush meaning this won't be triggered if FetchBatchSize is reached before timeout. */
	FetchBatchTimeout time.Duration

	//Debug topic prefix to log when FetchBatchTimeout is reached for auto-flushing
	BatchFlushTimeoutPrefix string

	/* Backoff between fetch requests if no messages were fetched from a previous fetch. */
	RequeueAskNextBackoff time.Duration

	/* Buffer size for ask next channel. This value shouldn't be less than number of partitions per fetch routine. */
	AskNextChannelSize int

	/* Maximum fetch retries if no messages were fetched from a previous fetch */
	FetchMaxRetries int

	/* Maximum retries to fetch topic metadata from one broker. */
	FetchTopicMetadataRetries int

	/* Backoff for fetch topic metadata request if the previous request failed. */
	FetchTopicMetadataBackoff time.Duration

	/* Backoff between two fetch requests for one fetch routine. Needed to prevent fetcher from querying the broker too frequently. */
	FetchRequestBackoff time.Duration

	/* Coordinator used to coordinate consumer's actions, e.g. trigger rebalance events, store offsets and consumer metadata etc. */
	Coordinator ConsumerCoordinator

	/* OffsetStorage is used to store and retrieve consumer offsets. */
	OffsetStorage OffsetStorage

	/* Indicates whether the client supports blue-green deployment.
	This config entry is needed because blue-green deployment won't work with RoundRobin partition assignment strategy.
	Defaults to true. */
	BlueGreenDeploymentEnabled bool

	/* Time to wait after consumer has registered itself in group */
	DeploymentTimeout time.Duration

	/* Service coordinator barrier timeout */
	BarrierTimeout time.Duration

	/* Low Level Kafka Client implementation. */
	LowLevelClient LowLevelClient

	/* Message keys decoder */
	KeyDecoder Decoder

	/* Message values decoder */
	ValueDecoder Decoder

	/* Flag for debug mode */
	Debug bool

	/* Metrics Prefix if the client wants to organize the way metric names are emitted. (optional) */
	MetricsPrefix string

	/* RoutinePoolSize defines the size of routine pools created within this consumer. */
	RoutinePoolSize int
}

type ConsumerCoordinator interface {
	/* Establish connection to this ConsumerCoordinator. Returns an error if fails to connect, nil otherwise. */
	Connect() error

	/* Close connection to this ConsumerCoordinator. */
	Disconnect()

	/* Registers a new consumer with Consumerid id and TopicCount subscription that is a part of consumer group Group in this ConsumerCoordinator. Returns an error if registration failed, nil otherwise. */
	RegisterConsumer(Consumerid string, Group string, TopicCount TopicsToNumStreams) error

	/* Deregisters consumer with Consumerid id that is a part of consumer group Group form this ConsumerCoordinator. Returns an error if deregistration failed, nil otherwise. */
	DeregisterConsumer(Consumerid string, Group string) error

	/* Gets the information about consumer with Consumerid id that is a part of consumer group Group from this ConsumerCoordinator.
	Returns ConsumerInfo on success and error otherwise (For example if consumer with given Consumerid does not exist). */
	GetConsumerInfo(Consumerid string, Group string) (*ConsumerInfo, error)

	/* Gets the information about consumers per topic in consumer group Group excluding internal topics (such as offsets) if ExcludeInternalTopics = true.
	Returns a map where keys are topic names and values are slices of consumer ids and fetcher ids associated with this topic and error on failure. */
	GetConsumersPerTopic(Group string, ExcludeInternalTopics bool) (map[string][]ConsumerThreadId, error)

	/* Gets the list of all consumer ids within a consumer group Group. Returns a slice containing all consumer ids in group and error on failure. */
	GetConsumersInGroup(Group string) ([]string, error)

	/* Gets the list of all topics registered in this ConsumerCoordinator. Returns a slice conaining topic names and error on failure. */
	GetAllTopics() ([]string, error)

	/* Gets the information about existing partitions for a given Topics.
	Returns a map where keys are topic names and values are slices of partition ids associated with this topic and error on failure. */
	GetPartitionsForTopics(Topics []string) (map[string][]int32, error)

	/* Gets the information about all Kafka brokers registered in this ConsumerCoordinator.
	Returns a slice of BrokerInfo and error on failure. */
	GetAllBrokers() ([]*BrokerInfo, error)

	/* Subscribes for any change that should trigger consumer rebalance on consumer group Group in this ConsumerCoordinator or trigger topic switch.
	Returns a read-only channel of CoordinatorEvent that will get values on any significant coordinator event (e.g. new consumer appeared, new broker appeared etc.) and error if failed to subscribe. */
	SubscribeForChanges(Group string) (<-chan CoordinatorEvent, error)

	/* Requests that a blue/green deployment be done.*/
	RequestBlueGreenDeployment(blue BlueGreenDeployment, green BlueGreenDeployment) error

	/* Gets all deployed topics for consume group Group from consumer coordinator.
	Returns a map where keys are notification ids and values are DeployedTopics. May also return an error (e.g. if failed to reach coordinator). */
	GetBlueGreenRequest(Group string) (map[string]*BlueGreenDeployment, error)

	/* Implements classic barrier synchronization primitive via service coordinator facilities */
	AwaitOnStateBarrier(consumerId string, group string, stateHash string, barrierSize int, api string, timeout time.Duration) bool

	/* Removes state barrier */
	RemoveStateBarrier(group string, stateHash string, api string) error

	/* Tells the ConsumerCoordinator to unsubscribe from events for the consumer it is associated with. */
	Unsubscribe()

	/* Tells the ConsumerCoordinator to claim partition topic Topic and partition Partition for ConsumerThreadId fetcher that works within a consumer group Group.
	Returns true if claim is successful, false and error explaining failure otherwise. */
	ClaimPartitionOwnership(Group string, Topic string, Partition int32, ConsumerThreadId ConsumerThreadId) (bool, error)

	/* Tells the ConsumerCoordinator to release partition ownership on topic Topic and partition Partition for consumer group Group.
	Returns error if failed to released partition ownership. */
	ReleasePartitionOwnership(Group string, Topic string, Partition int32) error

	/* Removes old api objects */
	RemoveOldApiRequests(group string) error
}

type ConsumerGroupApi string

type ConsumerInfo struct {
	Version      int16          `json:"version"`
	Subscription map[string]int `json:"subscription"`
	Pattern      string         `json:"pattern"`
	Timestamp    int64          `json:"timestamp,string"`
}

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

type ConsumerThreadId struct {
	Consumer string
	ThreadId int
}

type CoordinatorEvent string

type Decoder interface {
	Decode([]byte) (interface{}, error)
}

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

type EmptyEmitter string

type Encoder interface {
	Encode(interface{}) ([]byte, error)
}

type ErrorType int

type FailedAttemptCallback func(*Task, WorkerResult) FailedDecision

type FailedCallback func(*WorkerManager) FailedDecision

type FailedDecision int32

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

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

type KafkaLogger interface {
	//Formats a given message according to given params to log with level Trace.
	Trace(message string, params ...interface{})

	//Formats a given message according to given params to log with level Debug.
	Debug(message string, params ...interface{})

	//Formats a given message according to given params to log with level Info.
	Info(message string, params ...interface{})

	//Formats a given message according to given params to log with level Warn.
	Warn(message string, params ...interface{})

	//Formats a given message according to given params to log with level Error.
	Error(message string, params ...interface{})

	//Formats a given message according to given params to log with level Critical.
	Critical(message string, params ...interface{})

	GetLogLevel() LogLevel

	IsAllowed(logLevel LogLevel) bool
}

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

type LogEmitter interface {
	// Emit sends a single LogLine to some destination.
	Emit(*avro.LogLine)

	// Close closes the given emitter.
	Close()
}

type LogLevel string

type LowLevelClient interface {
	// This will be called right after connecting to ConsumerCoordinator so this client can initialize itself
	// with bootstrap broker list for example. May return an error to signal this client is unable to work with given configuration.
	Initialize() error

	// This will be called each time the fetch request to Kafka should be issued. Topic, partition and offset are self-explanatory.
	// Should return a slice of Messages and an error if a fetch error occurred.
	// Note that for performance reasons it makes sense to keep open broker connections and reuse them on every fetch call.
	Fetch(topic string, partition int32, offset int64) ([]*Message, error)

	// This will be called when call to Fetch returns an error. As every client has different error mapping we ask here explicitly
	// what kind of error was returned.
	GetErrorType(error) ErrorType

	// This will be called to handle OffsetOutOfRange error. OffsetTime will be either "smallest" or "largest".
	// Should return a corresponding offset value and an error if it occurred.
	GetAvailableOffset(topic string, partition int32, offsetTime string) (int64, error)

	// This will be called to gracefully shutdown this client.
	Close()
}

type Message struct {
	// Partition key.
	Key []byte
	// Message value.
	Value []byte
	// Decoded message key
	DecodedKey interface{}
	// Decoded message value
	DecodedValue interface{}
	// Topic this message came from.
	Topic string

	// Partition this message came from.
	Partition int32

	// Message offset.
	Offset int64

	// HighwaterMarkOffset is an offset of the last message in this topic-partition.
	HighwaterMarkOffset int64
}

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

type MirrorMakerConfig struct {
	// Whitelist of topics to mirror. Exactly one whitelist or blacklist is allowed.
	Whitelist string

	// Blacklist of topics to mirror. Exactly one whitelist or blacklist is allowed.
	Blacklist string

	// Consumer configurations to consume from a source cluster.
	ConsumerConfigs []string

	// Embedded producer config.
	ProducerConfig string

	// Number of producer instances.
	NumProducers int

	// Number of consumption streams.
	NumStreams int

	// Flag to preserve partition number. E.g. if message was read from partition 5 it'll be written to partition 5. Note that this can affect performance.
	PreservePartitions bool

	// Flag to preserve message order. E.g. message sequence 1, 2, 3, 4, 5 will remain 1, 2, 3, 4, 5 in destination topic. Note that this can affect performance.
	PreserveOrder bool

	// Destination topic prefix. E.g. if message was read from topic "test" and prefix is "dc1_" it'll be written to topic "dc1_test".
	TopicPrefix string

	// Number of messages that are buffered between the consumer and producer.
	ChannelSize int

	// Message keys encoder for producer
	KeyEncoder producer.Serializer

	// Message values encoder for producer
	ValueEncoder producer.Serializer

	// Message keys decoder for consumer
	KeyDecoder Decoder

	// Message values decoder for consumer
	ValueDecoder Decoder
}

type OffsetStorage interface {
	// Gets the offset for a given group, topic and partition.
	// May return an error if fails to retrieve the offset.
	GetOffset(group string, topic string, partition int32) (int64, error)

	// Commits the given offset for a given group, topic and partition.
	// May return an error if fails to commit the offset.
	CommitOffset(group string, topic string, partition int32, offset int64) error
}

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

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

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

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

type StateSnapshot struct {
	// Metrics are a map where keys are event names and values are maps holding event values grouped by meters (count, min, max, etc.).
	Metrics map[string]map[string]float64
	// Offsets are a map where keys are topics and values are maps where keys are partitions and values are offsets for these topic-partitions.
	Offsets map[string]map[int32]int64
}

type StaticTopicsToNumStreams struct {
	// Consumer id string.
	ConsumerId string
	// Map where keys are topic names and values are number of fetcher routines responsible for processing these topics.
	TopicsToNumStreamsMap map[string]int
}

type StringDecoder struct{}

type StringEncoder struct{}

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

type Task struct {
	// A message that should be processed.
	Msg *Message

	// Number of retries used for this task.
	Retries int

	// A worker that is responsible for processing this task.
	Callee *Worker
}

type TaskAndStrategy struct {
	WorkerTask *Task
	Strategy   WorkerStrategy
}

type TaskId struct {
	// Message's topic and partition
	TopicPartition TopicAndPartition

	// Message's offset
	Offset int64
}

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

type TopicAndPartition struct {
	Topic     string
	Partition int32
}

type TopicFilter interface {
	Regex() string
	TopicAllowed(topic string, excludeInternalTopics bool) bool
}

type TopicInfo struct {
	Version    int16
	Partitions map[string][]int32
}

type TopicsToNumStreams interface {
	//Creates a map descibing consumer subscription where keys are topic names and values are number of fetchers used to fetch these topics.
	GetTopicsToNumStreamsMap() map[string]int

	//Creates a map describing consumer subscription where keys are topic names and values are slices of ConsumerThreadIds used to fetch these topics.
	GetConsumerThreadIdsPerTopic() map[string][]ConsumerThreadId

	//Returns a pattern describing this TopicsToNumStreams.
	Pattern() string
}

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

type WildcardTopicsToNumStreams struct {
	Coordinator           ConsumerCoordinator
	ConsumerId            string
	TopicFilter           TopicFilter
	NumStreams            int
	ExcludeInternalTopics bool
}

type Worker struct {
	// Channel to write tasks to.
	InputChannel chan *TaskAndStrategy

	// Channel to write processing results to.
	OutputChannel chan WorkerResult

	// Intermediate channel for pushing result to strategy handler
	HandlerInputChannel chan *TaskAndStrategy

	// Intermediate channel for pushing result from strategy handler
	HandlerOutputChannel chan WorkerResult

	// Timeout for a single worker task.
	TaskTimeout time.Duration

	// Indicates whether this worker is closed and cannot accept new work.
	Closed bool
}

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

type WorkerResult interface {
	// Returns an id of task that was processed.
	Id() TaskId

	// Returns true if processing succeeded, false otherwise.
	Success() bool
}

type WorkerStrategy func(*Worker, *Message, TaskId) WorkerResult

type ZookeeperConfig struct {
	/* Zookeeper hosts */
	ZookeeperConnect []string

	/* Zookeeper session timeout */
	ZookeeperSessionTimeout time.Duration

	/* Max retries for any request except CommitOffset. CommitOffset is controlled by ConsumerConfig.OffsetsCommitMaxRetries. */
	MaxRequestRetries int

	/* Backoff to retry any request */
	RequestBackoff time.Duration

	/* kafka Root */
	Root string

	/* PanicHandler is a function that will be called when unrecoverable error occurs to give the possibility to perform cleanups, recover from panic etc */
	PanicHandler func(error)
}

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