void completeRead() {
      if (readIntoCallbackCalled.compareAndSet(false, true)) {
        Exception e = operationException.get();
        if (e instanceof RouterException) {
          RouterException routerException = (RouterException) e;
          if (routerException.getErrorCode() == RouterErrorCode.UnexpectedInternalError) {
            // Client deserves to know why there is an unexpected internal error
            e = new RestServiceException(routerException.getMessage(), RestServiceErrorCode.InternalServerError, true,
                false, null);
          }
        }
        readIntoFuture.done(bytesWritten.get(), e);
        if (readIntoCallback != null) {
          readIntoCallback.onCompletion(bytesWritten.get(), e);
        }
        if (e == null) {
          updateChunkingAndSizeMetricsOnSuccessfulGet();
        } else {
          logger.warn(
              "GetBlobOperationError BlobId: {}, numChunksRetrieved:{}, numChunksWrittenOut: {}. Time since last chunk write done: {}ms",
              blobId, numChunksRetrieved, numChunksWrittenOut,
              SystemTime.getInstance().milliseconds() - lastChunkWrittenDoneTime.get(), e);
          routerMetrics.onGetBlobError(e, options, isEncrypted);
        }
        long totalTime = time.milliseconds() - submissionTimeMs;
        if (isEncrypted) {
          routerMetrics.getEncryptedBlobOperationTotalTimeMs.update(totalTime);
        } else {
          routerMetrics.getBlobOperationTotalTimeMs.update(totalTime);
        }
      }
      setOperationCompleted();
    }
Complete the read from this {@link ReadableStreamChannel} by invoking the callback and marking the future.
ambry-router/src/main/java/com/github/ambry/router/GetBlobOperation.java