func (handler *GroupHandler) Setup(sarama.ConsumerGroupSession) error {
	close(handler.ready)
	return nil
}
Setup is run at the beginning of a new session, before ConsumeClaim
package kafka
import (
	"context"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"github.com/Shopify/sarama"
	"github.com/reugn/go-streams"
	"github.com/reugn/go-streams/flow"
	"github.com/reugn/go-streams/util"
)
type KafkaSource struct {
	consumer  sarama.ConsumerGroup
	handler   sarama.ConsumerGroupHandler
	topics    []string
	out       chan interface{}
	ctx       context.Context
	cancelCtx context.CancelFunc
	wg        *sync.WaitGroup
}
func NewKafkaSource(ctx context.Context, addrs []string, groupID string,
	config *sarama.Config, topics ...string) *KafkaSource {
	consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, config)
	util.Check(err)
	out := make(chan interface{})
	cctx, cancel := context.WithCancel(ctx)
	sink := &KafkaSource{
		consumerGroup,
		&GroupHandler{make(chan struct{}), out},
		topics,
		out,
		cctx,
		cancel,
		&sync.WaitGroup{},
	}
	go sink.init()
	return sink
}
func (ks *KafkaSource) claimLoop() {
	ks.wg.Add(1)
	defer func() {
		ks.wg.Done()
		log.Printf("Exiting the Kafka claimLoop")
	}()
	for {
		handler := ks.handler.(*GroupHandler)
		if err := ks.consumer.Consume(ks.ctx, ks.topics, handler); err != nil {
			log.Printf("Kafka consumer.Consume failed with: %v", err)
		}
		select {
		case <-ks.ctx.Done():
			return
		default:
		}
		handler.ready = make(chan struct{})
	}
}
func (ks *KafkaSource) init() {
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
	go ks.claimLoop()
	select {
	case <-sigchan:
		ks.cancelCtx()
	case <-ks.ctx.Done():
	}
	log.Printf("Closing the Kafka consumer")
	ks.wg.Wait()
	close(ks.out)
	ks.consumer.Close()
}
func (ks *KafkaSource) Via(_flow streams.Flow) streams.Flow {
	flow.DoStream(ks, _flow)
	return _flow
}
func (ks *KafkaSource) Out() <-chan interface{} {
	return ks.out
}
type GroupHandler struct {
	ready chan struct{}
	out   chan interface{}
}