code: async def remove_pump_async(self, partition_id, reason):
          
          if partition_id in self.partition_pumps:
              captured_pump = self.partition_pumps[partition_id]
              if not captured_pump.is_closing():
                  await captured_pump.close_async(reason)
              # else, pump is already closing/closed, don't need to try to shut it down again
              del self.partition_pumps[partition_id]  # remove pump
              _logger.debug("Removed pump %r %r", self.host.guid, partition_id)
              _logger.debug("%r pumps still running", len(self.partition_pumps))
          else:
              # PartitionManager main loop tries to remove pump for every partition that the
              # host does not own, just to be sure. Not finding a pump for a partition is normal
              # and expected most of the time.
              _logger.debug("No pump found to remove for this partition %r %r",
                            self.host.guid, partition_id)
docs: Stops a single partiton pump.
  
          :param partition_id: The partition ID.
          :type partition_id: str
          :param reason: A reason for closing.
          :type reason: str