code: 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
  }
docs: // ConsumePartition is called to start consuming given topic on partition with offset
  // Function can be called until the multiplexer is started, it returns an error otherwise.
  // The provided channel should be buffered, otherwise messages might be lost.