// Partitions returns a list of all partitions for the topic.
328034
func (t *Topic) Partitions() (PartitionList, error) {
	value, _, err := t.kz.conn.Get(t.metadataPath())
	if err != nil {
		return nil, err
	}

	return t.parsePartitions(value)
}
1101065
func partitionIds(maxIds int, ids []string) [][]string {
	index := 0
	total := len(ids)
	var partitions [][]string
	for remaining := total - index; remaining > 0; remaining = total - index {
		if remaining < maxIds {
			partitions = append(partitions, ids[index:])
			break
		} else {
			partitions = append(partitions, ids[index:index+maxIds])
			index += maxIds
		}
	}

	return partitions
}
192225
func (c Client) GetPartitions(appName, serviceName string) (*PartitionItemsPage, error) {
	var aggregatePartitionItemsPages PartitionItemsPage
	var continueToken string
	for {
		basePath := "Applications/" + appName + "/$/GetServices/" + serviceName + "/$/GetPartitions/"
		res, err := c.getHTTP(basePath, withContinue(continueToken))
		if err != nil {
			return nil, err
		}

		var partitionsItemsPage PartitionItemsPage
		err = json.Unmarshal(res, &partitionsItemsPage)
		if err != nil {
			return nil, fmt.Errorf("could not deserialise JSON response: %+v", err)
		}

		aggregatePartitionItemsPages.Items = append(aggregatePartitionItemsPages.Items, partitionsItemsPage.Items...)

		continueToken = getString(partitionsItemsPage.ContinuationToken)
		if continueToken == "" {
			break
		}
	}
	return &aggregatePartitionItemsPages, nil
}
781462
func (rs Ranges) Partition(n uint64) (Ranges, bool) {
	i := rs.Search(n)
	if i < 0 {
		return rs, false
	}

	pn := make(Ranges, 0, len(rs)+1)
	switch pn = append(pn, rs[:i]...); {
	case rs[i].Begin == rs[i].End: // delete
	case rs[i].Begin == n: // increment lower bound
		pn = append(pn, Value_Range{rs[i].Begin + 1, rs[i].End})
	case rs[i].End == n: // decrement upper bound
		pn = append(pn, Value_Range{rs[i].Begin, rs[i].End - 1})
	default: // split
		pn = append(pn, Value_Range{rs[i].Begin, n - 1}, Value_Range{n + 1, rs[i].End})
	}
	return append(pn, rs[i+1:]...), true
}
1044732
func newCPartsFromTopicPartitions(partitions []TopicPartition) (cparts *C.rd_kafka_topic_partition_list_t) {
	cparts = C.rd_kafka_topic_partition_list_new(C.int(len(partitions)))
	for _, part := range partitions {
		ctopic := C.CString(*part.Topic)
		defer C.free(unsafe.Pointer(ctopic))
		rktpar := C.rd_kafka_topic_partition_list_add(cparts, ctopic, C.int32_t(part.Partition))
		rktpar.offset = C.int64_t(part.Offset)
	}

	return cparts
}
1247679
func (n *HTTPNode) SetPartitions(part []int32) {
	n.Partitions = part
	n.Updated = time.Now()
}
125744
func (t *Topic) parsePartitions(value []byte) (PartitionList, error) {
	var tm topicMetadata
	if err := json.Unmarshal(value, &tm); err != nil {
		return nil, err
	}

	result := make(PartitionList, len(tm.Partitions))
	for partitionNumber, replicas := range tm.Partitions {
		partitionID, err := strconv.ParseInt(partitionNumber, 10, 32)
		if err != nil {
			return nil, err
		}

		replicaIDs := make([]int32, 0, len(replicas))
		for _, r := range replicas {
			replicaIDs = append(replicaIDs, int32(r))
		}
		result[partitionID] = t.Partition(int32(partitionID), replicaIDs)
	}

	return result, nil
}
217907
func findPartitions(topic string, partitions []Partition) []int32 {
	var ids []int32
	for _, partition := range partitions {
		if partition.Topic == topic {
			ids = append(ids, int32(partition.ID))
		}
	}
	return ids
}
931038
func (p *Partition) state() (partitionState, error) {
	var state partitionState
	node := fmt.Sprintf("%s/brokers/topics/%s/partitions/%d/state", p.topic.kz.conf.Chroot, p.topic.Name, p.ID)
	value, _, err := p.topic.kz.conn.Get(node)
	if err != nil {
		return state, err
	}

	if err := json.Unmarshal(value, &state); err != nil {
		return state, err
	}

	return state, nil
}
267153
func partition(s Sortable, from, to int) int {
	i := from
	for j := from; j < to-1; j++ {
		if s.Less(j, to-1) {
			s.Swap(i, j)
			i++
		}
	}
	s.Swap(to-1, i)
	return i
}
1258361
func (d *Dataset) Partition(name string, shard int, sortOption *SortOption) *Dataset {
	indexes := sortOption.Indexes()
	if intArrayEquals(d.IsPartitionedBy, indexes) && shard == len(d.Shards) {
		return d
	}
	if 1 == len(d.Shards) && shard == 1 {
		return d
	}
	ret := d.partition_scatter(name, shard, indexes)
	if shard > 1 {
		ret = ret.partition_collect(name, shard, indexes)
	}
	ret.IsPartitionedBy = indexes
	return ret
}
206237
func (pr *partitionRange) values() []map[string]dosa.FieldValue {
	return pr.entityRef[pr.partitionKey][pr.start : pr.end+1]
}
384525
func (s *Table) SetPartitionKeys(v []*Column) *Table {
	s.PartitionKeys = v
	return s
}
389604
func (n *Nodes) Allocations(nodeID string, q *QueryOptions) ([]*Allocation, *QueryMeta, error) {
	var resp []*Allocation
	qm, err := n.client.query("/v1/node/"+nodeID+"/allocations", &resp, q)
	if err != nil {
		return nil, nil, err
	}
	sort.Sort(AllocationSort(resp))
	return resp, qm, nil
}
548536
func (d *Dialer) LookupPartitions(ctx context.Context, network string, address string, topic string) ([]Partition, error) {
	conn, err := d.DialContext(ctx, network, address)
	if err != nil {
		return nil, err
	}
	defer conn.Close()

	prtch := make(chan []Partition, 1)
	errch := make(chan error, 1)

	go func() {
		if prt, err := conn.hideReadPartitions(topic); err != nil {
			errch <- err
		} else {
			prtch <- prt
		}
	}()

	var prt []Partition
	select {
	case prt = <-prtch:
	case err = <-errch:
	case <-ctx.Done():
		err = ctx.Err()
	}
	return prt, err
}
594766
func (c *Consumer) Assignment() (partitions []TopicPartition, err error) {
	var cParts *C.rd_kafka_topic_partition_list_t

	cErr := C.rd_kafka_assignment(c.handle.rk, &cParts)
	if cErr != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return nil, newError(cErr)
	}
	defer C.rd_kafka_topic_partition_list_destroy(cParts)

	partitions = newTopicPartitionsFromCparts(cParts)

	return partitions, nil
}
1004959
func (e *EntityDefinition) PartitionKeySet() map[string]struct{} {
	m := make(map[string]struct{})
	for _, p := range e.Key.PartitionKeys {
		m[p] = struct{}{}
	}
	return m
}