// LookupPartitions returns the list of partitions that exist for the given topic.
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
}
445369
func (i *Index) partition(key []byte) *Partition {
	return i.partitions[int(xxhash.Sum64(key)&(i.PartitionN-1))]
}
254706
func (h *Handle) GetPartitionStats(tblInfo *model.TableInfo, pid int64) *statistics.Table {
	tbl, ok := h.StatsCache.Load().(StatsCache)[pid]
	if !ok {
		tbl = statistics.PseudoTable(tblInfo)
		tbl.PhysicalID = pid
		h.UpdateTableStats([]*statistics.Table{tbl}, nil)
		return tbl
	}
	return tbl
}
68060
func (m *Multi) ListParts(ctx context.Context) ([]Part, error) {
	params := map[string][]string{
		"uploadId":  {m.UploadID},
		"max-parts": {strconv.FormatInt(int64(listPartsMax), 10)},
	}
	var parts partSlice
	for attempt := m.Bucket.S3.AttemptStrategy.Start(); attempt.Next(); {
		req := &request{
			method: "GET",
			bucket: m.Bucket.Name,
			path:   m.Key,
			params: params,
		}
		var resp listPartsResp
		err := m.Bucket.S3.query(ctx, req, &resp)
		if shouldRetry(err) && attempt.HasNext() {
			continue
		}
		if err != nil {
			return nil, err
		}
		parts = append(parts, resp.Part...)
		if !resp.IsTruncated {
			sort.Sort(parts)
			return parts, nil
		}
		params["part-number-marker"] = []string{resp.NextPartNumberMarker}
		attempt = m.Bucket.S3.AttemptStrategy.Start() // Last request worked.
	}
	panic("unreachable")
}
895837
func (s *DimensionKeyDescription) SetPartitions(v []*float64) *DimensionKeyDescription {
	s.Partitions = v
	return s
}
860626
func (s *DescribeDimensionKeysOutput) SetPartitionKeys(v []*ResponsePartitionKey) *DescribeDimensionKeysOutput {
	s.PartitionKeys = v
	return s
}
952689
func newStaticPartitioner(partition int32) sarama.PartitionerConstructor {
	return func(topic string) sarama.Partitioner {
		return &staticPartitioner{partition}
	}
}
781297
func (c *KafkaConsumer) Add(topic string, partition int32) error {
	c.partitionConsumersLock.Lock()
	defer c.partitionConsumersLock.Unlock()

	if _, exists := c.partitionConsumers[topic]; !exists {
		c.partitionConsumers[topic] = make(map[int32]PartitionConsumer)
	}

	if _, exists := c.partitionConsumers[topic][partition]; exists {
		log.Info("Partition consumer for topic %s, partition %d already exists", topic, partition)
		return ErrPartitionConsumerAlreadyExists
	}

	c.metrics.NumOwnedTopicPartitions(func(numOwnedTopicPartitions metrics.Counter) {
		numOwnedTopicPartitions.Inc(1)
	})

	c.partitionConsumers[topic][partition] = c.partitionConsumerFactory(c.client, c.config, topic, partition, c.strategy)
	c.assignmentsWaitGroup.Add(1)
	go c.partitionConsumers[topic][partition].Start()
	return nil
}
291664
func GetAllTopics() []string {
	session := db.Conn.Session.Copy()
	defer session.Close()

	topics := []string{}

	db.Conn.GetCursor(
		session, db.AvailableTopicCollection, utils.M{},
	).Distinct("ident", &topics)

	return topics
}
745755
func FindPartitionByName(meta *model.TableInfo, parName string) (int64, error) {
	// Hash partition table use p0, p1, p2, p3 as partition names automatically.
	parName = strings.ToLower(parName)
	for _, def := range meta.Partition.Definitions {
		if strings.EqualFold(def.Name.L, parName) {
			return def.ID, nil
		}
	}
	return -1, errors.Trace(table.ErrUnknownPartition.GenWithStackByArgs(parName, meta.Name.O))
}
1017765
func partition(s sortEntries) int {
	hi := len(s) - 1
	pivot := s[hi]
	i := 0
	for j := 0; j < hi; j++ {
		r := bytes.Compare(s[j].Name, pivot.Name)
		if r <= 0 {
			s[i], s[j] = s[j], s[i]
			i++
		}
	}
	s[i], s[hi] = s[hi], s[i]
	return i
}
556145
func (tl TopicList) Find(name string) *Topic {
	for _, topic := range tl {
		if topic.Name == name {
			return topic
		}
	}
	return nil
}
762097
func (m *Manager) GetPartitionStateAsync(
	name string,
	partitionId string,
	userCredentials *cli.UserCredentials,
) *tasks.Task {
	if name == "" {
		panic("name must be present")
	}
	if partitionId == "" {
		panic("partitionId must be present")
	}

	return m.client.GetPartitionStateAsync(m.httpEndpoint, name, partitionId, userCredentials)
}
1073830
func (c *cluster) partition(index string, shard uint64) int {
	var buf [8]byte
	binary.BigEndian.PutUint64(buf[:], shard)

	// Hash the bytes and mod by partition count.
	h := fnv.New64a()
	_, _ = h.Write([]byte(index))
	_, _ = h.Write(buf[:])
	return int(h.Sum64() % uint64(c.partitionN))
}
366983
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
	}

	// Create an ephemeral node for the partition to claim the partition for this instance
	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 a separate error for this, to allow for implementing a retry mechanism.
			return ErrPartitionClaimedByOther
		}
		return nil
	default:
		return err
	}
}
1247679
func (n *HTTPNode) SetPartitions(part []int32) {
	n.Partitions = part
	n.Updated = time.Now()
}
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
}