refactor: include original path on transform finisher

When producing an input stream, if no transformation is applied, then an input stream from the source file can be provided.
This will further be used to pass the path directly to the storage backend in a further PR.

Author: jeqo

core/src/main/java/io/aiven/kafka/tieredstorage/RemoteStorageManager.java

@@ -413,6 +413,7 @@ ChunkIndex uploadSegmentLog(
                     remoteLogSegmentMetadata.segmentSizeInBytes()
                 )
                 .withRateLimitingBucket(rateLimitingBucket)
+                .withOriginalFilePath(logSegmentData.logSegment())
                 .build();
 
             try (final var sis = transformFinisher.toInputStream()) {

core/src/main/java/io/aiven/kafka/tieredstorage/transform/TransformFinisher.java

@@ -17,10 +17,14 @@
 package io.aiven.kafka.tieredstorage.transform;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.SequenceInputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Enumeration;
 import java.util.Objects;
+import java.util.Optional;
 
 import io.aiven.kafka.tieredstorage.manifest.index.AbstractChunkIndexBuilder;
 import io.aiven.kafka.tieredstorage.manifest.index.ChunkIndex;
@@ -43,6 +47,8 @@
 public class TransformFinisher implements Enumeration<InputStream> {
     private final TransformChunkEnumeration inner;
     private final AbstractChunkIndexBuilder chunkIndexBuilder;
+    private final Path originalFilePath;
+    private final int originalFileSize;
     private ChunkIndex chunkIndex = null;
     private final Bucket rateLimitingBucket;
 
@@ -53,13 +59,16 @@ public static Builder newBuilder(final TransformChunkEnumeration inner, final in
     private TransformFinisher(
         final TransformChunkEnumeration inner,
         final boolean chunkingEnabled,
+        final Path originalFilePath,
         final int originalFileSize,
         final Bucket rateLimitingBucket
     ) {
         this.inner = Objects.requireNonNull(inner, "inner cannot be null");
 
         final int originalChunkSize = chunkingEnabled ? inner.originalChunkSize() : originalFileSize;
         this.chunkIndexBuilder = chunkIndexBuilder(inner, originalChunkSize, originalFileSize);
+        this.originalFilePath = originalFilePath;
+        this.originalFileSize = originalFileSize;
         this.rateLimitingBucket = rateLimitingBucket;
     }
 
@@ -102,24 +111,58 @@ public InputStream nextElement() {
 
     public ChunkIndex chunkIndex() {
         if (chunkIndex == null) {
-            throw new IllegalStateException("Chunk index was not built, was finisher used?");
+            if (isBaseTransform()) {
+                // as chunk index will not be built by consuming the input stream, calculate it from source file
+                return calculateChunkIndex();
+            } else {
+                throw new IllegalStateException("Chunk index was not built, was finisher used?");
+            }
         }
         return this.chunkIndex;
     }
 
-    public InputStream toInputStream() {
-        final SequenceInputStream sequencedInputStream = new SequenceInputStream(this);
-        if (rateLimitingBucket == null) {
-            return sequencedInputStream;
+    private ChunkIndex calculateChunkIndex() {
+        final var chunkSize = inner.transformedChunkSize();
+        var size = originalFileSize;
+        while (size > chunkSize) {
+            chunkIndexBuilder.addChunk(chunkSize);
+            size -= chunkSize;
+        }
+        return chunkIndexBuilder.finish(size);
+    }
+
+    public InputStream toInputStream() throws IOException {
+        if (isBaseTransform() && originalFilePath != null) {
+            // close inner input stream (based on source file)
+            final var sequenceInputStream = new SequenceInputStream(this);
+            sequenceInputStream.close();
+
+            return maybeToRateLimitedInputStream(Files.newInputStream(originalFilePath));
         } else {
-            return new RateLimitedInputStream(sequencedInputStream, rateLimitingBucket);
+            return maybeToRateLimitedInputStream(new SequenceInputStream(this));
         }
     }
 
+    private InputStream maybeToRateLimitedInputStream(final InputStream delegated) {
+        if (rateLimitingBucket == null) {
+            return delegated;
+        }
+        return new RateLimitedInputStream(delegated, rateLimitingBucket);
+    }
+
+    private boolean isBaseTransform() {
+        return inner instanceof BaseTransformChunkEnumeration;
+    }
+
+    Optional<Path> maybeOriginalFilePath() {
+        return Optional.ofNullable(originalFilePath);
+    }
+
     public static class Builder {
         final TransformChunkEnumeration inner;
         final Integer originalFileSize;
         boolean chunkingEnabled = true;
+        Path originalFilePath = null;
         Bucket rateLimitingBucket;
 
         public Builder(final TransformChunkEnumeration inner, final int originalFileSize) {
@@ -143,8 +186,14 @@ public Builder withChunkingDisabled() {
             return this;
         }
 
+        public Builder withOriginalFilePath(final Path originalFilePath) {
+            this.originalFilePath = Objects.requireNonNull(originalFilePath, "originalFilePath cannot be null");
+            return this;
+        }
+
         public TransformFinisher build() {
-            return new TransformFinisher(inner, chunkingEnabled, originalFileSize, rateLimitingBucket);
+            return new TransformFinisher(inner, chunkingEnabled, originalFilePath, originalFileSize,
+                rateLimitingBucket);
         }
     }
 }

core/src/test/java/io/aiven/kafka/tieredstorage/transform/TransformFinisherTest.java

@@ -17,6 +17,8 @@
 package io.aiven.kafka.tieredstorage.transform;
 
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Iterator;
 import java.util.List;
 
@@ -27,6 +29,7 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.Mock;
@@ -63,6 +66,34 @@ void negativeOriginalFileSize() {
             .hasMessage("originalFileSize must be non-negative, -1 given");
     }
 
+    @Test
+    void nullOriginalFilePath() {
+        assertThatThrownBy(() ->
+            TransformFinisher.newBuilder(inner, 100)
+                .withOriginalFilePath(null)
+                .build())
+            .isInstanceOf(NullPointerException.class)
+            .hasMessage("originalFilePath cannot be null");
+    }
+
+    @Test
+    void emptyOriginalFilePath() {
+        final var finisher = TransformFinisher.newBuilder(inner, 100).build();
+        assertThat(finisher.maybeOriginalFilePath()).isEmpty();
+    }
+
+    @Test
+    void presentOriginalFilePath(@TempDir final Path tmpDir) throws IOException {
+        final var originalFilePath = tmpDir.resolve("test.log");
+        Files.writeString(originalFilePath, "test");
+
+        final var finisher = TransformFinisher.newBuilder(inner, 100)
+            .withOriginalFilePath(originalFilePath)
+            .build();
+        assertThat(finisher.maybeOriginalFilePath()).isPresent();
+        assertThat(finisher.maybeOriginalFilePath().get()).hasContent("test");
+    }
+
     @ParameterizedTest
     @MethodSource("provideForBuildIndexAndReturnCorrectInputStreams")
     void buildIndexAndReturnCorrectInputStreams(final Integer transformedChunkSize,