consumer.go

/*
  A simple kafka consumer-group client

  Copyright 2016 MistSys
*/

package consumer

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"sort"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/mistsys/sarama-consumer/roundrobin"
)

const logging = true        // set to true to see log messages
const debug = false         // set to true to see debug messages
const per_msg_debug = false // set to true to see per-message debug messages

// low level logging function. Replace it with your own if desired before making any calls to the rest of the API
var Logf func(fmt string, args ...interface{}) = log.Printf

// logf logs a printf style message if log is enabled
func logf(fmt string, args ...interface{}) {
	if logging {
		Logf(fmt, args...)
	}
}

// dbgf logs a printf style message to somewhere reasonable if debug is enabled, and as efficiently as it can does nothing with any side effects if debug is disabled
func dbgf(fmt string, args ...interface{}) {
	if debug {
		Logf(fmt, args...)
	}
}

// msgf is similar to dbgf but used for per-message debug messages. since these are so numerous there's a separate compile-time flag to compile these out
// in addition, since the go 1.7 compiler isn't eliminating the call sufficiently to avoid allocating the []interface{} needed by "args...interface{}",
// we pass the pointer to the msg. That works and doesn't show up in the heap gc'ed trash profile (pprof -alloc_space)
func msgf(fmt string, msg *sarama.ConsumerMessage) {
	if per_msg_debug {
		Logf(fmt, msg.Topic, msg.Partition, msg.Offset)
	}
}

// minimum kafka API version required. Use this when constructing the sarama.Client's sarama.Config.MinVersion
var MinVersion = sarama.V0_9_0_0

// Error holds the errors generated by this package
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
	cl        *client
}

func (err *Error) Error() string {
	if err.Topic != "" {
		if err.Partition != -1 {
			return fmt.Sprintf("consumer-group %q: Error %s, topic %q, partition %d: %s", err.cl.group_name, err.Context, err.Topic, err.Partition, err.Err)
		}
		return fmt.Sprintf("consumer-group %q: Error %s, topic %q: %s", err.cl.group_name, err.Context, err.Topic, err.Err)
	}
	return fmt.Sprintf("consumer-group %q: Error %s: %s", err.cl.group_name, err.Context, err.Err)
}

// Config is the configuration of a Client. Typically you'd create a default configuration with
// NewConfig, modify any fields of interest, and pass it to NewClient. Once passed to NewClient the
// Config must not be modified. (doing so leads to data races, and may caused bugs as well).
//
// In addition to this config, consumer's code also looks at the sarama.Config of the sarama.Client
// supplied to NewClient, especially at the Consumer.Offsets settings, Version, Metadata.Retry.Backoff,
// Metadata.RefreshFrequency and ChannelBufferSize.
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),
	// and messages older than MaxMessageAge if MaxMessageAge is set.
	// 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

	// InOrderDone disables extra processing which permits Done() to be called out of ordera.
	// If InOrderDone is true then Done() does not have to be be called for every message.
	// The caller can wait and call Done() once to indicate that processing is complete for
	// all messages up to and including the argument message in the argument's partition.
	// This is not the historical behavior, so it is disabled by default.
	InOrderDone bool

	// NoMessages disables fetching and receiving sarama.ConsumerMessage. So the consumer group participation is performed,
	// but fetching the kafka messages is left to the caller's code (presumably using sarama.Broker.Fetch or similar low level API)
	// NoMessages requires InOrderDone, since without seeing the messages ingress the group consumer cannot keep track of which
	// haven't been completed.
	NoMessages bool

	// PartitionStartNotification is an optional callback to inform client code of the (partition,offset) at which we've
	// started consuming (or, if NoMessages, at which we think the caller should start consuming)
	PartitionStartNotification PartitionStartNotification

	// MaxMessageAge is a optional function mapping a topic to the maximum lag the consumer should try and maintain in that topic. If the consumer
	// lags more than MaxMessageAge (as compared with the sarama.ConsumerMessage.Timestamp) it declares an OffsetOutOfRange condition, and restarts
	// where OffsetOutOfRange() indicates. (by default that's OffsetNewest, which might not be what you want, so override OffsetOutOfRange too)
	// Note that this is not absolute, since there's always some lag, and pipelining, and in a low message frequency
	// partition MaxMessageAge might not make sense. Returning 0 disables this functionality.
	MaxMessageAge func(topic string) time.Duration
}

// types of the functions in the Config
type StartingOffset func(topic string, partition int32, committed_offset int64, client sarama.Client) (offset int64, err error)
type OffsetOutOfRange func(topic string, partition int32, client sarama.Client) (offset int64, err error)
type AssignmentNotification func(assignments map[string][]int32)                  // assignments is a map from topic -> list of partitions
type PartitionStartNotification func(topic string, partition int32, offset int64) // position at which we're going to start consuming from the partition

// default implementation of Config.OffsetOutOfRange jumps to the current head of the partition.
func DefaultOffsetOutOfRange(topic string, partition int32, client sarama.Client) (int64, error) {
	return sarama.OffsetNewest, nil
}

// default implementation of Config.StartingOffset starts at the committed offset, or at sarama.Config.Consumer.Offsets.Initial
// if there is no committed offset.
func DefaultStartingOffset(topic string, partition int32, offset int64, client sarama.Client) (int64, error) {
	if offset == sarama.OffsetNewest {
		offset = client.Config().Consumer.Offsets.Initial
	}
	return offset, nil
}

// NewConfig constructs a default configuration.
func NewConfig() *Config {
	cfg := &Config{}
	cfg.Session.Timeout = 30 * time.Second
	cfg.Rebalance.Timeout = 30 * time.Second
	cfg.Heartbeat.Interval = 3 * time.Second
	cfg.Partitioner = roundrobin.RoundRobin
	cfg.OffsetOutOfRange = DefaultOffsetOutOfRange
	cfg.StartingOffset = DefaultStartingOffset
	cfg.SidechannelTopic = "sarama-consumer-sidechannel-offsets"
	return cfg
}

/*
NewClient creates a new consumer group client on top of an existing
sarama.Client.

After this call the contents of config should be treated as read-only.
config can be nil if the defaults are acceptable.

The consumer group name is used to match this client with other
instances running elsewhere, but connected to the same cluster
of kafka brokers and using the same consumer group name.

The supplied sarama.Client should have been constructed with a sarama.Config
where sarama.Config.Version is >= consumer.MinVersion, and if full handling of
ErrOffsetOutOfRange is desired, sarama.Config.Consumer.Return.Errors = true.

In addition, this package uses the settings in sarama.Config.Consumer.Offsets
and sarama.Config.Metadata.RefreshFrequency
*/
func NewClient(group_name string, config *Config, sarama_client sarama.Client) (Client, error) {

	// sanity check
	if config.NoMessages && !config.InOrderDone {
		return nil, fmt.Errorf("invalid sarama-consumer.Config: .NoMessages requires .InOrderDone")
	}

	cl := &client{
		client:     sarama_client,
		config:     config,
		group_name: group_name,

		errors: make(chan error),

		closed:             make(chan struct{}),
		add_consumers:      make(chan add_consumers),
		rem_consumer:       make(chan *consumer),
		sidechannel_commit: make(chan map[string][]SidechannelOffset),
	}

	// start the client's manager goroutine
	rc := make(chan error)
	cl.wg.Add(1)
	go cl.run(rc)

	return cl, <-rc
}

/*
Client is a kafaka client belonging to a consumer group. It is created by NewClient.
*/
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

	// TODO have a Status() method for debug/logging? Or is Errors() enough?
}

/*
Consumer is a consumer of a topic.

Messages from any partition assigned to this client arrive on the
channel returned by Messages.

Every message read from the Messages channel should be eventually passed
to Done, or have its topic/partition/offset passed to MarkUpTo.
Calling Done is the signal that that one message has been consumed (possibly
out of receive order).
*/
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.
	// Calling Done on message out of order is supported, and the consumer keeps
	// track of the correct offset to commit to kafka.
	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()
}

/*
Partitioner maps partitions to consumer group members.

When the user wants control over the partitioning they should set
Config.Partitioner to their implementation of Partitioner.
*/
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)
}

// client implements the Client interface
type client struct {
	client     sarama.Client // the sarama client from which we were constructed
	config     *Config       // our configuration (read-only)
	group_name string        // the client-group name

	errors chan error // channel over which asynchronous errors are reported

	closed chan struct{}  // channel which is closed to cause the client to shutdown
	wg     sync.WaitGroup // waitgroup which is done when the client is shutdown

	add_consumers chan add_consumers // command channel used to add new consumers
	rem_consumer  chan *consumer     // command channel used to remove an existing consumer

	sidechannel_commit chan map[string][]SidechannelOffset // command channel used to commit to the sidechannel
}

// Errors returns the channel over which asynchronous errors are observed.
func (cl *client) Errors() <-chan error { return cl.errors }

// add_consumers are the messages sent over the client.add_consumers channel
type add_consumers struct {
	cons  []*consumer
	reply chan<- error
}

func (cl *client) Consume(topic string) (Consumer, error) {
	sarama_consumer, err := sarama.NewConsumerFromClient(cl.client)
	if err != nil {
		return nil, cl.makeError("Consume sarama.NewConsumerFromClient", err)
	}

	chanbufsize := cl.client.Config().ChannelBufferSize // give ourselves some capacity once I know it runs right without any (capacity hides bugs :-)

	con := &consumer{
		cl:            cl,
		consumer:      sarama_consumer,
		topic:         topic,
		in_order_done: cl.config.InOrderDone,

		messages: make(chan *sarama.ConsumerMessage, chanbufsize),

		closed: make(chan struct{}),
		exited: make(chan struct{}),

		assignments: make(chan *assignment, 1),
		commit_reqs: make(chan commit_req),

		done: make(chan *sarama.ConsumerMessage, chanbufsize),
	}
	if cl.config.MaxMessageAge != nil {
		con.max_age = cl.config.MaxMessageAge(topic)
	}
	if !con.in_order_done {
		con.premessages = make(chan *sarama.ConsumerMessage, chanbufsize)
	}
	if !con.cl.config.NoMessages {
		con.restart_partitions = make(chan *partition)
	}

	reply := make(chan error)
	cl.add_consumers <- add_consumers{[]*consumer{con}, reply}
	err = <-reply
	if err != nil {
		// if an error is returned then it is up to us to close the sarama.Consumer
		_ = sarama_consumer.Close() // we already have an error to return. a 2nd one is too much
		return nil, err
	}
	return con, nil
}

func (cl *client) ConsumeMany(topics []string) ([]Consumer, error) {
	sarama_consumer, err := sarama.NewConsumerFromClient(cl.client)
	if err != nil {
		return nil, cl.makeError("ConsumeMany sarama.NewConsumerFromClient", err)
	}

	chanbufsize := cl.client.Config().ChannelBufferSize // give ourselves some capacity once I know it runs right without any (capacity hides bugs :-)

	consumers := make([]*consumer, len(topics))
	for i, topic := range topics {
		consumers[i] = &consumer{
			cl:            cl,
			consumer:      sarama_consumer,
			topic:         topic,
			in_order_done: cl.config.InOrderDone,

			messages: make(chan *sarama.ConsumerMessage, chanbufsize),

			closed: make(chan struct{}),
			exited: make(chan struct{}),

			assignments: make(chan *assignment, 1),
			commit_reqs: make(chan commit_req),

			restart_partitions: make(chan *partition),
			done:               make(chan *sarama.ConsumerMessage, chanbufsize),
		}
		if cl.config.MaxMessageAge != nil {
			consumers[i].max_age = cl.config.MaxMessageAge(topic)
		}
		if !consumers[i].in_order_done {
			consumers[i].premessages = make(chan *sarama.ConsumerMessage, chanbufsize)
		}
	}

	reply := make(chan error)
	cl.add_consumers <- add_consumers{consumers, reply}
	err = <-reply
	if err != nil {
		// if an error is returned then it is up to us to close the sarama.Consumer
		_ = sarama_consumer.Close() // we already have an error to return. a 2nd one is too much
		return nil, err
	}

	cons := make([]Consumer, len(consumers))
	for i := range consumers {
		cons[i] = consumers[i]
	}
	return cons, nil
}

// Close shutsdown the client and any remaining Consumers.
func (cl *client) Close() {
	// signal to cl.run() that it should exit
	dbgf("Close client of consumer-group %q", cl.group_name)
	close(cl.closed)
	// and wait for the shutdown to be complete
	cl.wg.Wait()
}

// run is a long lived goroutine which manages this client's membership in the consumer group.
func (cl *client) run(early_rc chan<- error) {
	defer cl.wg.Done()

	var member_id string                    // our group member id, assigned to us by kafka when we first make contact
	consumers := make(map[string]*consumer) // map of topic -> consumer
	var assignments map[string][]int32      // nil, or our currently assigned partitions (map of topic -> list of partitions)
	var wg sync.WaitGroup                   // waitgroup used to wait for all consumers to exit

	defer dbgf("consumer-group %q client exiting", cl.group_name)

	// add a consumer
	add := func(add add_consumers) {
		dbgf("client.run add(%d topics)", len(add.cons))

		// first make sure we aren't already consuming any of these topics
		for _, con := range add.cons {
			if _, ok := consumers[con.topic]; ok {
				// topic already is being consumed. the way the standard kafka 0.9 group coordination works you cannot consume twice with the
				// same client. If you want to consume the same topic twice, use two Clients.
				add.reply <- cl.makeError("Consume", fmt.Errorf("topic %q is already being consumed", con.topic))
				return
			}
		}

		// then start them all at once, so that the next time we send a join we contain all these topics
		for _, con := range add.cons {
			dbgf("client.run add(topic %q)", con.topic)
			consumers[con.topic] = con
			wg.Add(1)
			go con.run(&wg)
		}
		add.reply <- nil
	}
	// remove a consumer
	rem := func(con *consumer) {
		dbgf("client.run rem(topic %q)", con.topic)
		existing_con := consumers[con.topic]
		if existing_con == con {
			delete(consumers, con.topic)
			delete(assignments, con.topic) // forget about the topic's partition assignment
		} // else it's some old consumer and we've already removed it
		// and let the consumer shutdown
		close(con.assignments)
		close(con.commit_reqs)
	}
	// shutdown the consumers. waits until they are all stopped. only call once and return afterwards, since it makes assumptions that hold only when it is used like that
	shutdown := func() {
		dbgf("client.run shutdown")
		// shutdown the remaining consumers
		for _, con := range consumers {
			con.AsyncClose()
		}
		// and consume any last rem_consumer messages from them
		go func() {
			wg.Wait()
			close(cl.rem_consumer)
		}()
		for con := range cl.rem_consumer {
			rem(con)
		}
		wg.Wait()
		// and shutdown the errors channel
		close(cl.errors)
	}

	// if enabled, subscribe to the side-channel topic on the appropriate partition
	var sidechannel_queries chan sidechannel_query // nil, or command channel used to request offsets from sidechannel
	if topic := cl.config.SidechannelTopic; topic != "" {
		sidechannel_queries = make(chan sidechannel_query)
		ready := make(chan error)
		go cl.sidechannel_consumer(topic, sidechannel_queries, ready)
		// want and log errors until sidechannel subscription is ready, since we want to capture the sidechannel msgs we will trigger by our join-group request
		for err := range ready {
			err = cl.makeError(fmt.Sprintf("consuming SidechannelTopic %q", topic), err)
			if early_rc != nil {
				early_rc <- err
				return
			}
			cl.deliverError("", err)
		}
	} // else leave sidechannel_queries nil

	// always start the producer, even if it is just a dummy routine that drains and throws away msgs in cl.sidechannel_commit
	cl.wg.Add(1)
	go cl.sidechannel_producer(cl.config.SidechannelTopic)

	// commitToSidechannel trys to send the partition offsets to the SidechannelTopic
	commitToSidechannel := func() {
		dbgf("commitToSidechannel()")

		if cl.config.SidechannelTopic == "" {
			// no side channel is configured
			return
		}

		// gather up the offsets to commit
		var wg sync.WaitGroup
		resp := make(chan commit_resp)
		for _, con := range consumers {
			wg.Add(1)
			con.commit_reqs <- commit_req{resp, &wg}
		}
		go func(resp chan commit_resp, wg *sync.WaitGroup) {
			wg.Wait()
			close(resp)
		}(resp, &wg)

		var offsets = make(map[string][]SidechannelOffset)
		for r := range resp {
			if r.offset != sarama.OffsetNewest && r.offset != sarama.OffsetOldest {
				dbgf("commit (%q, %d, %d)", r.topic, r.partition, r.offset)
				offsets[r.topic] = append(offsets[r.topic], SidechannelOffset{
					Partition: r.partition,
					Offset:    r.offset,
				})
			} // else don't commit these special offsets
		}

		select {
		case cl.sidechannel_commit <- offsets:
			// good
		case <-cl.closed:
			return
		}
	}

	// start the commit timer
	var commit_timer <-chan time.Time
	clconfig := cl.client.Config()
	commit_interval := clconfig.Consumer.Offsets.AutoCommit.Interval
	if commit_interval == 0 && clconfig.Consumer.Offsets.CommitInterval > 0 {
		// use the legacy option
		commit_interval = clconfig.Consumer.Offsets.CommitInterval
	}
	if commit_interval > 0 {
		commit_ticker := time.NewTicker(commit_interval)
		commit_timer = commit_ticker.C
		defer commit_ticker.Stop()
	} // else don't commit periodically (we still commit when closing down)

	pause := false
	refresh := false        // refresh the coordinating broker (after an I/O error or a ErrNotCoordinatorForConsumer)
	reopen := false         // reopen coordinating broker (after an I/O error)
	var coor *sarama.Broker // nil, or coordinating broker

	// loop rejoining the group each time the group reforms
join_loop:
	for {
		if pause {
			delay := cl.client.Config().Metadata.Retry.Backoff // TODO should we increase timeouts?
			dbgf("pausing %v", delay)
			// pause before continuing, so we don't fail continuously too fast
			timeout := time.After(delay)
		pause_loop:
			for {
				select {
				case <-timeout:
					break pause_loop
				case <-cl.closed:
					// shutdown the remaining consumers
					shutdown()
					return
				case a := <-cl.add_consumers:
					add(a)
				case r := <-cl.rem_consumer:
					rem(r)
				case <-commit_timer:
					commitToSidechannel()
				}
			}
			pause = false
		}

		if reopen {
			if coor != nil {
				dbgf("closing and reopening connection to coordinator %d %s", coor.ID(), coor.Addr())
				if ok, err := coor.Connected(); ok {
					err = coor.Close()
					if err != nil {
						cl.deliverError(fmt.Sprintf("Close()ing coordinating broker %d %s", coor.ID(), coor.Addr()), err)
					}
				} else if err != nil {
					// remote the earlier error
					cl.deliverError(fmt.Sprintf("past Open() of coordinating broker %d %s", coor.ID(), coor.Addr()), err)
				}

				err := coor.Open(cl.client.Config())
				if err != nil {
					cl.deliverError(fmt.Sprintf("re-Open()ing coordinating broker %d %s", coor.ID(), coor.Addr()), err)
				}
				// coor.Open() is asynchronous. We'll continue without waiting (without doing an coor.Connected() call)
				// because coor might not even be our coordinator anymore (and might not exist)
			}
			reopen = false

			// after reopening (successfully or not), always refresh the coordinator
			refresh = true
		}

		if refresh {
			dbgf("refreshing coordinating broker")

			// refresh the group coordinator (because sarama caches the result, and the cache must be manually refreshed by us when we decide an invalidate might be needed)
			err := cl.client.RefreshCoordinator(cl.group_name)
			if err != nil {
				err = cl.makeError("refreshing coordinating broker", err)
				if early_rc != nil {
					early_rc <- err
					return
				}
				cl.deliverError("", err)
				pause = true
				continue join_loop
			}
			refresh = false
		}

		// make contact with the kafka broker coordinating this group
		// NOTE: sarama keeps the result cached, so we aren't taking a round trip to the kafka brokers very time
		// (then again we need to manage sarama's cache too)
		var err error
		coor, err = cl.client.Coordinator(cl.group_name)
		if err != nil {
			err = cl.makeError("contacting coordinating broker "+coor.Addr(), err)
			if early_rc != nil {
				early_rc <- err
				return
			}
			cl.deliverError("", err)

			pause = true
			refresh = true
			continue join_loop
		}
		dbgf("Coordinator %v %v", coor.ID(), coor.Addr())

		// make sure we are connected to the broker
		if ok, err := coor.Connected(); !ok {
			err = cl.makeError("connecting coordinating broker "+coor.Addr(), err)
			if early_rc != nil {
				early_rc <- err
				return
			}
			cl.deliverError("", err)

			pause = true
			reopen = true
			continue join_loop
		}

		// join the group
		jreq := &sarama.JoinGroupRequest{
			GroupId:        cl.group_name,
			SessionTimeout: int32(cl.config.Session.Timeout / time.Millisecond),
			MemberId:       member_id,
			ProtocolType:   "consumer", // we implement the standard kafka 0.9 consumer protocol metadata
		}

		num_partitions := make(map[string]int, len(consumers))
		{ // prepare the join request
			var topics = make([]string, 0, len(consumers))
			var current_assignments = make(map[string][]int32, len(consumers))
			for topic := range consumers {
				topics = append(topics, topic)
				if a := assignments[topic]; len(a) != 0 { // omit any topics for which we are not assigned a partition
					current_assignments[topic] = a
				}

				// and keep track of the # of partitions we saw before we joined
				partitions, err := cl.client.Partitions(topic)
				if err != nil {
					cl.deliverError(fmt.Sprintf("looking up partitions of topic %q", topic), err)
				} else {
					num_partitions[topic] = len(partitions)
				}
			}
			logf("consumer %q proposing partitioner %q", cl.group_name, cl.config.Partitioner.Name())
			cl.config.Partitioner.PrepareJoin(jreq, topics, current_assignments)
		}

		// send and wait for join response while still committing to the side channel, since the JoinGroupResponse doesn't arrive until the broker is sure it has gathered them all
		var jresp *sarama.JoinGroupResponse
		done := make(chan struct{})
		go func(jreq *sarama.JoinGroupRequest) {
			dbgf("sending JoinGroupRequest %v", jreq)
			jresp, err = coor.JoinGroup(jreq)
			dbgf("received JoinGroupResponse %v, %v", jresp, err)
			close(done)
		}(jreq)
	wait_for_jresp:
		for {
			select {
			case <-done:
				break wait_for_jresp
			case <-commit_timer:
				commitToSidechannel()
			}
		}
		if err != nil {
			// some I/O error happened; we should reopen and refresh the current coordinator
			reopen = true
		} else if jresp.Err != 0 {
			switch jresp.Err {
			case sarama.ErrNotCoordinatorForConsumer, sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrRebalanceInProgress:
				refresh = true // the broker is no longer the coordinator. we should refresh the current coordinator
			case sarama.ErrUnknownMemberId:
				member_id = "" // the coordinator no longer knows who we are; have it assign us a new member id
			}
			err = jresp.Err
		}
		if err != nil {
			switch err {
			case sarama.ErrRebalanceInProgress:
				// The "error" whenever the kafka consumer group starts a new generation is correct, expected, and normal
				logf("new consumer group %q generation forming (discovered while joining group): %v", cl.group_name, err)
			default:
				err = cl.makeError("joining group", err)
				// if it is still early (the 1st iteration of this loop) then return the error and bail out
				if early_rc != nil {
					early_rc <- err
					return
				}
				cl.deliverError("", err)
			}

			pause = true
			continue join_loop
		}

		// we managed to get a successfull join-group response. that is far enough that basic communication is functioning
		// and we can declare that our early_rc is success and release the caller to NewClient
		if early_rc != nil {
			early_rc <- nil
			early_rc = nil
		}

		// save our member_id for next time we join, and the new generation id
		member_id = jresp.MemberId
		generation_id := jresp.GenerationId
		logf("consumer %q joining generation %d as member %q", cl.group_name, generation_id, member_id)

		// prepare a sync request
		sreq := &sarama.SyncGroupRequest{
			GroupId:      cl.group_name,
			GenerationId: generation_id,
			MemberId:     member_id,
		}

		// we have been chosen as the leader then we have to map the partitions
		if jresp.LeaderId == member_id {
			dbgf("leader is we; partitioning using partitioner %s", cl.config.Partitioner.Name())
			err := cl.config.Partitioner.Partition(sreq, jresp, cl.client)
			if err != nil {
				cl.deliverError("partitioning", err)
				// and rejoin (thus aborting this generation) since we can't partition it as needed
				pause = true
				continue join_loop
			}
		}

		// send SyncGroup
		var sresp *sarama.SyncGroupResponse
		done = make(chan struct{})
		go func(sreq *sarama.SyncGroupRequest) {
			dbgf("sending SyncGroupRequest %v", sreq)
			sresp, err = coor.SyncGroup(sreq)
			dbgf("received SyncGroupResponse %v, %v", sresp, err)
			close(done)
		}(sreq)
	wait_for_sresp:
		for {
			select {
			case <-done:
				break wait_for_sresp
			case <-commit_timer:
				commitToSidechannel()
			}
		}
		if err != nil {
			reopen = true
		} else if sresp.Err != 0 {
			switch sresp.Err {
			case sarama.ErrNotCoordinatorForConsumer, sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrRebalanceInProgress:
				refresh = true // the broker is no longer the coordinator. we should refresh the current coordinator
			case sarama.ErrUnknownMemberId:
				member_id = "" // the coordinator no longer knows who we are; have it assign us a new member id
			}
			err = sresp.Err
		}
		if err != nil {
			switch err {
			case sarama.ErrRebalanceInProgress:
				// The "error" whenever the kafka consumer group starts a new generation is correct, expected, and normal
				logf("new consumer group %q generation forming (discovered while synchronizing group): %v", cl.group_name, err)
			default:
				cl.deliverError("synchronizing group", err)
			}
			pause = true
			continue join_loop
		}
		new_assignments, err := cl.config.Partitioner.ParseSync(sresp)
		if err != nil {
			cl.deliverError("decoding member assignments", err)
			pause = true
			continue join_loop
		}

		// keep track of which and how many partitions we are assigned
		assignments = new_assignments
		num_assigned_partitions := 0
		for _, parts := range assignments {
			num_assigned_partitions += len(parts)
		}
		logf("consumer %q assigned %d partitions; assignment: %v", cl.group_name, num_assigned_partitions, assignments)
		if cl.config.AssignmentNotification != nil {
			// keep users from thinking they can alter assignments in the callback by making a deep copy
			acopy := make(map[string][]int32, len(assignments))
			n := 0
			for _, v := range assignments {
				n += len(v)
			}
			parts := make([]int32, 0, n) // space for all the partitions slices (better for gc and for perf)
			for k, v := range assignments {
				n := len(parts)
				parts := append(parts, v...)
				acopy[k] = parts[n:len(parts):len(parts)]
			}
			cl.config.AssignmentNotification(acopy)
		}

		// save and distribute the new assignments to our topic consumers
		a := &assignment{
			generation_id:       generation_id,
			coordinator:         coor,
			member_id:           member_id,
			assignments:         assignments,
			sidechannel_queries: sidechannel_queries,
		}
		for _, con := range consumers {
			// con.assignments has a capacity of 1. the chan is either empty, or contains a stale assignment we can remove and replace
			select {
			case con.assignments <- a:
				// a is delivered
			case <-con.assignments:
				// we've cleared out the stale assignment. since we've the only code which writes to this channel we now know we have room
				con.assignments <- a
			}
		}

		// start the heartbeat timer
		heartbeat_timer := time.After(cl.config.Heartbeat.Interval)
		// and the metadata check timer
		var metadata_timer <-chan time.Time
		if clconfig.Metadata.RefreshFrequency > 0 {
			metadata_timer = time.After(clconfig.Metadata.RefreshFrequency)
		}

		// and loop, sending heartbeats until something happens and we need to rejoin (or exit)
		for {
			select {
			case <-cl.closed:
				// cl.Close() has been called; time to exit

				// shutdown any remaining consumers (causing them to sync their final offsets)
				shutdown()

				// and nicely leave the consumer group
				req := &sarama.LeaveGroupRequest{
					GroupId:  cl.group_name,
					MemberId: member_id,
				}
				dbgf("sending LeaveGroupRequest %v", req)
				resp, err := coor.LeaveGroup(req)
				dbgf("received LeaveGroupResponse %v, %v", resp, err)
				// note: we don't bother with the full error handling code, since we're exiting anyway
				if err == nil && resp.Err != 0 {
					err = resp.Err
				}
				if err != nil {
					cl.deliverError("leaving group", err)
				}

				// and we're done
				return

			case <-heartbeat_timer:
				// send a heartbeat
				req := &sarama.HeartbeatRequest{
					GroupId:      cl.group_name,
					MemberId:     member_id,
					GenerationId: generation_id,
				}
				dbgf("sending HeartbeatRequest %v", req)
				resp, err := coor.Heartbeat(req)
				dbgf("received HeartbeatResponse %v, %v", resp, err)
				if err != nil {
					reopen = true
				} else if resp.Err != 0 {
					switch resp.Err {
					case sarama.ErrNotCoordinatorForConsumer, sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrRebalanceInProgress:
						refresh = true // the broker is no longer the coordinator. we should refresh the current coordinator
					case sarama.ErrUnknownMemberId:
						member_id = "" // the coordinator no longer knows who we are; have it assign us a new member id
					}
					err = resp.Err
				}
				if err != nil {
					switch err {
					case sarama.ErrRebalanceInProgress, sarama.ErrIllegalGeneration:
						// The "error" whenever the kafka consumer group starts a new generation is correct, expected, and normal
						logf("consumer group %q at %v is rebalancing: %v; rejoining new generation", cl.group_name, coor.Addr(), err)
					default:
						cl.deliverError("heartbeating with "+coor.Addr(), err)
					}
					// we've got heartbeat troubles of one kind or another; disconnect and reconnect
					continue join_loop
				}

				// and start the next heartbeat only after we get the response to this one
				// that way when the network or the broker are slow we back off.
				heartbeat_timer = time.After(cl.config.Heartbeat.Interval)

			case <-commit_timer:
				ocreq := &sarama.OffsetCommitRequest{
					ConsumerGroup:           cl.group_name,
					ConsumerGroupGeneration: generation_id,
					ConsumerID:              member_id,
					RetentionTime:           int64(clconfig.Consumer.Offsets.Retention / time.Millisecond),
					Version:                 2, // kafka 0.9.0 version, with RetentionTime
				}
				if clconfig.Consumer.Offsets.Retention == 0 { // note that this and the rounding math above means that if you wanted a retention time of 0 millseconds you could set Config.Offsets.RetentionTime to something < 1 ms, like 1 nanosecond
					ocreq.RetentionTime = -1 // use broker's value
				}
				var wg sync.WaitGroup
				resp := make(chan commit_resp, num_assigned_partitions) // allocating room for the responses helps the code run smoothly
				for _, con := range consumers {
					wg.Add(1)
					con.commit_reqs <- commit_req{resp, &wg}
				}
				go func(resp chan commit_resp, wg *sync.WaitGroup) {
					wg.Wait()
					close(resp)
				}(resp, &wg)
				empty := true
				for r := range resp {
					dbgf("ocreq.AddBlock(%q, %d, %d)", r.topic, r.partition, r.offset)
					ocreq.AddBlock(r.topic, r.partition, r.offset, -1, 0, "")
					empty = false
				}
				if empty {
					// no point in sending an empty commit message
					break
				}
				dbgf("sending OffsetCommitRequest %v", ocreq)
				ocresp, err := coor.CommitOffset(ocreq)
				dbgf("received OffsetCommitResponse %v, %v", ocresp, err)
				// log any errors we got. there isn't much we can do about them
				try_sidechannel := false
				if err != nil {
					cl.deliverError("committing offsets to "+coor.Addr(), err)
					reopen = true
					try_sidechannel = true
				} else {
					var prev_kerr sarama.KError // don't print the same error over and over. usually the same error will happen to all partitions
					for topic, partitions := range ocresp.Errors {
						for p, kerr := range partitions {
							if kerr != 0 {
								if kerr != prev_kerr {
									switch kerr {
									case sarama.ErrRebalanceInProgress, sarama.ErrIllegalGeneration:
										// The "error" whenever the kafka consumer group starts a new generation is correct, expected, and normal
										logf("new consumer group %q generation forming (discovered while committing offset of topic %q partition %d): %v; will publish to side-channel instead", cl.group_name, topic, p, kerr)
									default:
										cl.deliverError(fmt.Sprintf("committing offset of topic %q partition %d", topic, p), kerr)
									}
									prev_kerr = kerr
								} else {
									dbgf("same error committing offset of topic %q partition %d", topic, p, kerr)
								}
								switch kerr {
								case sarama.ErrNotCoordinatorForConsumer, sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrRebalanceInProgress:
									refresh = true         // the broker is no longer the coordinator. we should refresh the current coordinator
									try_sidechannel = true // and send the commits to the side-channel (if we have one) in the hope that that might work
								case sarama.ErrUnknownMemberId:
									member_id = "" // the coordinator no longer knows who we are; have it assign us a new member id
								case sarama.ErrIllegalGeneration:
									try_sidechannel = true // a new generation is forming; send our offsets to the sidechannel
								}
								err = kerr // any of the kerr's will do
							}
						}
					}
				}
				if try_sidechannel {
					// immediately send a commit to the side channel
					commitToSidechannel()
				}
				if err != nil {
					continue join_loop
				}

			case <-metadata_timer:
				dbgf("metadata timer")
				// the sarama.Client has refreshed its metadata within the interval
				// all we do is verify the number of partitions hasn't changed since we joined a topic
				// this is a local calculation, so no need for any fancy concurrency
				for topic := range consumers {
					partitions, err := cl.client.Partitions(topic)
					if err != nil {
						cl.deliverError(fmt.Sprintf("looking up the partitions of topic %q", topic), err)
						// and rejoin the groups
						continue join_loop
					}
					if len(partitions) != num_partitions[topic] {
						dbgf("num_partitions of topic %q changed from %d to %d; rejoining", topic, num_partitions[topic], len(partitions))
						// rejoin the new partition count (presumably some new partitions have been added, since you can't remove partitions from a running kafka broker)
						continue join_loop
					}
				}

				// this drifts slightly. is that good enough for this use case or must I use a time.Ticker? the worst that happens is an interval is skipped. That is ok, we'll
				// pick up the change in the next interval.
				metadata_timer = time.After(clconfig.Metadata.RefreshFrequency)

			case a := <-cl.add_consumers:
				add(a)
				// and rejoin so we can become a member of the new topic
				continue join_loop
			case r := <-cl.rem_consumer:
				rem(r)
				// and rejoin so we can be removed as member of the new topic
				continue join_loop
			}
		} // end of heartbeat loop
	} // end of join_loop
}

type sidechannel_key struct {
	topic     string
	partition int32
}

// msg sent to sidechannel goroutine to ask what it knows about a set of <topic,partition> pairs
type sidechannel_query struct {
	reply   chan<- sidechannel_offset
	queries []sidechannel_key
}

// msg send back in response to a sidechann_query
type sidechannel_offset struct {
	sidechannel_key
	offset int64
}

// consume from the sidechannel and keep a local table of what is learned. since cl.group_key is used to select
// the partition to which all members of this consumer group commit, we only need to consume from that one topic
func (cl *client) sidechannel_consumer(topic string, queries <-chan sidechannel_query, ready chan<- error) {
	dbgf("sidechannel_consumer(%q)", topic)
	defer dbgf("sidechannel_consumer(%q) exiting", topic)
	// local table of {topic,partition} -> offset
	offsets := make(map[sidechannel_key]int64)
	// our kafka msg key
	our_key := []byte(cl.group_name)

	// deliver errors to ready when we are starting up, or to cl.Errors later on
	deliverError := func(msg string, err error) {
		err = cl.makeError(msg, err)
		if ready != nil {
			select {
			case ready <- err:
			case <-cl.closed:
			}
		} else {
			cl.deliverError("", err)
		}
	}

	// consume from the appropriate partition of our side channel
	start := func() (sarama.Consumer, sarama.PartitionConsumer) { // func just to have a nice clean way to return out of error conditions
		dbgf("sidechannel start")
		if topic == "" {
			dbgf("sidechannel no topic (disabled)")
			return nil, nil
		}
		sconsumer, err := sarama.NewConsumerFromClient(cl.client)
		if err != nil {
			deliverError("creating sarama consumer", err)
			return nil, nil
		}
		dbgf("sidechannel got consumer")
		parts, err := sconsumer.Partitions(topic)
		if err != nil {
			deliverError("listing partitions of side-channel topic "+topic, err)
			return sconsumer, nil
		}
		dbgf("sidechannel got partitions %v", parts)
		var partition int32
		if len(parts) == 1 {
			partition = parts[0] // no choice
		} else {
			// what does the partitioner do with our consumer group name?
			parter := cl.client.Config().Producer.Partitioner(topic)
			dbgf("sidechannel got partitioner %v", parter)
			test_msg := &sarama.ProducerMessage{
				Topic: topic,
				Key:   sarama.StringEncoder(cl.group_name),
				Value: sarama.ByteEncoder(nil),
			}
			p1, err := parter.Partition(test_msg, int32(len(parts)))
			if err != nil {
				deliverError("can't pick a partition in side-channel topic "+topic, err)
				return sconsumer, nil
			}
			// try it again, to flush out round-robin style partitioners used by accident
			p2, err := parter.Partition(test_msg, int32(len(parts)))
			if err != nil {
				deliverError("can't pick a partition in side-channel topic "+topic, err)
				return sconsumer, nil
			}
			dbgf("sidechannel got partition responses %v, %v", p1, p2)
			if p1 != p2 {
				deliverError("", fmt.Errorf("partitioning is inconsistent in side-channel topic %q", topic))
				return sconsumer, nil
			}
			partition = parts[p1]
		}
		logf("consumer %q sidechannel consuming %q partition %d", cl.group_name, topic, partition)

		pconsumer, err := sconsumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
		if err != nil {
			deliverError(fmt.Sprintf("consuming side-channel partition %d of %s", partition, topic), err)
			return sconsumer, nil
		}
		dbgf("sidechannel got partition consumer %v", pconsumer)
		return sconsumer, pconsumer
	}

	var sconsumer sarama.Consumer
	var pconsumer sarama.PartitionConsumer
	// make sure we close things on the way out
	defer func() {
		if pconsumer != nil {
			pconsumer.Close()
		}
		if sconsumer != nil {
			sconsumer.Close()
		}
	}()

	for {
		sconsumer, pconsumer = start()
		var msgs <-chan *sarama.ConsumerMessage
		var errors <-chan *sarama.ConsumerError

		var retry <-chan time.Time // nil, or timer indicating when to attempt to reconnect to kafka
		if pconsumer == nil {
			if sconsumer != nil {
				sconsumer.Close()
				sconsumer = nil
			}
			if topic != "" {
				// try again after a short pause
				retry = time.After(cl.client.Config().Consumer.Retry.Backoff)
			} // else sidechannel use is disabled and we're just going to stuck around to return any requests without any responses
		}

		if pconsumer != nil {
			msgs = pconsumer.Messages()
			errors = pconsumer.Errors()
		}

		if ready != nil {
			close(ready)
			ready = nil
		}

	msg_loop:
		for {
			select {
			case <-retry:
				// retry connecting to kafka
				break msg_loop

			case <-cl.closed:
				dbgf("sidechannel client closed")
				// consumer is shutting down; let the defers do the cleanup
				return

			case err, ok := <-errors:
				dbgf("sidechannel consumer error %v, %v", err, ok)
				if !ok {
					errors = nil
					break
				}
				cl.deliverError("consuming side-channel topic "+topic, err)
				// disconnect and reconnect to broker
				break msg_loop

			case kmsg, ok := <-msgs:
				//dbgf("sidechannel msg %v, %v", kmsg, ok)
				if !ok {
					msgs = nil
					break
				}
				if !bytes.Equal(kmsg.Key, our_key) {
					// ignore the msg; it belongs to a different consumer group
					//dbgf("sidechannel ignore; key %q != %q", kmsg.Key, our_key)
					continue msg_loop
				}
				// decode the msg
				var msg SidechannelMsg
				err := json.Unmarshal(kmsg.Value, &msg)
				if err != nil {
					cl.deliverError("unmarshaling  side-channel msg", err)
					continue msg_loop
				}
				//dbgf("sidechannel msg %v", msg)
				if msg.Ver != 1 {
					cl.deliverError("", fmt.Errorf("unknown SidechannelMsg version %d", msg.Ver))
					continue msg_loop
				}

				// save/update the offsets in our local table
				for topic, topic_offsets := range msg.Offsets {
					for i := range topic_offsets {
						to := &topic_offsets[i]
						// record any new, gt offsets
						key := sidechannel_key{topic, to.Partition}
						o := offsets[key] // note zero-value works out fine
						if to.Offset > o {
							//dbgf("sidechannel noting %+v > %d", to, o)
							offsets[key] = to.Offset
						} else {
							//dbgf("sidechannel ignoring %+v <= %d", to, o)
						}
					}
				}

			case q := <-queries:
				//dbgf("sidechannel query %+v", q)
				// look through the request and send back any we have, then close the reply channel to indicate we're done
				for _, key := range q.queries {
					o, ok := offsets[key]
					if ok {
						//dbgf("sidechannel replying %+v", sidechannel_offset{key, o})
						q.reply <- sidechannel_offset{key, o}
					}
				}
				close(q.reply)
			}
		}

		// drain and throw away whatever is left in current consumer's errors and msgs
		if errors != nil {
			go func(errors <-chan *sarama.ConsumerError) {
				for err := range errors {
					dbgf("tossing stale side-channel error %v", err)
				}
			}(errors)
		}
		if msgs != nil {
			go func(msgs <-chan *sarama.ConsumerMessage) {
				for range msgs {
					// toss (too bad)
				}
			}(msgs)
		}

		if pconsumer != nil {
			pconsumer.Close()
			pconsumer = nil
		}
		if sconsumer != nil {
			sconsumer.Close()
			sconsumer = nil
		}
	}
}

// produce to the sidechannel partition when asked
// if topic == "" then silently throw away any requests to produce
func (cl *client) sidechannel_producer(topic string) {
	dbgf("sidechannel_producer(%q)", topic)
	defer dbgf("sidechannel_producer(%q) exiting", topic)
	defer cl.wg.Done()

	our_key := []byte(cl.group_name)

	var producer sarama.AsyncProducer
	defer func() {
		if producer != nil {
			producer.Close()
		}
	}()

	var perrors <-chan *sarama.ProducerError
	defer func() {
		if perrors != nil {
			// drain any and all remaining errors
			go func(perrors <-chan *sarama.ProducerError) {
				for range perrors {
					// toss error
				}
			}(perrors)
		}
	}()

	// send offsets to the sidechannel (as a function since returning from
	// error conditions makes the code easier to read than breaking out of case statements)
	send := func(offsets map[string][]SidechannelOffset) {
		if len(offsets) == 0 {
			// no offsets to commit
			return
		}
		if topic == "" {
			// no side-channel is configured
			return
		}

		var err error
		if producer == nil {
			// create the side-channel producer now
			producer, err = sarama.NewAsyncProducerFromClient(cl.client)
			if err != nil {
				cl.deliverError("creating side-channel producer", err)
				producer = nil // paranoia
				return
			}
			logf("consumer %q sidechannel producing to %q", cl.group_name, topic)

			// start consuming any async errors
			perrors = producer.Errors()

			// and if they are enabled, drain and throw away successes
			if cl.client.Config().Producer.Return.Successes {
				// start draining successes from this producer until it closes
				go func(producer sarama.AsyncProducer) {
					for range producer.Successes() {
						// toss
					}
				}(producer)
			}
		}

		// construct the message
		var msg = SidechannelMsg{
			Ver:           1,
			ConsumerGroup: cl.group_name,
			Offsets:       offsets,
		}

		// TODO should we use JSON or some faster/smaller encoding like Gob or protobuf? JSON for now until it is clear it is a problem.
		data, err := json.Marshal(msg)
		if err != nil {
			cl.deliverError("marshaling SidechannelMsg", err)
			return
		}

		dbgf("sending offsets to side-channel topic %q", cl.config.SidechannelTopic)
		select {
		case producer.Input() <- &sarama.ProducerMessage{
			Topic: cl.config.SidechannelTopic,
			Key:   sarama.ByteEncoder(our_key),
			Value: sarama.ByteEncoder(data),
		}:
		case <-cl.closed:
			return
		}
	}

	for {
		select {
		case <-cl.closed:
			dbgf("sidechannel producer closed")
			// consumer is shutting down; let the defers do the cleanup
			return

		case offsets := <-cl.sidechannel_commit:
			send(offsets)

		case err, ok := <-perrors:
			dbgf("sidechannel producer error %v, %v", err, ok)
			if !ok {
				perrors = nil
				break
			}
			cl.deliverError("producing to side-channel topic "+topic, err)
			// close the producer; we'll reopen it when we need it again
			if producer != nil {
				producer.Close()
				producer = nil
				perrors = nil
			}
		}
	}
}

// makeError wraps err into a *Error, associating it with context
func (cl *client) makeError(context string, err error) *Error {
	return &Error{
		cl:        cl,
		Err:       err,
		Context:   context,
		Topic:     "",
		Partition: -1,
	}
}

// deliverError builds an error and delivers it to the channel returned by cl.Errors
func (cl *client) deliverError(context string, err error) {
	if context != "" {
		err = cl.makeError(context, err)
	}
	logf("%v", err)
	select {
	case cl.errors <- err:
	case <-cl.closed:
	}
}

// consumer implements the Consumer interface
type consumer struct {
	cl            *client
	consumer      sarama.Consumer
	topic         string
	in_order_done bool // if true then calling Done() marks all messages up to and including the argument as done.
	// if false then Done() must be called for each message, but need not be called in message receive order.
	max_age time.Duration // 0, or maximum message age before we call OffsetOutOfRange() and [presumably] skip ahead

	messages chan *sarama.ConsumerMessage

	closed     chan struct{} // channel which is closed when the consumer is AsyncClose()ed
	close_once sync.Once     // Once used to make sure we close only once
	exited     chan struct{} // channel which is closed when the consumer is far enough along in exiting that consumer.Close can return

	assignments chan *assignment // channel over which client.run sends consumer.run each generation's partition assignments
	commit_reqs chan commit_req  // channel over which client.run sends consumer.run request to fill out a OffsetCommitRequest

	restart_partitions chan *partition              // channel through which partition.run delivers partition restart [at new offset] requests if !Config.NoMessages. nil otherwise
	premessages        chan *sarama.ConsumerMessage // channel through which partition.run delivers messages to consumer.run if !in_order_done. nil otherwise
	done               chan *sarama.ConsumerMessage // channel through which Done() returns messages
}

// commit_req is a request for a consumer to send back the client its part into a OffsetCommitRequest
type commit_req struct {
	resp chan<- commit_resp
	wg   *sync.WaitGroup
}

type commit_resp struct {
	topic     string
	partition int32
	offset    int64
}

// SidechannelMsg is what is published to and read from the Config.SidechannelTopic
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
}

// SidechannelOffset contains the offset a single partition
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
	// if someday we make use of the timestamp and metadata fields we'd add them here
}

// assignment is this client's assigned partitions
type assignment struct {
	generation_id       int32                    // the current generation
	coordinator         *sarama.Broker           // the current client-group coordinating broker
	member_id           string                   // the member_id assigned to us by the coordinator
	assignments         map[string][]int32       // map of topic -> list of partitions
	sidechannel_queries chan<- sidechannel_query // nil, or a channel over which consumers can ask about sidechannel information
}

// construct a *Error from this consumer
func (con *consumer) makeError(context string, err error) *Error {
	Err := con.cl.makeError(context, err)
	Err.Consumer = con
	Err.Topic = con.topic
	return Err
}

// construct and deliver an *Error from this consumer
func (con *consumer) deliverError(context string, partition int32, err error) {
	Err := con.makeError(context, err)
	Err.Partition = partition
	con.cl.deliverError("", Err)
}

func (con *consumer) Messages() <-chan *sarama.ConsumerMessage { return con.messages }

// close the consumer. it can safely be called multiple times
func (con *consumer) AsyncClose() {
	dbgf("AsyncClose consumer of topic %q", con.topic)
	con.close_once.Do(func() { close(con.closed) })
}

// close the consumer and wait
func (con *consumer) Close() {
	dbgf("Close consumer of topic %q", con.topic)
	con.AsyncClose() // initiate the shutdown
	<-con.exited     // and wait around until it is complete
}

// consumer goroutine coordinates consuming from multiple partitions in a topic
// NOTE WELL: this function must never do anything which would prevent it from processing message from client.run promptly.
// That means any channel I/O must include cases for con.assignments and con.commit_reqs.
func (con *consumer) run(wg *sync.WaitGroup) {
	var generation_id int32 // current generation
	var coor *sarama.Broker // current consumer group coordinating broker
	var member_id string    // our member id assigned by coor

	partitions := make(map[int32]*partition) // map of partition number -> partition consumer

	// shutdown the removed partitions, committing their last offset
	remove := func(removed []int32) {
		dbgf("consumer %q rem(%v)", con.topic, removed)
		if len(removed) == 0 {
			// nothing to do, and no point in sending an empty OffsetCommitRequest msg either
			return
		}
		clconfig := con.cl.client.Config()
		ocreq := &sarama.OffsetCommitRequest{
			ConsumerGroup:           con.cl.group_name,
			ConsumerGroupGeneration: generation_id,
			ConsumerID:              member_id,
			RetentionTime:           int64(clconfig.Consumer.Offsets.Retention / time.Millisecond),
			Version:                 2, // kafka 0.9.0 version, with RetentionTime
		}
		if clconfig.Consumer.Offsets.Retention == 0 { // note that this and the rounding math above means that if you wanted a retention time of 0 millseconds you could set Config.Offsets.RetentionTime to something < 1 ms, like 1 nanosecond
			ocreq.RetentionTime = -1 // use broker's value
		}
		var sidechannel_offsets = make([]SidechannelOffset, 0, len(removed))
		for _, p := range removed {
			// stop consuming from partition p
			if part, ok := partitions[p]; ok {
				delete(partitions, p)
				if part.consumer != nil {
					part.consumer.Close()
				}
				offset := part.compute_commit_offset()
				if offset == sarama.OffsetNewest || offset == sarama.OffsetOldest {
					continue // omit this partition, we don't have a proper offset for this partition b/c we have not yet received any msgs on this partition yet
				}
				dbgf("ocreq.AddBlock(%q, %d, %d)", con.topic, p, offset)
				ocreq.AddBlock(con.topic, p, offset, -1, 0, "")
				sidechannel_offsets = append(sidechannel_offsets, SidechannelOffset{p, offset})
				logf("consumer %q stopped consuming %q partition %d at offset %d", con.cl.group_name, con.topic, p, offset)
			}
		}
		dbgf("sending OffsetCommitRequest %v", ocreq)
		ocresp, err := coor.CommitOffset(ocreq)
		dbgf("received OffsetCommitResponse %v, %v", ocresp, err)
		// log any errors we got. there isn't much we can do about them; the next consumer will start at an older offset
		try_sidechannel := false
		if err != nil {
			con.deliverError("committing offsets", -1, err)
			try_sidechannel = true
		} else {
			var prev_kerr sarama.KError // don't print the same error over and over. usually the same error will happen to all partitions
			for _, partitions := range ocresp.Errors {
				for p, kerr := range partitions {
					if kerr != 0 {
						if kerr != prev_kerr {
							switch kerr {
							case sarama.ErrRebalanceInProgress, sarama.ErrIllegalGeneration:
								// The "error" whenever the kafka consumer group starts a new generation is correct, expected, and normal
								logf("new consumer group %q generation forming (discovered while committing offset of topic %q partition %d): %v; will publish to side-channel instead", con.cl.group_name, con.topic, p, kerr)
							default:
								con.deliverError("committing offset", p, kerr)
							}
							prev_kerr = kerr
						} else {
							dbgf("same error committing offset of topic %q partition %d", con.topic, p, kerr)
						}
						switch kerr {
						case sarama.ErrIllegalGeneration, sarama.ErrNotCoordinatorForConsumer, sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrRebalanceInProgress:
							try_sidechannel = true
						}
					}
				}
			}
		}
		if try_sidechannel {
			select {
			case con.cl.sidechannel_commit <- map[string][]SidechannelOffset{con.topic: sidechannel_offsets}:
			case <-con.cl.closed:
			}
		}
	}

	// handle a commit request from client.run
	commit_req := func(c commit_req) {
		dbgf("consumer %q commit_req(%v)", con.topic, c)
		for p, partition := range partitions {
			c.resp <- commit_resp{topic: con.topic, partition: p, offset: partition.compute_commit_offset()}
		}
		c.wg.Done()
	}

	defer func() {
		if len(partitions) != 0 {
			// cleanup the remaining partition consumers
			removed := make([]int32, 0, len(partitions))
			for p := range partitions {
				removed = append(removed, p)
			}
			remove(removed)
		}

		con.consumer.Close()
		close(con.messages)

		// send ourselves to rem_consumer
	rem_loop:
		for {
			select {
			case c := <-con.commit_reqs:
				commit_req(c)
			case <-con.assignments:
				// ignore them, we're shutting down
			case con.cl.rem_consumer <- con:
				break rem_loop
			}
			// NOTE: <-con.done is not a case above because there is no good way to shut it down. it is never closed,
			// so we'd never know when it was drained. As a consequence, it's client.Done() which aborts when the
			// consumer closes, rather than us draining con.done
		}

		// drain any remaining requests from run.client, until
		// run.client closes the channels
		for c := range con.commit_reqs {
			commit_req(c)
		}
		for range con.assignments {
			// ignore them
		}

		dbgf("consumer of topic %q exiting", con.topic)
		close(con.exited)
		wg.Done()
	}()

	// handle a message sent to us via con.done
	done := func(msg *sarama.ConsumerMessage) {
		if msg.Topic == "" { // a blank topic can happen when the caller faked the ConsumerMessage and doesn't set .Topic. It's better to have a topic for logging purposes, so fill it in
			msg.Topic = con.topic
		}

		msgf("consumer done(%q:%d/%d)", msg)

		// a sanity check, just in case someone passes the msg into the wrong consumer
		if con.topic != msg.Topic {
			con.deliverError("Done()", -1, fmt.Errorf("BUG: Message from topic %q passed to consumer(%q).Done()", msg.Topic, con.topic))
			return
		}

		part := partitions[msg.Partition]
		if part == nil {
			dbgf("no partition %d in topic %q", msg.Partition, con.topic)
			return
		}

		if con.in_order_done {
			// if this advances the commit offset, then record it. otherwise ignore it
			if part.next_commit_offset <= msg.Offset {
				part.next_commit_offset = msg.Offset + 1
			}
		} else {
			// keep track of exactly which offsets have been committed
			delta := msg.Offset - part.next_commit_offset
			if delta < 0 {
				dbgf("stale message %q:%d/%d", msg.Topic, msg.Partition, msg.Offset)
				return
			}
			index := int(delta) >> lg2_offsets_per_bucket
			if index >= len(part.buckets) {
				dbgf("early message %d/%d", msg.Partition, msg.Offset)
				return
			}
			part.buckets[index].done++
			if index == 0 {
				// we might be able to advance the bucket 0 highwater mark
				if part.buckets[0].read == part.buckets[0].done {
					// we know, since messages a read in offset order, that the range of offsets from the start
					// of the bucket to .done is completely Done() and can be committed. (this is useful when the
					// traffic rate is low or a client shuts down cleanly, since in these cases there is a good
					// chance there are no outstanding offsets in the pipelines)
					part.bucket_0_highwater = part.buckets[0].done
				}
				// we might have finished the oldest bucket, and any later waiting, completed buckets
				for part.buckets[0].done == offsets_per_bucket {
					// the oldest bucket is complete; bump the last committed offset and advance to the next bucket
					part.next_commit_offset += offsets_per_bucket
					part.bucket_0_highwater = 0
					part.buckets = part.buckets[1:]
					if len(part.buckets) == 0 {
						break
					}
				}
			}
		}
	}

	// handle an assignment message
	assignment := func(a *assignment) {
		dbgf("consumer %q assignment(%v)", con.topic, a)
		// see what has changed in the partition assignment of our topic
		new_partitions := a.assignments[con.topic]
		added, removed := difference(partitions, new_partitions)
		dbgf("consumer %q added %v, removed %v", con.topic, added, removed)

		// shutdown the partitions while we still belong to the previous generation
		remove(removed)

		// update the current generation and related info after committing the last offsets from the previous generation
		generation_id = a.generation_id
		coor = a.coordinator
		member_id = a.member_id

		if len(added) == 0 {
			// we're done early
			return
		}

		// the sarama-cluster code pauses here so that other consumers have time to sync their offsets. Should we do the same?
		// I've observed with kafka 0.9.0.1 that once the coordinator bumps the generation_id the client can't commit an offset with
		// the old id. So unless the client lies and sends generation_id+1 when it commits there is nothing it can commit, and there
		// is no point in waiting. So for now, no waiting.

		// fetch the last committed offsets of the new partitions from sarama and, if available, from our side-channel consumer

		oreq := &sarama.OffsetFetchRequest{
			ConsumerGroup: con.cl.group_name,
			Version:       1, // kafka 0.9.0 expects version 1 offset requests
		}
		queries := make([]sidechannel_key, len(added))
		for i, p := range added {
			oreq.AddPartition(con.topic, p)
			queries[i].topic = con.topic
			queries[i].partition = p
		}

		sidechannel_replies := make(chan sidechannel_offset, len(queries))
		if a.sidechannel_queries != nil {
			dbgf("asked sidechannel what it knows")
			// send the request async, just in case the sidechannel consumer is busy (which it might be if we are in the middle of a rebalance)
			go func(c chan<- sidechannel_query, q sidechannel_query) {
				c <- q
			}(a.sidechannel_queries, sidechannel_query{
				reply:   sidechannel_replies,
				queries: queries,
			})
		} else {
			close(sidechannel_replies)
			queries = nil
		}

		dbgf("consumer %q of %q sending OffsetFetchRequest %v", con.cl.group_name, con.topic, oreq)
		oresp, err := a.coordinator.FetchOffset(oreq)
		dbgf("consumer %q of %q received OffsetFetchResponse %v, %v", con.cl.group_name, con.topic, oresp, err)
		if err != nil {
			con.deliverError("fetching offsets", -1, err)
			// and we can't consume any of the new partitions without the offsets
			return
		}

		// merge any sidechannel results into the sarama results
		for r := range sidechannel_replies {
			b := oresp.GetBlock(r.topic, r.partition)
			dbgf("sidechannel says %+v, broker said %v", r, b)
			if b == nil || b.Offset < r.offset {
				// add/replace the result
				oresp.AddBlock(r.topic, r.partition, &sarama.OffsetFetchResponseBlock{
					Offset: r.offset,
					// Metadata goes here, if we ever need it
				})
			}
		}

		// start consuming from the added partitions at each partition's last committed offset (which by convention kafaka defines as the last consumed offset+1)
		// since computing the starting offset and beginning to consume requires several round trips to the kafka brokers we start all the
		// partitions concurrently. That reduces the startup time to a couple RTTs even for topics with a numerous partitions.
		started := make(chan *partition)
		var wg sync.WaitGroup
		for _, p := range added {
			wg.Add(1)
			go func(p int32) {
				defer wg.Done()
				ob := oresp.GetBlock(con.topic, p)
				if ob == nil {
					// can't start this partition without an offset
					con.deliverError("FetchOffset response", p, fmt.Errorf("partition %d missing", p))
					return
				}
				if ob.Err != 0 {
					con.deliverError("FetchOffset response", p, ob.Err)
					return
				}

				// run the committed offset through the StartingOffset() hook
				offset, err := con.cl.config.StartingOffset(con.topic, p, ob.Offset, con.cl.client)
				if err != nil {
					con.deliverError("StartingOffset", p, err)
					return
				}

				logf("consumer %q consuming %q partition %d at offset %d", con.cl.group_name, con.topic, p, offset)

				var consumer sarama.PartitionConsumer
				if con.cl.config.NoMessages {
					// don't fetch messages. Let the caller do their own msg fetching.
					if con.cl.config.PartitionStartNotification != nil {
						con.cl.config.PartitionStartNotification(con.topic, p, offset)
					}
				} else {
					consumer, err = con.consumer.ConsumePartition(con.topic, p, offset)
					if err != nil {
						con.deliverError(fmt.Sprintf("sarama.ConsumePartition at offset %d", offset), p, err)

						// If the error is ErrOffsetOutOfRange then give ourselves one chance to recover
						if err != sarama.ErrOffsetOutOfRange {
							// otherwise we can't consume this partition.
							return
						}

						offset, err = con.cl.config.OffsetOutOfRange(con.topic, p, con.cl.client)
						if err != nil {
							// should we deliver them their own error? I guess so.
							con.deliverError("OffsetOutOfRange callback", p, err)
							return
						}

						logf("consumer %q skipping to %q partition %d offset %d", con.cl.group_name, con.topic, p, offset)
						consumer, err = con.consumer.ConsumePartition(con.topic, p, offset)
						if err != nil {
							con.deliverError(fmt.Sprintf("sarama.ConsumePartition at offset %d", offset), p, err)
							// it didn't work with their offset either. give up
							// (we could go into a loop and call them again, but what would that solve?)
							return
						}
						// it worked with the new offset; carry on
					}
				}

				part := &partition{
					con:                con,
					consumer:           consumer,
					partition:          p,
					next_commit_offset: offset,
				}

				if !con.cl.config.NoMessages {
					go part.run()
				}

				started <- part
			}(p)
		}
		go func() {
			wg.Wait()
			close(started)
		}()

		for part := range started {
			partitions[part.partition] = part
		}
	}

	// restart consuming a partition at a new[er] offset
	restart_partition := func(part *partition) {
		// we kill the old and start a new partition consumer since there is no way to seek an existing sarama.PartitionConsumer in sarama's November 2016 API)
		p := part.partition

		// first remove the old partition consumer. Once it gets a ErrOffsetOutOfRange it's unable to function.
		// since it had an out-of-range offset, it can't commit its offset either
		if pa, ok := partitions[p]; !ok || part != pa {
			// this is an unknown partition, or we've already killed it; ignore the request
			return
		}
		delete(partitions, p)
		part.consumer.Close()

		// then ask what the new starting offset should be
		offset, err := con.cl.config.OffsetOutOfRange(con.topic, p, con.cl.client)
		if err != nil {
			// should we deliver them their own error? I guess so.
			con.deliverError("OffsetOutOfRange callback", p, err)
			return
		}

		// finally make a new partition consuming starting at the given offset
		consumer, err := con.consumer.ConsumePartition(con.topic, p, offset)
		if err != nil {
			con.deliverError(fmt.Sprintf("sarama.ConsumePartition at offset %d", offset), p, err)
			return
		}

		logf("consumer %q restarting consuming %q partition %d at offset %d", con.cl.group_name, con.topic, p, offset)

		part = &partition{
			con:                con,
			consumer:           consumer,
			partition:          p,
			next_commit_offset: offset,
		}
		go part.run()
		partitions[p] = part
	}

	for {
		select {
		case msg := <-con.premessages:
			msgf("premessage msg %q:%d/%d", msg)
			// keep track of msg's offset so we can match it with Done, and deliver the msg
			part := partitions[msg.Partition]
			if part == nil {
				// message from a stale consumer; ignore it
				dbgf("no partition %d", msg.Partition)
				continue
			}
			if part.next_commit_offset == sarama.OffsetNewest || part.next_commit_offset == sarama.OffsetOldest {
				// we now know the starting offset. make as if we'd been asked to start there
				part.next_commit_offset = msg.Offset
			}
			delta := msg.Offset - part.next_commit_offset
			if delta < 0 { // || delta > max-out-of-order  (TODO if needed, which so far hasn't been the case)
				dbgf("stale message %q:%d/%d", msg.Topic, msg.Partition, msg.Offset)
				// we can't take this message into account
				continue
			}
			index := int(delta) >> lg2_offsets_per_bucket
			for index >= len(part.buckets) {
				// add a new bucket
				part.buckets = append(part.buckets, bucket{})
			}
			part.buckets[index].read++

			// and deliver the msg (or handle any of the other messages which can arrive)
		deliver_loop:
			for {
				select {
				case con.messages <- msg:
					msgf("delivered msg %q:%d/%d", msg)
					// success
					break deliver_loop

				case msg2 := <-con.done:
					done(msg2)
				case a := <-con.assignments:
					assignment(a)
				case c := <-con.commit_reqs:
					commit_req(c)
				case p := <-con.restart_partitions:
					restart_partition(p)
				case <-con.closed:
					// the defered operations do the work
					return
				}
			}

		case msg := <-con.done:
			done(msg)
		case a := <-con.assignments:
			assignment(a)
		case c := <-con.commit_reqs:
			commit_req(c)
		case p := <-con.restart_partitions:
			restart_partition(p)
		case <-con.closed:
			// the defered operations do the work
			return
		}
	}
}

func (con *consumer) Done(msg *sarama.ConsumerMessage) {
	// send it back to consumer.run to be processed synchronously
	msgf("Done(%q:%d/%d)", msg)
	select {
	case con.done <- msg:
		// great, msg delivered
	case <-con.closed:
		// consumer has closed
	}
}

// partition contains the data associated with us consuming one partition
type partition struct {
	con       *consumer
	consumer  sarama.PartitionConsumer // nil if Config.NoMessages is set
	partition int32                    // partition number

	next_commit_offset int64 // the offset to commit to kafka (by convention the most recently completed msg's Offset+1). When !in_order_done this is the offset of bucket[0]. Can be OffsetNewest or OffsetOldest if we haven't received any msgs and started at one of those offsets.

	// buckets of # of offsets read from kafka, and the # of offsets completed by a call to Done(). the difference is the # of offsets in flight in the calling code
	// we group offsets in groups of 128 (offsets_per_bucket) and simply keep a count of how many are outstanding
	// any time the two counts are equal then the offsets are committable. Otherwise we can't tell which is the not yet Done() offset and so we don't know
	// These are used only if con.in_order_done is disabled.
	buckets            []bucket
	bucket_0_highwater uint8 // highwater mark of commits from buckets[0]
}

// a bucket of message offsets. It contains counts of the msgs with offsets in the range base to base+offsets_per_bucket
type bucket struct {
	read uint8 // count of how many messages have been read from kafka
	done uint8 // count of how many messages are Done()
}

// log base 2 of the number of offsets in a bucket
const lg2_offsets_per_bucket = 7 // using 128 gets rid of the edge case of 0 == 256, which I fear would be a source of bugs
const offsets_per_bucket = 1 << lg2_offsets_per_bucket

// wrap a sarama.ConsumerError into an *Error
func (part *partition) makeConsumerError(cerr *sarama.ConsumerError) *Error {
	Err := part.con.makeError("consuming from sarama", cerr.Err)
	Err.Topic = cerr.Topic
	Err.Partition = cerr.Partition
	return Err
}

// return the offset to commit to kafka
func (part *partition) compute_commit_offset() int64 {
	offset := part.next_commit_offset
	if !part.con.in_order_done {
		if offset != sarama.OffsetNewest && offset != sarama.OffsetOldest {
			// add to that the portion of the last block we know has been completed (this is often useful when the traffic rate is low or a client shuts down cleanly, since it has probably cleanly returned all offsets we've delivered)
			offset += int64(part.bucket_0_highwater)
		} // else return the special case offset as-is
	}
	return offset
}

// run consumes from the partition and delivers it to the consumer
func (part *partition) run() {
	con := part.con
	defer dbgf("partition consumer of %q partition %d exiting", con.topic, part.partition)
	msgs := part.consumer.Messages()
	errors := part.consumer.Errors()
	sink := con.messages
	if !con.in_order_done {
		// messages have to go throught a pre-delivery step
		sink = con.premessages
	}
	restart_requested := false
	for {
		select {
		case msg, ok := <-msgs:
			if ok {
				msgf("got msg %q:%d/%d", msg)
				if con.max_age > 0 && !restart_requested && !msg.Timestamp.IsZero() {
					if age := time.Since(msg.Timestamp); age > con.max_age {
						if depth := part.consumer.HighWaterMarkOffset() - msg.Offset; depth > 80 { // TODO make 80 configurable? Or tune it to the observed message rate?
							logf("consumer %q lagging by %v (%d offsets) in %q partition %d offset %d; asking to skip ahead", con.cl.group_name, age, depth, con.topic, part.partition, msg.Offset)
							select {
							case con.restart_partitions <- part:
								restart_requested = true
							case <-con.closed:
								return
							}
						} // else we are fewer than 80 messages behind head. don't bother skipping ahead
					}
				} // else max_age isn't enforced, or we're already requested to be restarted
				select {
				case sink <- msg:
				case <-con.closed:
					return
				}
			} else {
				dbgf("draining topic %q partition %d errors", con.topic, part.partition)
				// deliver any remaining errors, and exit
				for sarama_err := range errors {
					con.cl.deliverError("", part.makeConsumerError(sarama_err))
				}
				return
			}
		case sarama_err, ok := <-errors:
			if ok {
				// pick out ErrOffsetOutOfRange errors. These happen if the consumer offset falls off the tail of the kafka log.
				// this easily happens in two cases: when the consumer is too slow, or when the consumer has been stopped for too long.
				// This error cannot be fixed without seeking to a valid offset. However we can't assume that OffsetNewest is the right
				// choice, nor OffsetOldest, nor "5 minutes ago" or anything else. It's up to the user to decide.
				if sarama_err.Err == sarama.ErrOffsetOutOfRange {
					logf("consumer %q of %q partition %d received ErrOffsetOutOfRange and will be restarted", con.cl.group_name, con.topic, part.partition)
					select {
					case con.restart_partitions <- part:
						restart_requested = true
					case <-con.closed:
						return
					}
					// should we keep reading from the partition? it's unlikely to produce much
				}
				// and always deliver the error
				con.cl.deliverError("", part.makeConsumerError(sarama_err))
			} else {
				// finish off any remaining messages, and exit
				dbgf("draining topic %q partition %d msgs", con.topic, part.partition)
				for msg := range msgs {
					select {
					case sink <- msg:
					case <-con.closed:
						return
					}
				}
				return
			}
		}
	}
}

// difference returns the differences (additions and subtractions) between two slices of int32.
// typically the slices contain partition numbers.
func difference(old map[int32]*partition, next []int32) (added, removed []int32) {
	o := make(int32Slice, 0, len(old))
	for p := range old {
		o = append(o, p)
	}

	n := make(int32Slice, len(next))
	copy(n, next)

	sort.Sort(o)
	sort.Sort(n)

	i, j := 0, 0
	for i < len(o) && j < len(n) {
		if o[i] < n[j] {
			removed = append(removed, o[i])
			i++
		} else if o[i] > n[j] {
			added = append(added, n[j])
			j++
		} else {
			i++
			j++
		}
	}
	removed = append(removed, o[i:]...)
	added = append(added, n[j:]...)

	return
}

// a sortable []int32
type int32Slice []int32

func (p int32Slice) Len() int           { return len(p) }
func (p int32Slice) Less(i, j int) bool { return p[i] < p[j] }
func (p int32Slice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }