consumer

type AssignmentNotification func(assignments map[string][]int32) // assignments is a map from topic -> list of partitions

type Client interface {
	// Consume returns a consumer of the given topic
	Consume(topic string) (Consumer, error)

	// ConsumeMany starts consuming many topics at once. It is much more efficient than calling Consume
	// repeatedly because kafka brokers serialize joining topics
	ConsumeMany(topics []string) ([]Consumer, error)

	// Close closes the client. It must be called to shutdown
	// the client. It cleans up any unclosed topic Consumers created by this Client.
	// It does NOT close the inner sarama.Client.
	// Calling twice is NOT supported.
	Close()

	// Errors returns a channel which can (should) be monitored
	// for errors. callers should probably log or otherwise report
	// the returned errors. The channel closes when the client
	// is closed.
	Errors() <-chan error
}

type Config struct {
	Session struct {
		// The allowed session timeout for registered consumers (defaults to 30s).
		// Must be within the allowed server range.
		Timeout time.Duration
	}
	Rebalance struct {
		// The allowed rebalance timeout for registered consumers (defaults to 30s).
		// Must be within the allowed server range. Only functions if sarama.Config.Version >= 0.10.1
		// Otherwise Session.Timeout is used for rebalancing too.
		Timeout time.Duration
	}
	Heartbeat struct {
		// Interval between each heartbeat (defaults to 3s). It should be no more
		// than 1/3rd of the Group.Session.Timout setting
		Interval time.Duration
	}

	// the partitioner used to map partitions to consumer group members (defaults to a round-robin partitioner)
	Partitioner Partitioner

	// OffsetOutOfRange is the handler for sarama.ErrOffsetOutOfRange errors (defaults to sarama.OffsetNewest,nil).
	// Implementations must return the new starting offset in the partition, or an error. The sarama.Client is included
	// for convenience, since handling this might involve querying the partition's current offsets.
	OffsetOutOfRange OffsetOutOfRange

	// StartingOffset is a hook to allow modifying the starting offset when a Consumer begins to consume
	// a partition. (defaults to returning the last committed offset). Some consumers might want to jump
	// ahead to fresh messages. The sarama.Client is included for convenience, since handling this might involve
	// looking up a partition's offset by time. When no committed offset could be found -1 (sarama.OffsetNewest)
	// is passed in. An implementation might want to return client.Config().Consumer.Offsets.Initial in that case.
	StartingOffset StartingOffset

	// SidechannelOffset is a kafka topic used to exchange partition offsets between dying and rebalancing consumers.
	// It defaults to "sarama-consumer-sidechannel-offsets".  If SidechannelTopic is "" then this feature is disabled,
	// and consumers can rewind as much as Config.Rebalance.Timeout + sarama.Config.Offset.CommitInterval
	// when a partition is reassigned. That's always possible (kafka only promises at-least-once), but in high frequency
	// topics rewinding the default 30 seconds creates a measureable burst).
	// We can't comit offsets normally during a rebalance because at that point in time we still belong to the old generation,
	// but the broker belongs to the new generation. Hence this side channel.
	SidechannelTopic string

	// AssignmentNotification is an optional callback to inform the client code whenever the client gets a new
	// partition assignment.
	AssignmentNotification AssignmentNotification
}

type Consumer interface {
	// Messages returns the channel of messages arriving from kafka. It always
	// returns the same result, so it is safe to call once and store the result.
	// Every message read from the channel should be passed to Done when processing
	// of the message is complete.
	// It is not necessary to call Done in the same order as messages are received
	// from this channel.
	Messages() <-chan *sarama.ConsumerMessage

	// Done indicates the processing of the message is complete, and its offset can
	// be committed to kafka. Calling Done twice with the same message, or with a
	// garbage message, can cause trouble.
	Done(*sarama.ConsumerMessage)

	// AsyncClose terminates the consumer cleanly. Callers can continue to read from
	// Messages channel until it is closed, or not, as they wish.
	// Calling Client.Close() performs a AsyncClose() on any remaining consumers.
	// Calling AsyncClose multiple times is permitted. Only the first call has any effect.
	// Never calling AsyncClose is also permitted. Client.Close() implies Consumer.AsyncClose.
	AsyncClose()

	// Close terminates the consumer and waits for it to be finished committing the current
	// offsets to kafka. Calling twice happens to work at the moment, but let's not encourage it.
	Close()
}

type Error struct {
	Err       error    // underlying error
	Context   string   // description of the context surrounding the error
	Consumer  Consumer // nil, or Consumer which produced the error
	Topic     string   // "", or the topic which had the error
	Partition int32    // -1, or the partition which had the error
	// contains filtered or unexported fields
}

type OffsetOutOfRange func(topic string, partition int32, client sarama.Client) (offset int64, err error)

type Partitioner interface {
	// name this partitioner (used for log messages)
	Name() string

	// PrepareJoin prepares a JoinGroupRequest given the topics supplied.
	// The simplest implementation would be something like
	//   join_req.AddGroupProtocolMetadata("<partitioner name>", &sarama.ConsumerGroupMemberMetadata{ Version: 1, Topics:  topics, })
	PrepareJoin(join_req *sarama.JoinGroupRequest, topics []string, current_assignments map[string][]int32)

	// Partition performs the partitioning. Given the requested
	// memberships from the JoinGroupResponse, it adds the results
	// to the SyncGroupRequest. Returning an error cancels everything.
	// The sarama.Client supplied to NewClient is included for convenince,
	// since performing the partitioning probably requires looking at each
	// topic's metadata, especially its list of partitions.
	Partition(*sarama.SyncGroupRequest, *sarama.JoinGroupResponse, sarama.Client) error

	// ParseSync parses the SyncGroupResponse and returns the map of topics
	// to partitions assigned to this client, or an error if the information
	// is not parsable.
	ParseSync(*sarama.SyncGroupResponse) (map[string][]int32, error)
}

type SidechannelMsg struct {
	Ver           int                            // should be 1
	ConsumerGroup string                         // name of the consumer group sending the offsets (also used as the kafka message key)
	Offsets       map[string][]SidechannelOffset // map from topic to list of <partition,offset> pairs
}

type SidechannelOffset struct {
	Partition int32 `json:"p"` // use short field names in JSON to keep the size of the messages low
	Offset    int64 `json:"o"` // since there can be a lot of SidechannelOffsets in a SidechannelMsg

}

type StartingOffset func(topic string, partition int32, committed_offset int64, client sarama.Client) (offset int64, err error)