code: @VisibleForTesting
  	@Override
  	public void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
  		if (partitionRequestClient == null) {
  			// Create a client and request the partition
  			partitionRequestClient = connectionManager
  				.createPartitionRequestClient(connectionId);
  
  			partitionRequestClient.requestSubpartition(partitionId, subpartitionIndex, this, 0);
  		}
  	}
docs: Requests a remote subpartition.