// Partition returns a Partition instance for the topic.
497573
func (t *Topic) Partition(id int32, replicas []int32) *Partition {
	return &Partition{ID: id, Replicas: replicas, topic: t}
}
428920
func (p *partition) add(key []byte, entry *entry) {
	p.mu.Lock()
	p.store[string(key)] = entry
	p.mu.Unlock()
}
347465
func (t Topic) Split() (p, n string) {
	p, n, _ = t.SplitErr()
	return
}
902381
func (conn *ProtoManualConnection) ConsumePartition(msgClb func(messaging.ProtoMessage), topic string, partition int32, offset int64) error {
	conn.multiplexer.rwlock.Lock()
	defer conn.multiplexer.rwlock.Unlock()
	var err error

	// check if we have already consumed the topic on partition and offset
	var found bool
	var subs *consumerSubscription

	for _, subscription := range conn.multiplexer.mapping {
		if subscription.manual == false {
			// do not mix dynamic and manual mode
			continue
		}
		if subscription.topic == topic && subscription.partition == partition && subscription.offset == offset {
			found = true
			subs = subscription
			break
		}
	}

	byteClb := func(bm *client.ConsumerMessage) {
		pm := client.NewProtoConsumerMessage(bm, conn.serializer)
		msgClb(pm)
	}

	if !found {
		subs = &consumerSubscription{
			manual:         true, // manual example
			topic:          topic,
			partition:      partition,
			offset:         offset,
			connectionName: conn.name,
			byteConsMsg:    byteClb,
		}
		// subscribe new topic on partition
		conn.multiplexer.mapping = append(conn.multiplexer.mapping, subs)
	}

	// add subscription to consumerList
	subs.byteConsMsg = byteClb

	if conn.multiplexer.started {
		conn.multiplexer.Infof("Starting 'post-init' manual Consumer")
		subs.partitionConsumer, err = conn.StartPostInitConsumer(topic, partition, offset)
		if err != nil {
			return err
		}
		if subs.partitionConsumer == nil {
			return nil
		}
	}

	return nil
}
1169453
func (c *KafkaConsumer) PartitionConsumerMetrics(topic string, partition int32) (PartitionConsumerMetrics, error) {
	if !c.config.EnableMetrics {
		return nil, ErrMetricsDisabled
	}

	c.partitionConsumersLock.Lock()
	defer c.partitionConsumersLock.Unlock()

	if !c.exists(topic, partition) {
		log.Info("Partition consumer for topic %s, partition %d does not exist", topic, partition)
		return nil, ErrPartitionConsumerDoesNotExist
	}

	return c.partitionConsumers[topic][partition].Metrics()
}
1010813
func (sc scanCloser) Split(split bufio.SplitFunc) {
	sc.scanner.Split(split)
}
109457
func (p *Murmur2HashPartitioner) Partition(message *kafka.ProducerMessage, numPartitions int32) (int32, error) {
	if message.Key == nil {
		return p.random.Partition(message, numPartitions)
	}
	key, err := message.Key.Encode()
	if err != nil {
		return -1, err
	}
	// murmur2 implementation based on https://github.com/aappleby/smhasher/blob/master/src/MurmurHash2.cpp
	m := uint32(0x5bd1e995)
	r := uint32(24)
	seed := uint32(0x9747b28c)
	dataLen := uint32(len(key))
	/* Initialize the hash to a 'random' value */

	h := seed ^ dataLen

	/* Mix 4 bytes at a time into the hash */
	data := key
	i := 0
	for {
		if dataLen < 4 {
			break
		}
		k := *(*uint32)(unsafe.Pointer(&data[i*4]))
		k *= m
		k ^= k >> r
		k *= m

		h *= m
		h ^= k
		i++
		dataLen -= 4
	}

	/* Handle the last few bytes of the input array */

	switch dataLen {
	case 3:
		h ^= uint32(data[i*4+2]) << 16
		h ^= uint32(data[i*4+1]) << 8
		h ^= uint32(data[i*4])
		h *= m
	case 2:
		h ^= uint32(data[i*4+1]) << 8
		h ^= uint32(data[i*4])
		h *= m
	case 1:
		h ^= uint32(data[i*4])
		h *= m
	default:
	}

	/* Do a few final mixes of the hash to ensure the last few
	 * bytes are well-incorporated. */

	h ^= h >> 13
	h *= m
	h ^= h >> 15

	// convert h to positive
	partition := (int32(h) & 0x7fffffff) % numPartitions

	return partition, nil

}
588017
func (p *Partition) Open() error {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.closing = make(chan struct{})

	if p.opened {
		return errors.New("index partition already open")
	}

	// Validate path is correct.
	p.id = filepath.Base(p.path)
	_, err := strconv.Atoi(p.id)
	if err != nil {
		return err
	}

	// Create directory if it doesn't exist.
	if err := os.MkdirAll(p.path, 0777); err != nil {
		return err
	}

	// Read manifest file.
	m, manifestSize, err := ReadManifestFile(filepath.Join(p.path, ManifestFileName))
	if os.IsNotExist(err) {
		m = NewManifest(p.ManifestPath())
	} else if err != nil {
		return err
	}
	// Set manifest size on the partition
	p.manifestSize = manifestSize

	// Check to see if the MANIFEST file is compatible with the current Index.
	if err := m.Validate(); err != nil {
		return err
	}

	// Read stats file.
	if err := p.readStatsFile(); err != nil {
		return err
	}

	// Copy compaction levels to the index.
	p.levels = make([]CompactionLevel, len(m.Levels))
	copy(p.levels, m.Levels)

	// Set up flags to track whether a level is compacting.
	p.levelCompacting = make([]bool, len(p.levels))

	// Open each file in the manifest.
	var files []File
	for _, filename := range m.Files {
		switch filepath.Ext(filename) {
		case LogFileExt:
			f, err := p.openLogFile(filepath.Join(p.path, filename))
			if err != nil {
				return err
			}
			files = append(files, f)

			// Make first log file active, if within threshold.
			sz, _ := f.Stat()
			if p.activeLogFile == nil && sz < p.MaxLogFileSize {
				p.activeLogFile = f
			}

		case IndexFileExt:
			f, err := p.openIndexFile(filepath.Join(p.path, filename))
			if err != nil {
				return err
			}
			files = append(files, f)
		}
	}
	fs, err := NewFileSet(p.levels, p.sfile, files)
	if err != nil {
		return err
	}
	p.fileSet = fs

	// Set initial sequence number.
	p.seq = p.fileSet.MaxID()

	// Delete any files not in the manifest.
	if err := p.deleteNonManifestFiles(m); err != nil {
		return err
	}

	// Ensure a log file exists.
	if p.activeLogFile == nil {
		if err := p.prependActiveLogFile(); err != nil {
			return err
		}
	}

	// Build series existance set.
	if err := p.buildSeriesSet(); err != nil {
		return err
	}
	p.tracker.SetSeries(p.seriesIDSet.Cardinality())
	p.tracker.SetFiles(uint64(len(p.fileSet.IndexFiles())), "index")
	p.tracker.SetFiles(uint64(len(p.fileSet.LogFiles())), "log")
	p.tracker.SetDiskSize(uint64(p.fileSet.Size()))

	// Mark opened.
	p.opened = true

	// Send a compaction request on start up.
	p.compact()

	return nil
}
243823
func partition(data sort.Interface, lo, hi int) (int, int) {
	med := median(data, lo, hi)
	idx := lo
	data.Swap(med, hi)
	for i := lo; i < hi; i++ {
		if data.Less(i, hi) {
			data.Swap(i, idx)
			idx++
		}
	}
	data.Swap(idx, hi)
	return idx - 1, idx + 1
}
1322879
func (kz *Kazoo) Topic(topic string) *Topic {
	return &Topic{Name: topic, kz: kz}
}
47164
func (c *offsetCoordinator) Commit(topic string, partition int32, offset int64) error {
	return c.commit(topic, partition, offset, "")
}
1108426
func (r *ring) getPartition(key []byte) *partition {
	return r.partitions[int(xxhash.Sum64(key)%partitions)]
}
125286
func (p *bytesManualSyncPublisherKafka) Put(key string, data []byte) error {
	_, err := p.conn.SendSyncMessageToPartition(p.topic, p.partition, sarama.StringEncoder(key), sarama.ByteEncoder(data))
	return err
}
16078
func (s *Resource) SetPartition(v string) *Resource {
	s.Partition = &v
	return s
}
185691
func NewPartition(r io.Reader) (*Partition, error) {
	scanner := bufio.NewScanner(r)
	words := make(map[string]int)
	classes := make(map[int][]string)

	i := 0
	for scanner.Scan() {
		line := scanner.Text()

		fields := strings.Fields(line)
		if len(fields) != 2 {
			return nil, fmt.Errorf("[line: %d] expected 2 fields, instead got: %v", i+1, len(fields))
		}

		w := fields[0]
		c, err := strconv.Atoi(fields[1])
		if err != nil {
			return nil, fmt.Errorf("[line: %d] error parsing integer %#v: %v", i+1, fields[1], err)
		}

		words[w] = c
		classes[c] = append(classes[c], w)
		i++
	}

	err := scanner.Err()
	if err != nil {
		return nil, fmt.Errorf("[line: %d] scanner error: %v", i+1, err)
	}

	return &Partition{
		words:   words,
		classes: classes,
	}, nil
}
291898
func (ptn *Partition) String() string {
	return fmt.Sprintf("%s:%d", ptn.Namespace, ptn.PartitionId)
}
477169
func readPartition(reader io.Reader, size uint32) (p Partition, err error) {
	read := func(data interface{}) {
		if err == nil {
			err = binary.Read(reader, binary.LittleEndian, data)
		}
	}

	p.TrailingBytes = make([]byte, size-standardPartitionEntrySize)

	read(&p.Type)
	read(&p.Id)
	read(&p.FirstLBA)
	read(&p.LastLBA)
	read(&p.Flags)
	read(&p.PartNameUTF16)
	read(&p.TrailingBytes)

	return
}