code: @Override
  	void releaseAllResources() throws IOException {
  		if (isReleased.compareAndSet(false, true)) {
  
  			// Gather all exclusive buffers and recycle them to global pool in batch, because
  			// we do not want to trigger redistribution of buffers after each recycle.
  			final List<MemorySegment> exclusiveRecyclingSegments = new ArrayList<>();
  
  			synchronized (receivedBuffers) {
  				Buffer buffer;
  				while ((buffer = receivedBuffers.poll()) != null) {
  					if (buffer.getRecycler() == this) {
  						exclusiveRecyclingSegments.add(buffer.getMemorySegment());
  					} else {
  						buffer.recycleBuffer();
  					}
  				}
  			}
  			synchronized (bufferQueue) {
  				bufferQueue.releaseAll(exclusiveRecyclingSegments);
  			}
  
  			if (exclusiveRecyclingSegments.size() > 0) {
  				inputGate.returnExclusiveSegments(exclusiveRecyclingSegments);
  			}
  
  			// The released flag has to be set before closing the connection to ensure that
  			// buffers received concurrently with closing are properly recycled.
  			if (partitionRequestClient != null) {
  				partitionRequestClient.close(this);
  			} else {
  				connectionManager.closeOpenChannelConnections(connectionId);
  			}
  		}
  	}
docs: Releases all exclusive and floating buffers, closes the partition request client.