func (m *Marshaler) GetPartitionClaim(topicName string, partID int) PartitionClaim {
	topic := m.cluster.getPartitionState(m.groupID, topicName, partID)
	topic.lock.RLock()
	defer topic.lock.RUnlock()
	if topic.partitions[partID].claimed(m.cluster.ts) {
		return topic.partitions[partID]  
	}
	return PartitionClaim{}
}
GetPartitionClaim returns a PartitionClaim structure for a given partition. The structure
describes the consumer that is currently claiming this partition. This is a copy of the
claim structure, so changing it cannot change the world state.
package marshal
import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/zorkian/kafka"
	"github.com/zorkian/kafka/proto"
	"github.com/pborman/uuid"
)
const (
	MarshalTopic = "__marshal"
	HeartbeatInterval = 60  
)
type Marshaler struct {
	quit        *int32
	cluster     *KafkaCluster
	ownsCluster bool
	instanceID  string
	clientID    string
	groupID     string
	offsets     kafka.OffsetCoordinator
	lock      *sync.RWMutex
	consumers []*Consumer
}
func NewMarshaler(clientID, groupID string, brokers []string) (*Marshaler, error) {
	cluster, err := Dial("automatic", brokers, NewMarshalOptions())
	if err != nil {
		return nil, err
	}
	m, err := cluster.NewMarshaler(clientID, groupID)
	if err != nil {
		m.ownsCluster = true
	}
	return m, err
}
func newInstanceID() string {
	return uuid.New()[0:8]
}
func (m *Marshaler) addNewConsumer(c *Consumer) {
	m.lock.Lock()
	defer m.lock.Unlock()
	m.consumers = append(m.consumers, c)
}
func (m *Marshaler) removeConsumer(c *Consumer) {
	m.lock.Lock()
	defer m.lock.Unlock()
	for i, cn := range m.consumers {
		if cn == c {
			m.consumers = append(m.consumers[:i], m.consumers[i+1:]...)
			break
		}
	}
}
func (m *Marshaler) getClaimedPartitionState(topicName string, partID int) (
	*topicState, error) {
	topic := m.cluster.getPartitionState(m.groupID, topicName, partID)
	topic.lock.RLock()
	defer topic.lock.RUnlock()
	if !topic.partitions[partID].claimed(m.cluster.ts) {
		return nil, fmt.Errorf("Partition %s:%d is not claimed!", topicName, partID)
	}
	if topic.partitions[partID].GroupID != m.groupID ||
		topic.partitions[partID].ClientID != m.clientID {
		return nil, fmt.Errorf("Partition %s:%d is not claimed by us!", topicName, partID)
	}
	return topic, nil
}
func (m *Marshaler) Topics() []string {
	return m.cluster.getTopics()
}
func (m *Marshaler) Partitions(topicName string) int {
	return m.cluster.getTopicPartitions(topicName)
}
func (m *Marshaler) terminateAndCleanup(remove bool) {
	if !atomic.CompareAndSwapInt32(m.quit, 0, 1) {
		return
	}
	m.lock.Lock()
	defer m.lock.Unlock()
	for _, cn := range m.consumers {
		cn.terminateAndCleanup(false, false)
	}
	m.consumers = nil
	if m.ownsCluster {
		m.cluster.Terminate()
	}
	if remove {
		m.cluster.removeMarshal(m)
	}
}
func (m *Marshaler) Terminate() {
	m.terminateAndCleanup(true)
}
func (m *Marshaler) Terminated() bool {
	return atomic.LoadInt32(m.quit) == 1
}
func (m *Marshaler) Claimed(topicName string, partID int) bool {
	claim := m.GetPartitionClaim(topicName, partID)
	return claim.LastHeartbeat > 0
}