code: void onSenderBacklog(int backlog) throws IOException {
  		int numRequestedBuffers = 0;
  
  		synchronized (bufferQueue) {
  			// Similar to notifyBufferAvailable(), make sure that we never add a buffer
  			// after releaseAllResources() released all buffers (see above for details).
  			if (isReleased.get()) {
  				return;
  			}
  
  			numRequiredBuffers = backlog + initialCredit;
  			while (bufferQueue.getAvailableBufferSize() < numRequiredBuffers && !isWaitingForFloatingBuffers) {
  				Buffer buffer = inputGate.getBufferPool().requestBuffer();
  				if (buffer != null) {
  					bufferQueue.addFloatingBuffer(buffer);
  					numRequestedBuffers++;
  				} else if (inputGate.getBufferProvider().addBufferListener(this)) {
  					// If the channel has not got enough buffers, register it as listener to wait for more floating buffers.
  					isWaitingForFloatingBuffers = true;
  					break;
  				}
  			}
  		}
  
  		if (numRequestedBuffers > 0 && unannouncedCredit.getAndAdd(numRequestedBuffers) == 0) {
  			notifyCreditAvailable();
  		}
  	}
docs: Receives the backlog from the producer's buffer response. If the number of available
  buffers is less than backlog + initialCredit, it will request floating buffers from the buffer
  pool, and then notify unannounced credits to the producer.
  
  @param backlog The number of unsent buffers in the producer's sub partition.