func (cg *Consumergroup) FetchAllOffsets() (map[string]map[int32]int64, error) {
	result := make(map[string]map[int32]int64)
	offsetsNode := fmt.Sprintf("%s/consumers/%s/offsets", cg.kz.conf.Chroot, cg.Name)
	topics, _, err := cg.kz.conn.Children(offsetsNode)
	if err == zk.ErrNoNode {
		return result, nil
	} else if err != nil {
		return nil, err
	}
	for _, topic := range topics {
		result[topic] = make(map[int32]int64)
		topicNode := fmt.Sprintf("%s/consumers/%s/offsets/%s", cg.kz.conf.Chroot, cg.Name, topic)
		partitions, _, err := cg.kz.conn.Children(topicNode)
		if err != nil {
			return nil, err
		}
		for _, partition := range partitions {
			partitionNode := fmt.Sprintf("%s/consumers/%s/offsets/%s/%s", cg.kz.conf.Chroot, cg.Name, topic, partition)
			val, _, err := cg.kz.conn.Get(partitionNode)
			if err != nil {
				return nil, err
			}
			partition, err := strconv.ParseInt(partition, 10, 32)
			if err != nil {
				return nil, err
			}
			offset, err := strconv.ParseInt(string(val), 10, 64)
			if err != nil {
				return nil, err
			}
			result[topic][int32(partition)] = offset
		}
	}
	return result, nil
}
FetchOffset retrieves all the commmitted offsets for a group
package kazoo
import (
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"strconv"
	"time"
	"github.com/samuel/go-zookeeper/zk"
)
var (
	ErrRunningInstances          = errors.New("Cannot deregister a consumergroup with running instances")
	ErrInstanceAlreadyRegistered = errors.New("Cannot register consumer instance because it already is registered")
	ErrInstanceNotRegistered     = errors.New("Cannot deregister consumer instance because it not registered")
	ErrPartitionClaimedByOther   = errors.New("Cannot claim partition: it is already claimed by another instance")
	ErrPartitionNotClaimed       = errors.New("Cannot release partition: it is not claimed by this instance")
)
type Consumergroup struct {
	kz   *Kazoo
	Name string
}
type ConsumergroupInstance struct {
	cg *Consumergroup
	ID string
}
type ConsumergroupList []*Consumergroup
type ConsumergroupInstanceList []*ConsumergroupInstance
type Registration struct {
	Pattern      RegPattern     `json:"pattern"`
	Subscription map[string]int `json:"subscription"`
	Timestamp    int64          `json:"timestamp"`
	Version      RegVersion     `json:"version"`
}
type RegPattern string
const (
	RegPatternStatic    RegPattern = "static"
	RegPatternWhiteList RegPattern = "white_list"
	RegPatternBlackList RegPattern = "black_list"
)
type RegVersion int
const (
	RegDefaultVersion RegVersion = 1
)
func (kz *Kazoo) Consumergroups() (ConsumergroupList, error) {
	root := fmt.Sprintf("%s/consumers", kz.conf.Chroot)
	cgs, _, err := kz.conn.Children(root)
	if err != nil {
		return nil, err
	}
	result := make(ConsumergroupList, 0, len(cgs))
	for _, cg := range cgs {
		result = append(result, kz.Consumergroup(cg))
	}
	return result, nil
}
func (kz *Kazoo) Consumergroup(name string) *Consumergroup {
	return &Consumergroup{Name: name, kz: kz}
}
func (cg *Consumergroup) Exists() (bool, error) {
	return cg.kz.exists(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
func (cg *Consumergroup) Create() error {
	return cg.kz.mkdirRecursive(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
func (cg *Consumergroup) Delete() error {
	if instances, err := cg.Instances(); err != nil {
		return err
	} else if len(instances) > 0 {
		return ErrRunningInstances
	}
	return cg.kz.deleteRecursive(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
func (cg *Consumergroup) Instances() (ConsumergroupInstanceList, error) {
	root := fmt.Sprintf("%s/consumers/%s/ids", cg.kz.conf.Chroot, cg.Name)
	if exists, err := cg.kz.exists(root); err != nil {
		return nil, err
	} else if exists {
		cgis, _, err := cg.kz.conn.Children(root)
		if err != nil {
			return nil, err
		}
		result := make(ConsumergroupInstanceList, 0, len(cgis))
		for _, cgi := range cgis {
			result = append(result, cg.Instance(cgi))
		}
		return result, nil
	} else {
		result := make(ConsumergroupInstanceList, 0)
		return result, nil
	}
}
func (cg *Consumergroup) WatchInstances() (ConsumergroupInstanceList, <-chan zk.Event, error) {
	node := fmt.Sprintf("%s/consumers/%s/ids", cg.kz.conf.Chroot, cg.Name)
	if exists, err := cg.kz.exists(node); err != nil {
		return nil, nil, err
	} else if !exists {
		if err := cg.kz.mkdirRecursive(node); err != nil {
			return nil, nil, err
		}
	}
	cgis, _, c, err := cg.kz.conn.ChildrenW(node)
	if err != nil {
		return nil, nil, err
	}
	result := make(ConsumergroupInstanceList, 0, len(cgis))
	for _, cgi := range cgis {
		result = append(result, cg.Instance(cgi))
	}
	return result, c, nil
}
func (cg *Consumergroup) NewInstance() *ConsumergroupInstance {
	id, err := generateConsumerInstanceID()
	if err != nil {
		panic(err)
	}
	return cg.Instance(id)
}
func (cg *Consumergroup) Instance(id string) *ConsumergroupInstance {
	return &ConsumergroupInstance{cg: cg, ID: id}
}
func (cg *Consumergroup) PartitionOwner(topic string, partition int32) (*ConsumergroupInstance, error) {
	node := fmt.Sprintf("%s/consumers/%s/owners/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	val, _, err := cg.kz.conn.Get(node)
	switch err {
	case nil:
		return &ConsumergroupInstance{cg: cg, ID: string(val)}, nil
	case zk.ErrNoNode:
		return nil, nil
	default:
		return nil, err
	}
}
func (cg *Consumergroup) WatchPartitionOwner(topic string, partition int32) (*ConsumergroupInstance, <-chan zk.Event, error) {
	node := fmt.Sprintf("%s/consumers/%s/owners/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	instanceID, _, changed, err := cg.kz.conn.GetW(node)
	switch err {
	case nil:
		return &ConsumergroupInstance{cg: cg, ID: string(instanceID)}, changed, nil
	case zk.ErrNoNode:
		return nil, nil, nil
	default:
		return nil, nil, err
	}
}
func (cgi *ConsumergroupInstance) Registered() (bool, error) {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	return cgi.cg.kz.exists(node)
}
func (cgi *ConsumergroupInstance) Registration() (*Registration, error) {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	val, _, err := cgi.cg.kz.conn.Get(node)
	if err != nil {
		return nil, err
	}
	reg := &Registration{}
	if err := json.Unmarshal(val, reg); err != nil {
		return nil, err
	}
	return reg, nil
}
func (cgi *ConsumergroupInstance) RegisterWithSubscription(subscriptionJSON []byte) error {
	if exists, err := cgi.Registered(); err != nil {
		return err
	} else if exists {
		return ErrInstanceAlreadyRegistered
	}
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	return cgi.cg.kz.create(node, subscriptionJSON, true)
}
func (cgi *ConsumergroupInstance) Register(topics []string) error {
	subscription := make(map[string]int)
	for _, topic := range topics {
		subscription[topic] = 1
	}
	data, err := json.Marshal(&Registration{
		Pattern:      RegPatternStatic,
		Subscription: subscription,
		Timestamp:    time.Now().Unix(),
		Version:      RegDefaultVersion,
	})
	if err != nil {
		return err
	}
	return cgi.RegisterWithSubscription(data)
}
func (cgi *ConsumergroupInstance) Deregister() error {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	exists, stat, err := cgi.cg.kz.conn.Exists(node)
	if err != nil {
		return err
	} else if !exists {
		return ErrInstanceNotRegistered
	}
	return cgi.cg.kz.conn.Delete(node, stat.Version)
}
func (cgi *ConsumergroupInstance) ClaimPartition(topic string, partition int32) error {
	root := fmt.Sprintf("%s/consumers/%s/owners/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, topic)
	if err := cgi.cg.kz.mkdirRecursive(root); err != nil {
		return err
	}
	node := fmt.Sprintf("%s/%d", root, partition)
	err := cgi.cg.kz.create(node, []byte(cgi.ID), true)
	switch err {
	case zk.ErrNodeExists:
		data, _, err := cgi.cg.kz.conn.Get(node)
		if err != nil {
			return err
		}
		if string(data) != cgi.ID {
			return ErrPartitionClaimedByOther
		}
		return nil
	default:
		return err
	}
}
func (cgi *ConsumergroupInstance) ReleasePartition(topic string, partition int32) error {
	owner, err := cgi.cg.PartitionOwner(topic, partition)
	if err != nil {
		return err
	}
	if owner == nil || owner.ID != cgi.ID {
		return ErrPartitionNotClaimed
	}
	node := fmt.Sprintf("%s/consumers/%s/owners/%s/%d", cgi.cg.kz.conf.Chroot, cgi.cg.Name, topic, partition)
	return cgi.cg.kz.conn.Delete(node, 0)
}
func (cg *Consumergroup) Topics() (TopicList, error) {
	root := fmt.Sprintf("%s/consumers/%s/owners", cg.kz.conf.Chroot, cg.Name)
	children, _, err := cg.kz.conn.Children(root)
	if err != nil {
		return nil, err
	}
	result := make(TopicList, 0, len(children))
	for _, name := range children {
		result = append(result, cg.kz.Topic(name))
	}
	return result, nil
}
func (cg *Consumergroup) CommitOffset(topic string, partition int32, offset int64) error {
	node := fmt.Sprintf("%s/consumers/%s/offsets/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	data := []byte(fmt.Sprintf("%d", offset))
	_, stat, err := cg.kz.conn.Get(node)
	switch err {
	case zk.ErrNoNode:  
		return cg.kz.create(node, data, false)
	case nil:  
		_, err := cg.kz.conn.Set(node, data, stat.Version)
		return err
	default:
		return err
	}
}
func (cg *Consumergroup) FetchOffset(topic string, partition int32) (int64, error) {
	node := fmt.Sprintf("%s/consumers/%s/offsets/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	val, _, err := cg.kz.conn.Get(node)
	if err == zk.ErrNoNode {
		return -1, nil
	} else if err != nil {
		return -1, err
	}
	return strconv.ParseInt(string(val), 10, 64)
}