func (m *Model) CreatePartitionOwner(topic string, partition int32) error {
	path := m.partitionOwnerZNodePath(topic, partition)
	err := m.durableCreateZNode(path, []byte(m.memberID), zk.FlagEphemeral)
	if errors.Cause(err) == zk.ErrNodeExists {
		rawOwnerID, _, err := m.zkConn.Get(path)
		ownerID := string(rawOwnerID)
		if err != nil {
			return err
		}
		if ownerID != m.memberID {
			return errors.Errorf("owned by %s", ownerID)
		}
		return nil
	}
	if err != nil {
		return errors.Wrapf(err, "while deep creating %s", path)
	}
	return nil
}
CreatePartitionOwner creates a partition owner znode, but only if none
exists for the given topic-partition. An error is returned if a partition
owner znode exists but belongs to another member.
package kazoo
import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"
	"github.com/davecgh/go-spew/spew"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/pkg/errors"
	"github.com/samuel/go-zookeeper/zk"
	"github.com/sirupsen/logrus"
)
const (
	versionAny = -1
)
type Model struct {
	zkConn      *zk.Conn
	log         *logrus.Entry
	groupPath   string
	membersPath string
	ownersPath  string
	memberID    string
	memberPath  string
}
func NewModel(zkConn *zk.Conn, chroot, group, memberID string, log *logrus.Entry) Model {
	groupPath := fmt.Sprintf("%s/consumers/%s", chroot, group)
	membersPath := groupPath + "/ids"
	return Model{
		zkConn:      zkConn,
		log:         log,
		groupPath:   groupPath,
		membersPath: membersPath,
		ownersPath:  groupPath + "/owners",
		memberID:    memberID,
		memberPath:  membersPath + "/" + memberID,
	}
}
func (m *Model) EnsureMemberSubscription(topics []string) error {
	if len(topics) == 0 {
		if err := m.zkConn.Delete(m.memberPath, versionAny); err != nil {
			if err == zk.ErrNoNode {
				return nil
			}
			return errors.Wrapf(err, "while deleting %v", m.memberPath)
		}
		return nil
	}
	memberSpec := newMemberSpec(topics)
	memberSpecJSON, err := json.Marshal(memberSpec)
	if err != nil {
		return errors.Wrapf(err, "while JSON encoding %s", spew.Sdump(memberSpec))
	}
	if err := m.durableUpsertZNode(m.memberPath, memberSpecJSON, zk.FlagEphemeral); err != nil {
		return errors.WithStack(err)
	}
	return nil
}
func (m *Model) FetchGroupSubscriptions() (map[string][]string, <-chan none.T, context.CancelFunc, error) {
	members, memberWatchCh, err := m.watchZNodeChildren(m.membersPath)
	if err != nil {
		return nil, nil, nil, errors.Wrapf(err, "failed to watch members")
	}
	memberUpdateWatchChs := make(map[string]<-chan zk.Event, len(members))
	subscriptions := make(map[string][]string, len(members))
	for _, memberID := range members {
		memberPath := m.memberZNodePath(memberID)
		jsonMemberSpec, _, memberUpdateWatchCh, err := m.zkConn.GetW(memberPath)
		if err == zk.ErrNoNode {
			continue
		}
		if err != nil {
			return nil, nil, nil, errors.Wrapf(err, "while getting znode %s", memberPath)
		}
		var memberSpec memberSpec
		if err := json.Unmarshal(jsonMemberSpec, &memberSpec); err != nil {
			return nil, nil, nil, errors.Wrapf(err, "while parsing member %s, data=%s", memberID, string(jsonMemberSpec))
		}
		memberUpdateWatchChs[memberID] = memberUpdateWatchCh
		subscriptions[memberID] = memberSpec.topics()
	}
	aggregateWatchCh := make(chan none.T)
	ctx, cancel := context.WithCancel(context.Background())
	go m.forwardWatch(ctx, "members", memberWatchCh, aggregateWatchCh)
	for memberID, memberUpdateWatchCh := range memberUpdateWatchChs {
		go m.forwardWatch(ctx, memberID, memberUpdateWatchCh, aggregateWatchCh)
	}
	return subscriptions, aggregateWatchCh, cancel, nil
}