[ENG-43320] Fix InputStream leak in PresignedUrlFileUploader + error-path tests

[vishalk9]

Summary

The small-file branch of PresignedUrlFileUploader#getRequest called IOUtils.toByteArray(fileStreamData.getInputStream()) without closing the stream, leaking file descriptors / underlying HTTP connections from S3/GCS/Azure SDKs on every upload. The large-file streaming branch already does it right via try-with-resources — this PR aligns the small-file branch.

Then adds the missing error-path tests (the existing class only covered happy, 5xx-failure, and large-file paths — 3 tests / 177 LOC).

Code change (PresignedUrlFileUploader.java)

Before:

if (fileStreamData.getFileSize() <= fileUploadStreamBatchSize) {
  RequestBody requestBody;
  try {
    requestBody = RequestBody.create(mediaType, IOUtils.toByteArray(fileStreamData.getInputStream()));
    request = new Request.Builder().url(presignedUrl).put(requestBody).build();
  } catch (IOException e) {
    throw new FileUploadException(e);
  }
}

After:

if (fileStreamData.getFileSize() <= fileUploadStreamBatchSize) {
  RequestBody requestBody;
  try (InputStream is = fileStreamData.getInputStream()) {
    requestBody = RequestBody.create(mediaType, IOUtils.toByteArray(is));
    request = new Request.Builder().url(presignedUrl).put(requestBody).build();
  } catch (IOException e) {
    throw new FileUploadException(e);
  }
}

Tests added (PresignedUrlFileUploaderTest.java)

1. testUploadFileToPresignedUrl_smallFileClosesInputStreamOnSuccess — wraps the test InputStream in a Mockito spy and asserts verify(inputStream).close() after a successful upload.
2. testUploadFileToPresignedUrl_smallFileClosesInputStreamOnReadFailure — uses an InputStream whose read(...) throws IOException; asserts the wrapped FileUploadException surfaces with the original IOException cause AND that close() was still called.
3. testUploadFileToPresignedUrl_acceptable4xxNotRetried — MockWebServer returns 403 (acceptable failure code per ACCEPTABLE_HTTP_FAILURE_STATUS_CODES); asserts getRequestCount() == 1 to prove retry was skipped even with maxRetries=3, AND that the failure metric is emitted.

Refactored @BeforeEach to move the streamFileAsync stub into a stubStreamFile(InputStream, long) helper so the new error-path tests can supply their own InputStream without tripping strict-stubbing on the existing 3 tests.

Why this matters

• Long-running data-plane extractor pods leak file descriptors over time on every small-file upload — eventually Too many open files if the leak compounds against the S3/GCS/Azure SDK's internal HTTP connection pool.
• The retry path on the acceptable-4xx code was untested. A future regression that fails to short-circuit on ACCEPTABLE_HTTP_FAILURE_STATUS_CODES would silently inflate API spend (3× requests for every expired/invalid pre-signed URL).

Regression risk

Zero. The only behavior change in the production code is that the InputStream from the small-file path now closes after the byte buffer is materialised. Nothing else uses the InputStream after getRequest returns — IOUtils.toByteArray has already drained it into a byte[] that backs the RequestBody. The streaming large-file path is unchanged.

Test plan

• ./gradlew :lakeview:test --tests PresignedUrlFileUploaderTest — all 6 tests green (3 existing + 3 new) locally.
• ./gradlew :lakeview:test --tests "ai.onehouse.storage.*" — full storage package green.
• ./gradlew :lakeview:spotlessCheck — green.
• CI: build, sonar, coverage gates.

Background code-flow doc for the metadata-extractor data-plane Push-Model job lives at 10XEngineer/docs/lakeview-metadata-extractor-flow.md (just merged).

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud