code: func (conn *BytesManualConnectionStr) StartPostInitConsumer(topic string, partition int32, offset int64) (*sarama.PartitionConsumer, error) {
  	multiplexer := conn.multiplexer
  	multiplexer.WithFields(logging.Fields{"topic": topic}).Debugf("Post-init consuming started")
  
  	if multiplexer.Consumer == nil || multiplexer.Consumer.SConsumer == nil {
  		multiplexer.Warn("Unable to start post-init Consumer, client not available in the mux")
  		return nil, nil
  	}
  
  	// Consumer that reads topic/partition/offset. Throws error if offset is 'in the future' (message with offset does not exist yet)
  	partitionConsumer, err := multiplexer.Consumer.SConsumer.ConsumePartition(topic, partition, offset)
  	if err != nil {
  		return nil, err
  	}
  	multiplexer.Consumer.StartConsumerManualHandlers(partitionConsumer)
  
  	return &partitionConsumer, nil
  }
docs: // StartPostInitConsumer allows to start a new partition consumer after mux is initialized