func (b *Broker) isController() bool {
	return b.isLeader()
}
isController returns true if this is the cluster controller.
package jocko
import (
	"bytes"
	"container/ring"
	"context"
	"fmt"
	"io"
	"math/rand"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/davecgh/go-spew/spew"
	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb"
	"github.com/hashicorp/serf/serf"
	opentracing "github.com/opentracing/opentracing-go"
	"github.com/pkg/errors"
	uuid "github.com/satori/go.uuid"
	"github.com/travisjeffery/jocko/commitlog"
	"github.com/travisjeffery/jocko/jocko/config"
	"github.com/travisjeffery/jocko/jocko/fsm"
	"github.com/travisjeffery/jocko/jocko/metadata"
	"github.com/travisjeffery/jocko/jocko/structs"
	"github.com/travisjeffery/jocko/jocko/util"
	"github.com/travisjeffery/jocko/log"
	"github.com/travisjeffery/jocko/protocol"
)
var (
	brokerVerboseLogs bool
	ErrTopicExists            = errors.New("topic exists already")
	ErrInvalidArgument        = errors.New("no logger set")
	OffsetsTopicName          = "__consumer_offsets"
	OffsetsTopicNumPartitions = 50
)
const (
	serfLANSnapshot   = "serf/local.snapshot"
	raftState         = "raft/"
	raftLogCacheSize  = 512
	snapshotsRetained = 2
)
func init() {
	spew.Config.Indent = ""
	e := os.Getenv("JOCKODEBUG")
	if strings.Contains(e, "broker=1") {
		brokerVerboseLogs = true
	}
}
type Broker struct {
	sync.RWMutex
	config *config.Config
	readyForConsistentReads int32
	brokerLookup  *brokerLookup
	replicaLookup *replicaLookup
	raft          *raft.Raft
	raftStore     *raftboltdb.BoltStore
	raftTransport *raft.NetworkTransport
	raftInmem     *raft.InmemStore
	raftNotifyCh <-chan bool
	reconcileCh      chan serf.Member
	serf             *serf.Serf
	fsm              *fsm.FSM
	eventChLAN       chan serf.Event
	logStateInterval time.Duration
	tracer opentracing.Tracer
	shutdownCh   chan struct{}
	shutdown     bool
	shutdownLock sync.Mutex
}
func NewBroker(config *config.Config, tracer opentracing.Tracer) (*Broker, error) {
	b := &Broker{
		config:           config,
		shutdownCh:       make(chan struct{}),
		eventChLAN:       make(chan serf.Event, 256),
		brokerLookup:     NewBrokerLookup(),
		replicaLookup:    NewReplicaLookup(),
		reconcileCh:      make(chan serf.Member, 32),
		tracer:           tracer,
		logStateInterval: time.Millisecond * 250,
	}
	if err := b.setupRaft(); err != nil {
		b.Shutdown()
		return nil, fmt.Errorf("start raft: %v", err)
	}
	var err error
	b.serf, err = b.setupSerf(config.SerfLANConfig, b.eventChLAN, serfLANSnapshot)
	if err != nil {
		return nil, err
	}
	go b.lanEventHandler()
	go b.monitorLeadership()
	go b.logState()
	return b, nil
}
func (b *Broker) Run(ctx context.Context, requests <-chan *Context, responses chan<- *Context) {
	for {
		select {
		case reqCtx := <-requests:
			log.Debug.Printf("broker/%d: request: %v", b.config.ID, reqCtx)
			if reqCtx == nil {
				goto DONE
			}
			queueSpan, ok := reqCtx.Value(requestQueueSpanKey).(opentracing.Span)
			if ok {
				queueSpan.Finish()
			}
			var res protocol.ResponseBody
			switch req := reqCtx.req.(type) {
			case *protocol.ProduceRequest:
				res = b.handleProduce(reqCtx, req)
			case *protocol.FetchRequest:
				res = b.handleFetch(reqCtx, req)
			case *protocol.OffsetsRequest:
				res = b.handleOffsets(reqCtx, req)
			case *protocol.MetadataRequest:
				res = b.handleMetadata(reqCtx, req)
			case *protocol.LeaderAndISRRequest:
				res = b.handleLeaderAndISR(reqCtx, req)
			case *protocol.StopReplicaRequest:
				res = b.handleStopReplica(reqCtx, req)
			case *protocol.UpdateMetadataRequest:
				res = b.handleUpdateMetadata(reqCtx, req)
			case *protocol.ControlledShutdownRequest:
				res = b.handleControlledShutdown(reqCtx, req)
			case *protocol.OffsetCommitRequest:
				res = b.handleOffsetCommit(reqCtx, req)
			case *protocol.OffsetFetchRequest:
				res = b.handleOffsetFetch(reqCtx, req)
			case *protocol.FindCoordinatorRequest:
				res = b.handleFindCoordinator(reqCtx, req)
			case *protocol.JoinGroupRequest:
				res = b.handleJoinGroup(reqCtx, req)
			case *protocol.HeartbeatRequest:
				res = b.handleHeartbeat(reqCtx, req)
			case *protocol.LeaveGroupRequest:
				res = b.handleLeaveGroup(reqCtx, req)
			case *protocol.SyncGroupRequest:
				res = b.handleSyncGroup(reqCtx, req)
			case *protocol.DescribeGroupsRequest:
				res = b.handleDescribeGroups(reqCtx, req)
			case *protocol.ListGroupsRequest:
				res = b.handleListGroups(reqCtx, req)
			case *protocol.SaslHandshakeRequest:
				res = b.handleSaslHandshake(reqCtx, req)
			case *protocol.APIVersionsRequest:
				res = b.handleAPIVersions(reqCtx, req)
			case *protocol.CreateTopicRequests:
				res = b.handleCreateTopic(reqCtx, req)
			case *protocol.DeleteTopicsRequest:
				res = b.handleDeleteTopics(reqCtx, req)
			}
			parentSpan := opentracing.SpanFromContext(reqCtx)
			queueSpan = b.tracer.StartSpan("broker: queue response", opentracing.ChildOf(parentSpan.Context()))
			responseCtx := context.WithValue(reqCtx, responseQueueSpanKey, queueSpan)
			responses <- &Context{
				parent: responseCtx,
				conn:   reqCtx.conn,
				header: reqCtx.header,
				res: &protocol.Response{
					CorrelationID: reqCtx.header.CorrelationID,
					Body:          res,
				},
			}
		case <-ctx.Done():
			goto DONE
		}
	}
DONE:
	log.Debug.Printf("broker/%d: run done", b.config.ID)
	return
}
func (b *Broker) JoinLAN(addrs ...string) protocol.Error {
	if _, err := b.serf.Join(addrs, true); err != nil {
		return protocol.ErrUnknown.WithErr(err)
	}
	return protocol.ErrNone
}
func span(ctx context.Context, tracer opentracing.Tracer, op string) opentracing.Span {
	if ctx == nil {
		return tracer.StartSpan("broker: " + op)
	}
	parentSpan := opentracing.SpanFromContext(ctx)
	if parentSpan == nil {
		return tracer.StartSpan("broker: " + op)
	}
	return tracer.StartSpan("broker: "+op, opentracing.ChildOf(parentSpan.Context()))
}
var apiVersions = &protocol.APIVersionsResponse{APIVersions: protocol.APIVersions}
func (b *Broker) handleAPIVersions(ctx *Context, req *protocol.APIVersionsRequest) *protocol.APIVersionsResponse {
	sp := span(ctx, b.tracer, "api versions")
	defer sp.Finish()
	return apiVersions
}
func (b *Broker) handleCreateTopic(ctx *Context, reqs *protocol.CreateTopicRequests) *protocol.CreateTopicsResponse {
	sp := span(ctx, b.tracer, "create topic")
	defer sp.Finish()
	res := new(protocol.CreateTopicsResponse)
	res.APIVersion = reqs.Version()
	res.TopicErrorCodes = make([]*protocol.TopicErrorCode, len(reqs.Requests))
	isController := b.isController()
	sp.LogKV("is controller", isController)
	for i, req := range reqs.Requests {
		if !isController {
			res.TopicErrorCodes[i] = &protocol.TopicErrorCode{
				Topic:     req.Topic,
				ErrorCode: protocol.ErrNotController.Code(),
			}
			continue
		}
		if req.ReplicationFactor > int16(len(b.LANMembers())) {
			res.TopicErrorCodes[i] = &protocol.TopicErrorCode{
				Topic:     req.Topic,
				ErrorCode: protocol.ErrInvalidReplicationFactor.Code(),
			}
			continue
		}
		err := b.withTimeout(reqs.Timeout, func() protocol.Error {
			return b.createTopic(ctx, req)
		})
		res.TopicErrorCodes[i] = &protocol.TopicErrorCode{
			Topic:     req.Topic,
			ErrorCode: err.Code(),
		}
	}
	return res
}
func (b *Broker) handleDeleteTopics(ctx *Context, reqs *protocol.DeleteTopicsRequest) *protocol.DeleteTopicsResponse {
	sp := span(ctx, b.tracer, "delete topics")
	defer sp.Finish()
	res := new(protocol.DeleteTopicsResponse)
	res.APIVersion = reqs.Version()
	res.TopicErrorCodes = make([]*protocol.TopicErrorCode, len(reqs.Topics))
	isController := b.isController()
	for i, topic := range reqs.Topics {
		if !isController {
			res.TopicErrorCodes[i] = &protocol.TopicErrorCode{
				Topic:     topic,
				ErrorCode: protocol.ErrNotController.Code(),
			}
			continue
		}
		err := b.withTimeout(reqs.Timeout, func() protocol.Error {
			_, err := b.raftApply(structs.DeregisterTopicRequestType, structs.DeregisterTopicRequest{
				structs.Topic{
					Topic: topic,
				},
			})
			if err != nil {
				return protocol.ErrUnknown.WithErr(err)
			}
			return protocol.ErrNone
		})
		res.TopicErrorCodes[i] = &protocol.TopicErrorCode{
			Topic:     topic,
			ErrorCode: err.Code(),
		}
	}
	return res
}
func (b *Broker) handleLeaderAndISR(ctx *Context, req *protocol.LeaderAndISRRequest) *protocol.LeaderAndISRResponse {
	sp := span(ctx, b.tracer, "leader and isr")
	defer sp.Finish()
	res := &protocol.LeaderAndISRResponse{
		Partitions: make([]*protocol.LeaderAndISRPartition, len(req.PartitionStates)),
	}
	res.APIVersion = req.Version()
	setErr := func(i int, p *protocol.PartitionState, err protocol.Error) {
		res.Partitions[i] = &protocol.LeaderAndISRPartition{
			ErrorCode: err.Code(),
			Partition: p.Partition,
			Topic:     p.Topic,
		}
	}
	for i, p := range req.PartitionStates {
		replica := &Replica{
			BrokerID: b.config.ID,
			Partition: structs.Partition{
				ID:              p.Partition,
				Partition:       p.Partition,
				Topic:           p.Topic,
				ISR:             p.ISR,
				AR:              p.Replicas,
				ControllerEpoch: p.ZKVersion,
				LeaderEpoch:     p.LeaderEpoch,
				Leader:          p.Leader,
			},
			IsLocal: true,
		}
		b.replicaLookup.AddReplica(replica)
		if p.Leader == b.config.ID && (replica.Partition.Leader == b.config.ID) {
			if err := b.startReplica(replica); err != protocol.ErrNone {
				setErr(i, p, err)
				continue
			}
			if err := b.becomeLeader(replica, p); err != protocol.ErrNone {
				setErr(i, p, err)
				continue
			}
		} else if contains(p.Replicas, b.config.ID) && (p.Leader != b.config.ID) {
			if err := b.startReplica(replica); err != protocol.ErrNone {
				setErr(i, p, err)
				continue
			}
			if err := b.becomeFollower(replica, p); err != protocol.ErrNone {
				setErr(i, p, err)
				continue
			}
		}
		res.Partitions[i] = &protocol.LeaderAndISRPartition{Partition: p.Partition, Topic: p.Topic, ErrorCode: protocol.ErrNone.Code()}
	}
	return res
}
func (b *Broker) handleOffsets(ctx *Context, req *protocol.OffsetsRequest) *protocol.OffsetsResponse {
	sp := span(ctx, b.tracer, "offsets")
	defer sp.Finish()
	res := new(protocol.OffsetsResponse)
	res.APIVersion = req.Version()
	res.Responses = make([]*protocol.OffsetResponse, len(req.Topics))
	for i, t := range req.Topics {
		res.Responses[i] = new(protocol.OffsetResponse)
		res.Responses[i].Topic = t.Topic
		res.Responses[i].PartitionResponses = make([]*protocol.PartitionResponse, 0, len(t.Partitions))
		for _, p := range t.Partitions {
			pres := new(protocol.PartitionResponse)
			pres.Partition = p.Partition
			replica, err := b.replicaLookup.Replica(t.Topic, p.Partition)
			if err != nil {
				pres.ErrorCode = protocol.ErrUnknown.Code()
				continue
			}
			var offset int64
			if p.Timestamp == -2 {
				offset = replica.Log.OldestOffset()
			} else {
				offset = replica.Log.NewestOffset()
			}
			pres.Offsets = []int64{offset}
			res.Responses[i].PartitionResponses = append(res.Responses[i].PartitionResponses, pres)
		}
	}
	return res
}
func (b *Broker) handleProduce(ctx *Context, req *protocol.ProduceRequest) *protocol.ProduceResponse {
	sp := span(ctx, b.tracer, "produce")
	defer sp.Finish()
	res := new(protocol.ProduceResponse)
	res.APIVersion = req.Version()
	res.Responses = make([]*protocol.ProduceTopicResponse, len(req.TopicData))
	log.Debug.Printf("broker/%d: produce: %#v", b.config.ID, req)
	for i, td := range req.TopicData {
		log.Debug.Printf("broker/%d: produce to partition: %d: %v", b.config.ID, i, td)
		tres := make([]*protocol.ProducePartitionResponse, len(td.Data))
		for j, p := range td.Data {
			pres := &protocol.ProducePartitionResponse{}
			pres.Partition = p.Partition
			err := b.withTimeout(req.Timeout, func() protocol.Error {
				state := b.fsm.State()
				_, t, err := state.GetTopic(td.Topic)
				if err != nil {
					log.Error.Printf("broker/%d: produce to partition error: get topic: %s", b.config.ID, err)
					return protocol.ErrUnknown.WithErr(err)
				}
				if t == nil {
					log.Error.Printf("broker/%d: produce to partition error: unknown topic", b.config.ID)
					return protocol.ErrUnknownTopicOrPartition
				}
				replica, err := b.replicaLookup.Replica(td.Topic, p.Partition)
				if err != nil || replica == nil || replica.Log == nil {
					log.Error.Printf("broker/%d: produce to partition error: %s", b.config.ID, err)
					pres.Partition = p.Partition
					return protocol.ErrReplicaNotAvailable
				}
				offset, appendErr := replica.Log.Append(p.RecordSet)
				if appendErr != nil {
					log.Error.Printf("broker/%d: log append error: %s", b.config.ID, err)
					return protocol.ErrUnknown
				}
				pres.BaseOffset = offset
				pres.LogAppendTime = time.Now()
				return protocol.ErrNone
			})
			pres.ErrorCode = err.Code()
			tres[j] = pres
		}
		res.Responses[i] = &protocol.ProduceTopicResponse{
			Topic:              td.Topic,
			PartitionResponses: tres,
		}
	}
	return res
}
func (b *Broker) handleMetadata(ctx *Context, req *protocol.MetadataRequest) *protocol.MetadataResponse {
	sp := span(ctx, b.tracer, "metadata")
	defer sp.Finish()
	state := b.fsm.State()
	brokers := make([]*protocol.Broker, 0, len(b.LANMembers()))
	_, nodes, err := state.GetNodes()
	if err != nil {
		panic(err)
	}
	var passing []*structs.Node
	for _, n := range nodes {
		if n.Check.Status == structs.HealthPassing {
			passing = append(passing, n)
		}
	}
	for _, mem := range b.LANMembers() {
		if mem.Status != serf.StatusAlive {
			continue
		}
		m, ok := metadata.IsBroker(mem)
		if !ok {
			continue
		}
		brokers = append(brokers, &protocol.Broker{
			NodeID: m.ID.Int32(),
			Host:   m.Host(),
			Port:   m.Port(),
		})
	}
	var topicMetadata []*protocol.TopicMetadata
	topicMetadataFn := func(topic *structs.Topic, err protocol.Error) *protocol.TopicMetadata {
		if err != protocol.ErrNone {
			return &protocol.TopicMetadata{
				TopicErrorCode: err.Code(),
				Topic:          topic.Topic,
			}
		}
		partitionMetadata := make([]*protocol.PartitionMetadata, 0, len(topic.Partitions))
		for id := range topic.Partitions {
			_, p, err := state.GetPartition(topic.Topic, id)
			if err != nil {
				partitionMetadata = append(partitionMetadata, &protocol.PartitionMetadata{
					PartitionID:        id,
					PartitionErrorCode: protocol.ErrUnknown.Code(),
				})
				continue
			}
			if p == nil {
				partitionMetadata = append(partitionMetadata, &protocol.PartitionMetadata{
					PartitionID:        id,
					PartitionErrorCode: protocol.ErrUnknownTopicOrPartition.Code(),
				})
				continue
			}
			partitionMetadata = append(partitionMetadata, &protocol.PartitionMetadata{
				PartitionID:        p.ID,
				PartitionErrorCode: protocol.ErrNone.Code(),
				Leader:             p.Leader,
				Replicas:           p.AR,
				ISR:                p.ISR,
			})
		}
		return &protocol.TopicMetadata{
			TopicErrorCode:    protocol.ErrNone.Code(),
			Topic:             topic.Topic,
			PartitionMetadata: partitionMetadata,
		}
	}
	if len(req.Topics) == 0 {
		_, topics, _ := state.GetTopics()
		topicMetadata = make([]*protocol.TopicMetadata, 0, len(topics))
		for _, topic := range topics {
			topicMetadata = append(topicMetadata, topicMetadataFn(topic, protocol.ErrNone))
		}
	} else {
		topicMetadata = make([]*protocol.TopicMetadata, 0, len(req.Topics))
		for _, topicName := range req.Topics {
			_, topic, err := state.GetTopic(topicName)
			if topic == nil {
				topicMetadata = append(topicMetadata, topicMetadataFn(&structs.Topic{Topic: topicName}, protocol.ErrUnknownTopicOrPartition))
			} else if err != nil {
				topicMetadata = append(topicMetadata, topicMetadataFn(&structs.Topic{Topic: topicName}, protocol.ErrUnknown.WithErr(err)))
			} else {
				topicMetadata = append(topicMetadata, topicMetadataFn(topic, protocol.ErrNone))
			}
		}
	}
	res := &protocol.MetadataResponse{
		Brokers:       brokers,
		TopicMetadata: topicMetadata,
	}
	res.APIVersion = req.Version()
	return res
}
func (b *Broker) handleFindCoordinator(ctx *Context, req *protocol.FindCoordinatorRequest) *protocol.FindCoordinatorResponse {
	sp := span(ctx, b.tracer, "find coordinator")
	defer sp.Finish()
	res := &protocol.FindCoordinatorResponse{}
	res.APIVersion = req.Version()
	var broker *metadata.Broker
	var p *structs.Partition
	var i int32
	state := b.fsm.State()
	topic, err := b.offsetsTopic(ctx)
	if err != nil {
		goto ERROR
	}
	i = int32(util.Hash(req.CoordinatorKey) % uint64(len(topic.Partitions)))
	_, p, err = state.GetPartition(OffsetsTopicName, i)
	if err != nil {
		goto ERROR
	}
	if p == nil {
		res.ErrorCode = protocol.ErrUnknownTopicOrPartition.Code()
		goto ERROR
	}
	broker = b.brokerLookup.BrokerByID(raft.ServerID(fmt.Sprintf("%d", p.Leader)))
	res.Coordinator.NodeID = broker.ID.Int32()
	res.Coordinator.Host = broker.Host()
	res.Coordinator.Port = broker.Port()
	return res
ERROR:
	if res.ErrorCode == 0 {
		res.ErrorCode = protocol.ErrUnknown.Code()
	}
	log.Error.Printf("broker/%d: broker: %v: coordinator error: %s", b.config.ID, broker, err)
	return res
}
func (b *Broker) handleJoinGroup(ctx *Context, r *protocol.JoinGroupRequest) *protocol.JoinGroupResponse {
	sp := span(ctx, b.tracer, "join group")
	defer sp.Finish()
	res := &protocol.JoinGroupResponse{}
	res.APIVersion = r.Version()
	state := b.fsm.State()
	_, group, err := state.GetGroup(r.GroupID)
	if err != nil {
		log.Error.Printf("broker/%d: get group error: %s", b.config.ID, err)
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	if group == nil {
		group = &structs.Group{
			Group:       r.GroupID,
			Coordinator: b.config.ID,
			Members:     make(map[string]structs.Member),
		}
	}
	if r.MemberID == "" {
		r.MemberID = ctx.Header().ClientID + "-" + uuid.NewV1().String()
		group.Members[r.MemberID] = structs.Member{ID: r.MemberID}
	}
	if group.LeaderID == "" {
		group.LeaderID = r.MemberID
	}
	_, err = b.raftApply(structs.RegisterGroupRequestType, structs.RegisterGroupRequest{
		Group: *group,
	})
	if err != nil {
		log.Error.Printf("broker/%d: register group error: %s", b.config.ID, err)
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	res.GenerationID = 0
	res.LeaderID = group.LeaderID
	res.MemberID = r.MemberID
	if res.LeaderID == res.MemberID {
		for _, m := range group.Members {
			res.Members = append(res.Members, protocol.Member{MemberID: m.ID, MemberMetadata: m.Metadata})
		}
	}
	return res
}
func (b *Broker) handleLeaveGroup(ctx *Context, r *protocol.LeaveGroupRequest) *protocol.LeaveGroupResponse {
	sp := span(ctx, b.tracer, "leave group")
	defer sp.Finish()
	res := &protocol.LeaveGroupResponse{}
	res.APIVersion = r.Version()
	state := b.fsm.State()
	_, group, err := state.GetGroup(r.GroupID)
	if err != nil {
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	if group == nil {
		res.ErrorCode = protocol.ErrInvalidGroupId.Code()
		return res
	}
	if _, ok := group.Members[r.MemberID]; !ok {
		res.ErrorCode = protocol.ErrUnknownMemberId.Code()
		return res
	}
	delete(group.Members, r.MemberID)
	_, err = b.raftApply(structs.RegisterGroupRequestType, structs.RegisterGroupRequest{
		Group: *group,
	})
	if err != nil {
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	return res
}
func (b *Broker) handleSyncGroup(ctx *Context, r *protocol.SyncGroupRequest) *protocol.SyncGroupResponse {
	sp := span(ctx, b.tracer, "sync group")
	defer sp.Finish()
	state := b.fsm.State()
	res := &protocol.SyncGroupResponse{}
	res.APIVersion = r.Version()
	_, group, err := state.GetGroup(r.GroupID)
	if err != nil {
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	if group == nil {
		res.ErrorCode = protocol.ErrInvalidGroupId.Code()
		return res
	}
	if _, ok := group.Members[r.MemberID]; !ok {
		res.ErrorCode = protocol.ErrUnknownMemberId.Code()
		return res
	}
	if r.GenerationID != group.GenerationID {
		res.ErrorCode = protocol.ErrIllegalGeneration.Code()
		return res
	}
	switch group.State {
	case structs.GroupStateEmpty, structs.GroupStateDead:
		res.ErrorCode = protocol.ErrUnknownMemberId.Code()
		return res
	case structs.GroupStatePreparingRebalance:
		res.ErrorCode = protocol.ErrRebalanceInProgress.Code()
		return res
	case structs.GroupStateCompletingRebalance:
		if group.LeaderID == r.MemberID {
			var assignment []protocol.GroupAssignment
			for _, ga := range r.GroupAssignments {
				if _, ok := group.Members[ga.MemberID]; !ok {
					assignment = append(assignment, protocol.GroupAssignment{
						MemberID:         ga.MemberID,
						MemberAssignment: nil,
					})
				} else {
					assignment = append(assignment, ga)
				}
			}
		}
	case structs.GroupStateStable:
	}
	if group.LeaderID == r.MemberID {
		for _, ga := range r.GroupAssignments {
			if m, ok := group.Members[ga.MemberID]; ok {
				m.Assignment = ga.MemberAssignment
			} else {
				panic("sync group: unknown member")
			}
		}
		_, err = b.raftApply(structs.RegisterGroupRequestType, structs.RegisterGroupRequest{
			Group: *group,
		})
		if err != nil {
			res.ErrorCode = protocol.ErrUnknown.Code()
			return res
		}
	} else {
		if m, ok := group.Members[r.MemberID]; ok {
			res.MemberAssignment = m.Assignment
		} else {
			panic(fmt.Errorf("sync group: unknown member: %s", r.MemberID))
		}
	}
	return res
}
func (b *Broker) handleHeartbeat(ctx *Context, r *protocol.HeartbeatRequest) *protocol.HeartbeatResponse {
	sp := span(ctx, b.tracer, "heartbeat")
	defer sp.Finish()
	res := &protocol.HeartbeatResponse{}
	res.APIVersion = r.Version()
	state := b.fsm.State()
	_, group, err := state.GetGroup(r.GroupID)
	if err != nil {
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	if group == nil {
		res.ErrorCode = protocol.ErrInvalidGroupId.Code()
		return res
	}
	res.ErrorCode = protocol.ErrNone.Code()
	return res
}
func (b *Broker) handleFetch(ctx *Context, r *protocol.FetchRequest) *protocol.FetchResponse {
	sp := span(ctx, b.tracer, "fetch")
	defer sp.Finish()
	fres := &protocol.FetchResponse{
		Responses: make(protocol.FetchTopicResponses, len(r.Topics)),
	}
	fres.APIVersion = r.Version()
	for i, topic := range r.Topics {
		fr := &protocol.FetchTopicResponse{
			Topic:              topic.Topic,
			PartitionResponses: make([]*protocol.FetchPartitionResponse, len(topic.Partitions)),
		}
		for j, p := range topic.Partitions {
			fpres := &protocol.FetchPartitionResponse{}
			fpres.Partition = p.Partition
			err := b.withTimeout(r.MaxWaitTime, func() protocol.Error {
				replica, err := b.replicaLookup.Replica(topic.Topic, p.Partition)
				if err != nil {
					return protocol.ErrReplicaNotAvailable
				}
				if replica.Partition.Leader != b.config.ID {
					return protocol.ErrNotLeaderForPartition
				}
				if replica.Log == nil {
					return protocol.ErrReplicaNotAvailable
				}
				rdr, rdrErr := replica.Log.NewReader(p.FetchOffset, p.MaxBytes)
				if rdrErr != nil {
					log.Error.Printf("broker/%d: replica log read error: %s", b.config.ID, rdrErr)
					return protocol.ErrUnknown.WithErr(rdrErr)
				}
				buf := new(bytes.Buffer)
				var n int32
				for n < r.MinBytes {
					nn, err := io.Copy(buf, rdr)
					if err != nil && err != io.EOF {
						log.Error.Printf("broker/%d: reader copy error", b.config.ID, err)
						return protocol.ErrUnknown.WithErr(rdrErr)
					}
					n += int32(nn)
					if err == io.EOF {
						break
					}
				}
				fpres.HighWatermark = replica.Log.NewestOffset() - 1
				fpres.RecordSet = buf.Bytes()
				return protocol.ErrNone
			})
			fpres.ErrorCode = err.Code()
			fr.PartitionResponses[j] = fpres
		}
		fres.Responses[i] = fr
	}
	return fres
}
func (b *Broker) handleSaslHandshake(ctx *Context, req *protocol.SaslHandshakeRequest) *protocol.SaslHandshakeResponse {
	panic("not implemented: sasl handshake")
	return nil
}
func (b *Broker) handleListGroups(ctx *Context, req *protocol.ListGroupsRequest) *protocol.ListGroupsResponse {
	sp := span(ctx, b.tracer, "create topic")
	defer sp.Finish()
	res := new(protocol.ListGroupsResponse)
	res.APIVersion = req.Version()
	state := b.fsm.State()
	fmt.Println("list")
	fmt.Println("list")
	fmt.Println("list")
	_, groups, err := state.GetGroups()
	if err != nil {
		res.ErrorCode = protocol.ErrUnknown.Code()
		return res
	}
	for _, group := range groups {
		res.Groups = append(res.Groups, protocol.ListGroup{
			GroupID: group.Group,
			ProtocolType: "consumer",
		})
	}
	return res
}
func (b *Broker) handleDescribeGroups(ctx *Context, req *protocol.DescribeGroupsRequest) *protocol.DescribeGroupsResponse {
	sp := span(ctx, b.tracer, "create topic")
	defer sp.Finish()
	res := new(protocol.DescribeGroupsResponse)
	res.APIVersion = req.Version()
	state := b.fsm.State()
	fmt.Println("describe")
	fmt.Println("describe")
	fmt.Println("describe")
	for _, id := range req.GroupIDs {
		group := protocol.Group{}
		_, g, err := state.GetGroup(id)
		if err != nil {
			group.ErrorCode = protocol.ErrUnknown.Code()
			group.GroupID = id
			res.Groups = append(res.Groups, group)
			return res
		}
		group.GroupID = id
		group.State = "Stable"
		group.ProtocolType = "consumer"
		group.Protocol = "consumer"
		for id, member := range g.Members {
			group.GroupMembers[id] = &protocol.GroupMember{
				ClientID: member.ID,
				ClientHost:            "",
				GroupMemberMetadata:   member.Metadata,
				GroupMemberAssignment: member.Assignment,
			}
		}
		res.Groups = append(res.Groups)
	}
	return res
}
func (b *Broker) handleStopReplica(ctx *Context, req *protocol.StopReplicaRequest) *protocol.StopReplicaResponse {
	panic("not implemented: stop replica")
	return nil
}
func (b *Broker) handleUpdateMetadata(ctx *Context, req *protocol.UpdateMetadataRequest) *protocol.UpdateMetadataResponse {
	panic("not implemented: update metadata")
	return nil
}
func (b *Broker) handleControlledShutdown(ctx *Context, req *protocol.ControlledShutdownRequest) *protocol.ControlledShutdownResponse {
	panic("not implemented: controlled shutdown")
	return nil
}
func (b *Broker) handleOffsetCommit(ctx *Context, req *protocol.OffsetCommitRequest) *protocol.OffsetCommitResponse {
	panic("not implemented: offset commit")
	return nil
}
func (b *Broker) handleOffsetFetch(ctx *Context, req *protocol.OffsetFetchRequest) *protocol.OffsetFetchResponse {
	sp := span(ctx, b.tracer, "create topic")
	defer sp.Finish()
	res := new(protocol.OffsetFetchResponse)
	res.APIVersion = req.Version()
	res.Responses = make([]protocol.OffsetFetchTopicResponse, len(req.Topics))
	return res
}