feat: add upload from file to storage backend

Backend SDKs may have optimized ways to upload directly from files, instead of loading to memory as part of the plugin.

checkstyle/suppressions.xml

@@ -37,6 +37,7 @@
     <suppress checks="NPathComplexity" files="SingleBrokerTest.java"/>
     <suppress checks="JavaNCSSCheck" files="MetricsRegistry.java"/>
     <suppress checks="JavaNCSSCheck" files="RemoteStorageManagerMetricsTest.java"/>
+    <suppress checks="JavaNCSSCheck" files="S3StorageMetricsTest.java"/>
     <suppress checks="JavaNCSSCheck" files="SingleBrokerTest.java"/>
     <suppress checks="NPathComplexity" files="CredentialsBuilder.java"/>
 </suppressions>

storage/core/src/main/java/io/aiven/kafka/tieredstorage/storage/ObjectUploader.java

@@ -16,13 +16,31 @@
 
 package io.aiven.kafka.tieredstorage.storage;
 
+import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 
 public interface ObjectUploader {
     /**
      * @param inputStream content to upload. Not closed as part of the upload.
-     * @param key         path to an object within a storage backend.
+     * @param key         destintion path to an object within a storage backend.
      * @return number of bytes uploaded
      */
     long upload(InputStream inputStream, ObjectKey key) throws StorageBackendException;
+
+    /**
+     * Enable backend to use optimized uploading implementations based on source files
+     *
+     * @param path source path to the object to upload
+     * @param size size of the object to upload
+     * @param key  destination path to an object within a storage backend
+     */
+    default long upload(Path path, int size, ObjectKey key) throws StorageBackendException {
+        try (final var inputStream = Files.newInputStream(path)) {
+            return upload(inputStream, key);
+        } catch (IOException e) {
+            throw new StorageBackendException("Error uploading file from path: " + path, e);
+        }
+    }
 }

storage/s3/src/integration-test/java/io/aiven/kafka/tieredstorage/storage/s3/S3StorageMetricsTest.java

@@ -23,6 +23,9 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.management.ManagementFactory;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Map;
 import java.util.Set;
 
@@ -34,6 +37,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.testcontainers.containers.localstack.LocalStackContainer;
 import org.testcontainers.junit.jupiter.Container;
@@ -99,7 +103,7 @@ void setupStorage() {
     }
 
     @Test
-    void metricsShouldBeReported() throws Exception {
+    void metricsShouldBeReported(@TempDir final Path tmpDir) throws Exception {
         final byte[] data = new byte[PART_SIZE + 1];
 
         final ObjectKey key = new TestObjectKey("x");
@@ -214,5 +218,25 @@ void metricsShouldBeReported() throws Exception {
         assertThat(MBEAN_SERVER.getAttribute(s3ClientMetrics, "abort-multipart-upload-time-max"))
             .asInstanceOf(DOUBLE)
             .isGreaterThan(0.0);
+
+        final var tmpPath = tmpDir.resolve("test.log");
+        final var testContent = "test".getBytes(StandardCharsets.UTF_8);
+        Files.write(tmpPath, testContent);
+
+        final var anotherKey = new TestObjectKey("y");
+        storage.upload(tmpPath, testContent.length, anotherKey);
+
+        assertThat(MBEAN_SERVER.getAttribute(s3ClientMetrics, "put-object-requests-rate"))
+            .asInstanceOf(DOUBLE)
+            .isGreaterThan(0.0);
+        assertThat(MBEAN_SERVER.getAttribute(s3ClientMetrics, "put-object-requests-total"))
+            .asInstanceOf(DOUBLE)
+            .isEqualTo(1.0);
+        assertThat(MBEAN_SERVER.getAttribute(s3ClientMetrics, "put-object-time-avg"))
+            .asInstanceOf(DOUBLE)
+            .isGreaterThan(0.0);
+        assertThat(MBEAN_SERVER.getAttribute(s3ClientMetrics, "put-object-time-max"))
+            .asInstanceOf(DOUBLE)
+            .isGreaterThan(0.0);
     }
 }

storage/s3/src/main/java/io/aiven/kafka/tieredstorage/storage/s3/S3Storage.java

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.Path;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -37,6 +38,7 @@
 import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 
 public class S3Storage implements StorageBackend {
 
@@ -62,6 +64,13 @@ public long upload(final InputStream inputStream, final ObjectKey key) throws St
         }
     }
 
+    @Override
+    public long upload(final Path path, final int size, final ObjectKey key) throws StorageBackendException {
+        final var request = PutObjectRequest.builder().bucket(bucketName).key(key.value()).build();
+        s3Client.putObject(request, path);
+        return size;
+    }
+
     S3MultiPartOutputStream s3OutputStream(final ObjectKey key) {
         return new S3MultiPartOutputStream(bucketName, key, partSize, s3Client);
     }