code: public void start() {
      if (blockIds.length == 0) {
        throw new IllegalArgumentException("Zero-sized blockIds array");
      }
  
      client.sendRpc(openMessage.toByteBuffer(), new RpcResponseCallback() {
        @Override
        public void onSuccess(ByteBuffer response) {
          try {
            streamHandle = (StreamHandle) BlockTransferMessage.Decoder.fromByteBuffer(response);
            logger.trace("Successfully opened blocks {}, preparing to fetch chunks.", streamHandle);
  
            // Immediately request all chunks -- we expect that the total size of the request is
            // reasonable due to higher level chunking in [[ShuffleBlockFetcherIterator]].
            for (int i = 0; i < streamHandle.numChunks; i++) {
              if (downloadFileManager != null) {
                client.stream(OneForOneStreamManager.genStreamChunkId(streamHandle.streamId, i),
                  new DownloadCallback(i));
              } else {
                client.fetchChunk(streamHandle.streamId, i, chunkCallback);
              }
            }
          } catch (Exception e) {
            logger.error("Failed while starting block fetches after success", e);
            failRemainingBlocks(blockIds, e);
          }
        }
  
        @Override
        public void onFailure(Throwable e) {
          logger.error("Failed while starting block fetches", e);
          failRemainingBlocks(blockIds, e);
        }
      });
    }
docs: Begins the fetching process, calling the listener with every block fetched.
  The given message will be serialized with the Java serializer, and the RPC must return a
  {@link StreamHandle}. We will send all fetch requests immediately, without throttling.