func withRetry(
	b *sarama.Broker,
	cfg *sarama.Config,
	f func() error,
) error {
	var err error
	for max := 0; max < cfg.Metadata.Retry.Max; max++ {
		if ok, _ := b.Connected(); !ok {
			if err = b.Open(cfg); err == nil {
				err = f()
			}
		} else {
			err = f()
		}

		if err == nil {
			return nil
		}

		retry, reconnect := checkRetryQuery(err)
		if !retry {
			return err
		}

		time.Sleep(cfg.Metadata.Retry.Backoff)
		if reconnect {
			closeBroker(b)
		}
	}
	return err
}

// Start a rebalance cycle
func (cg *ConsumerGroup) rebalance() (err error) {
	var cids []string
	var pids []int32

	// Fetch a list of consumers and listen for changes
	if cids, cg.zkchange, err = cg.zoo.Consumers(cg.name); err != nil {
		cg.zkchange = nil
		return
	}

	// Fetch a list of partition IDs
	if pids, err = cg.client.Partitions(cg.topic); err != nil {
		cg.zkchange = nil
		return
	}

	// Get leaders for each partition ID
	parts := make(PartitionSlice, len(pids))
	for i, pid := range pids {
		var broker *sarama.Broker
		if broker, err = cg.client.Leader(cg.topic, pid); err != nil {
			cg.zkchange = nil
			return
		}
		defer broker.Close()
		parts[i] = Partition{Id: pid, Addr: broker.Addr()}
	}

	if err = cg.makeClaims(cids, parts); err != nil {
		cg.zkchange = nil
		cg.releaseClaims()
		return
	}
	return
}

// commitOffset sends an offset message to kafka for the given consumer group
func commitOffset(broker *sarama.Broker, topic string, partition int32, group string, offset int64, generationID int32, memberID string) {

	v := int16(0)
	if config.offset.version.IsAtLeast(sarama.V0_8_2_0) {
		v = 1
	}
	if config.offset.version.IsAtLeast(sarama.V0_9_0_0) {
		v = 2
	}

	req := &sarama.OffsetCommitRequest{
		Version:                 v,
		ConsumerGroup:           group,
		ConsumerGroupGeneration: generationID,
		ConsumerID:              memberID,
		RetentionTime:           -1,
	}
	req.AddBlock(topic, partition, offset, 0, "")

	offsetResp, err := broker.CommitOffset(req)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to commit offsets. err=%v\n", err)
		os.Exit(1)
	} else if len(offsetResp.Errors) > 0 {
		for topic, perrs := range offsetResp.Errors {
			for partition, kerr := range perrs {
				if kerr != sarama.ErrNoError {
					fmt.Fprintf(os.Stderr, "Failed to commit offsets topic=%s, partition=%s. err=%v\n", topic, partition, err)
					os.Exit(1)
				}
			}
		}
	}
}

// joinGroup joins a consumer group and returns the group memberID and generationID
func joinGroup(broker *sarama.Broker, group string, topic string) (string, int32) {
	joinGroupReq := &sarama.JoinGroupRequest{
		GroupId:        group,
		SessionTimeout: int32((30 * time.Second) / time.Millisecond),
		ProtocolType:   "consumer",
	}

	meta := &sarama.ConsumerGroupMemberMetadata{
		Version: 1,
		Topics:  []string{topic},
	}
	err := joinGroupReq.AddGroupProtocolMetadata("range", meta)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to add meta data err=%v\n", err)
		os.Exit(1)
	}
	err = joinGroupReq.AddGroupProtocolMetadata("roundrobin", meta)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to add meta data err=%v\n", err)
		os.Exit(1)
	}

	resp, err := broker.JoinGroup(joinGroupReq)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to join consumer group err=%v\n", err)
		os.Exit(1)
	} else if resp.Err != sarama.ErrNoError {
		fmt.Fprintf(os.Stderr, "Failed to join consumer group err=%v\n", resp.Err)
		os.Exit(1)
	}

	return resp.MemberId, resp.GenerationId
}

func queryMetadataWithRetry(
	b *sarama.Broker,
	cfg *sarama.Config,
	topics []string,
) (r *sarama.MetadataResponse, err error) {
	err = withRetry(b, cfg, func() (e error) {
		r, e = b.GetMetadata(&sarama.MetadataRequest{topics})
		return
	})
	return
}

func queryOffset(
	b *sarama.Broker,
	replicaID int32,
	topic string,
	partition int32,
	time int64,
) (int64, bool, error) {
	req := &sarama.OffsetRequest{}
	if replicaID != noID {
		req.SetReplicaID(replicaID)
	}
	req.AddBlock(topic, partition, time, 1)
	resp, err := b.GetAvailableOffsets(req)
	if err != nil {
		return -1, false, err
	}

	block := resp.GetBlock(topic, partition)
	if len(block.Offsets) == 0 {
		return -1, false, nil
	}

	return block.Offsets[0], true, nil
}

// Connects to the broker that handles all produce and consume
// requests for the given chain (Partition Leader Replica)
func newBroker(brokers []string, cp ChainPartition) (Broker, error) {
	var candidateBroker, connectedBroker, leaderBroker *sarama.Broker

	// Connect to one of the given brokers
	for _, hostPort := range brokers {
		candidateBroker = sarama.NewBroker(hostPort)
		if err := candidateBroker.Open(nil); err != nil {
			logger.Warningf("Failed to connect to broker %s: %s", hostPort, err)
			continue
		}
		if connected, err := candidateBroker.Connected(); !connected {
			logger.Warningf("Failed to connect to broker %s: %s", hostPort, err)
			continue
		}
		connectedBroker = candidateBroker
		break
	}

	if connectedBroker == nil {
		return nil, fmt.Errorf("Failed to connect to any of the given brokers (%v) for metadata request", brokers)
	}
	logger.Debugf("Connected to broker %s", connectedBroker.Addr())

	// Get metadata for the topic that corresponds to this chain
	metadata, err := connectedBroker.GetMetadata(&sarama.MetadataRequest{Topics: []string{cp.Topic()}})
	if err != nil {
		return nil, fmt.Errorf("Failed to get metadata for topic %s: %s", cp, err)
	}

	// Get the leader broker for this chain partition
	if (cp.Partition() >= 0) && (cp.Partition() < int32(len(metadata.Topics[0].Partitions))) {
		leaderBrokerID := metadata.Topics[0].Partitions[cp.Partition()].Leader
		logger.Debugf("Leading broker for chain %s is broker ID %d", cp, leaderBrokerID)
		for _, availableBroker := range metadata.Brokers {
			if availableBroker.ID() == leaderBrokerID {
				leaderBroker = availableBroker
				break
			}
		}
	}

	if leaderBroker == nil {
		return nil, fmt.Errorf("Can't find leader for chain %s", cp)
	}

	// Connect to broker
	if err := leaderBroker.Open(nil); err != nil {
		return nil, fmt.Errorf("Failed to connect ho Kafka broker: %s", err)
	}
	if connected, err := leaderBroker.Connected(); !connected {
		return nil, fmt.Errorf("Failed to connect to Kafka broker: %s", err)
	}

	return &brokerImpl{broker: leaderBroker}, nil
}

func closeBroker(b *sarama.Broker) {
	if ok, _ := b.Connected(); ok {
		b.Close()
	}
}

func brokerAddress(b *sarama.Broker) string {
	return strings.ToLower(b.Addr())
}