InputStream stream = uploadRequest.getStreamProvider().get();
    try {
      long streamPosition = 0;

      List<BufferedChunk> cache = new ArrayList<>();

      // Phase 2 & 3 Loop: Transmit Chunks & Query Recovery
      while (true) {
        try {
          checkDeadline(deadline);

          // Find chunk in cache or read from stream
          BufferedChunk chunk = null;
          for (BufferedChunk cached : cache) {
            if (cached.offset == offset) {
              chunk = cached;
              break;
            } 
          }

          if (chunk == null) {
            // Read from stream
            if (streamPosition != offset) {
              if (stream != null) {
                stream.close();
              }
              stream = uploadRequest.getStreamProvider().get();
              long skipped = skipFully(stream, offset);
              if (skipped < offset) {
                throw new IOException("Failed to skip stream bytes to offset: " + offset);
              }
              streamPosition = offset;
            }

            byte[] buffer = new byte[adjustedChunkSize];
            int bytesRead = readFully(stream, buffer, adjustedChunkSize);
            if (bytesRead > 0) {
              chunk = new BufferedChunk(offset, buffer, bytesRead);
              cache.add(chunk);
              if (cache.size() > 2) {
                cache.remove(0);
              }
              streamPosition += bytesRead;
            }
          }

          if (chunk == null) {
            // Stream was empty or exact chunk multiple and fully uploaded.
            // Send finalize only
            return sendFinalizeOnly(uploadUrl, offset, deadline);
          }

          // Check if this is the last chunk
          boolean isEof = (chunk.length < adjustedChunkSize);

          if (isEof) {
            // Send upload, finalize for the last chunk
            return sendUploadFinalize(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline);
          }

          // Send intermediate chunk (upload command)
          sendChunk(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline);

          // Successful chunk transmission! Update offset to next chunk
          offset = chunk.offset + chunk.length;
          attempt = 0; // Reset backoff attempts on progress

        } catch (UploadAlreadyFinalizedException uafe) {
          updateProgress(
              uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset,
              ResumableUploadProgressListener.State.COMPLETED);
          return (ResponseT) uafe.getResponse();
        } catch (Exception e) {
          checkDeadline(deadline);
          ErrorCategory category = getErrorCategory(e);

          if (category == ErrorCategory.CATEGORY_2_MISMATCH) {
            logger.log(Level.WARNING, "State mismatch detected. Triggering recovery...", e);
            updateProgress(offset, ResumableUploadProgressListener.State.RECOVERING);

            long serverOffset = 0;
            try {
              serverOffset = recoverOffset(uploadUrl, deadline);
            } catch (UploadAlreadyFinalizedException uafe) {
              updateProgress(
                  uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset,
                  ResumableUploadProgressListener.State.COMPLETED);
              return (ResponseT) uafe.getResponse();
            }

            logger.log(Level.INFO, "Recovery completed. Server received bytes: {0}", serverOffset);

            if (serverOffset == previousOffset) {
              attempt++;
              long delayMs = calculateBackoff(attempt);
              sleep(delayMs);
            } else {
              attempt = 0;
              previousOffset = serverOffset;
            }

            offset = serverOffset;
            // Loop will handle finding the chunk in cache or seeking/recreating the stream!
          } else if (category == ErrorCategory.CATEGORY_1_TRANSIENT) {
            attempt++;
            long delayMs = calculateBackoff(attempt);
            logger.log(
                Level.WARNING,
                "Transient error. Backing off for {0} ms (attempt {1})",
                new Object[] {delayMs, attempt});
            sleep(delayMs);
          } else {
            updateProgress(offset, ResumableUploadProgressListener.State.FAILED);
            throw e; // Fatal, bubble up
          }
        }
      }
    } finally {
      if (stream != null) {
        try {
          stream.close();
        } catch (IOException e) {
          logger.log(Level.WARNING, "Failed to close stream", e);
        }
      }
    }

  private void sleep(final long ms) throws InterruptedException {
    Thread.sleep(ms);
  }

    if (clientContext.getCredentials() != null) {
      com.google.api.gax.httpjson.InstantiatingHttpJsonChannelProvider.Builder httpJsonProviderBuilder =
          EchoStubSettings.defaultHttpJsonTransportProviderBuilder();
      if (settings.getEndpoint() != null) {
        httpJsonProviderBuilder.setEndpoint(settings.getEndpoint());
      }
      TransportChannelProvider httpJsonProvider = httpJsonProviderBuilder.build();
      ClientContext backgroundHttpContext =
          clientContext.withTransportChannelProvider(httpJsonProvider);
      this.httpJsonStub = new HttpJsonEchoStub(settings, backgroundHttpContext);
    } else {
      this.httpJsonStub = null;
    }

  @Override
  public void shutdown() {
    try {
      if (httpJsonStub != null) {
        httpJsonStub.shutdown();
      }
    } finally {
      backgroundResources.shutdown();
    }
  }