code: func (mux *Multiplexer) stopConsumingPartition(topic string, partition int32, offset int64, name string) error {
  	mux.rwlock.Lock()
  	defer mux.rwlock.Unlock()
  
  	var wasError error
  	var topicFound bool
  	// Remove consumer from subscription
  	for index, subs := range mux.mapping {
  		if subs.manual && subs.topic == topic && subs.partition == partition && subs.offset == offset && subs.connectionName == name {
  			topicFound = true
  			mux.mapping = append(mux.mapping[:index], mux.mapping[index+1:]...)
  		}
  		// Close the partition consumer related to the subscription
  		safeclose.Close(subs.partitionConsumer)
  	}
  	if !topicFound {
  		wasError = fmt.Errorf("topic %s, partition %v and offset %v was not consumed by '%s'",
  			topic, partition, offset, name)
  	}
  	// Stop partition consumer
  	return wasError
  }
docs: // Remove consumer subscription on given topic, partition and initial offset. If there is no such a subscription
  // (all fields must match), return error.