fixup! refactor: fetch and catch parts of segment instead of chunks

Aiven-Open · Oct 2, 2023 · c765a28 · c765a28
1 parent 8bea380
commit c765a28
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 15 deletions.
diff --git a/core/src/main/java/io/aiven/kafka/tieredstorage/RemoteStorageManager.java b/core/src/main/java/io/aiven/kafka/tieredstorage/RemoteStorageManager.java
@@ -145,8 +145,7 @@ public void configure(final Map<String, ?> configs) {
         if (encryptionEnabled) {
             final Map<String, KeyPair> keyRing = new HashMap<>();
             config.encryptionKeyRing().forEach((keyId, keyPaths) ->
-                keyRing.put(keyId, RsaKeyReader.read(keyPaths.publicKey, keyPaths.privateKey)
-            ));
+                keyRing.put(keyId, RsaKeyReader.read(keyPaths.publicKey, keyPaths.privateKey)));
             rsaEncryptionProvider = new RsaEncryptionProvider(config.encryptionKeyPairId(), keyRing);
             aesEncryptionProvider = new AesEncryptionProvider();
         }

diff --git a/core/src/main/java/io/aiven/kafka/tieredstorage/fetch/FetchEnumeration.java b/core/src/main/java/io/aiven/kafka/tieredstorage/fetch/FetchEnumeration.java
@@ -57,11 +57,11 @@ public class FetchEnumeration implements Enumeration<InputStream> {
     final Chunk lastChunk;
 
     /**
-     * @param fetchManager  provides part input to fetch from
-     * @param objectKey required by chunkManager
-     * @param manifest      provides to index to build response from
-     * @param range         original offset range start/end position
-     * @param partSize      fetch part size
+     * @param fetchManager provides part input to fetch from
+     * @param objectKey    required by chunkManager
+     * @param manifest     provides to index to build response from
+     * @param range        original offset range start/end position
+     * @param partSize     fetch part size
      */
     public FetchEnumeration(final FetchManager fetchManager,
                             final ObjectKey objectKey,

diff --git a/core/src/main/java/io/aiven/kafka/tieredstorage/fetch/FetchPart.java b/core/src/main/java/io/aiven/kafka/tieredstorage/fetch/FetchPart.java
@@ -83,6 +83,10 @@ public Optional<FetchPart> next() {
         }
     }
 
+    public int startPosition() {
+        return chunks.get(0).originalPosition;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -102,10 +106,6 @@ public int hashCode() {
         return Objects.hash(firstChunkId, range, chunks);
     }
 
-    public int startPosition() {
-        return chunks.get(0).originalPosition;
-    }
-
     @Override
     public String toString() {
         return "Part{"

diff --git a/core/src/test/java/io/aiven/kafka/tieredstorage/fetch/cache/FetchCacheTest.java b/core/src/test/java/io/aiven/kafka/tieredstorage/fetch/cache/FetchCacheTest.java
@@ -113,15 +113,15 @@ void noEviction() throws IOException, StorageBackendException {
             final InputStream part0 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, firstPart);
             assertThat(part0).hasBinaryContent(CHUNK_0);
             verify(fetchManager).partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, firstPart);
-            final InputStream cachedpart0 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, firstPart);
-            assertThat(cachedpart0).hasBinaryContent(CHUNK_0);
+            final InputStream cachedPart0 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, firstPart);
+            assertThat(cachedPart0).hasBinaryContent(CHUNK_0);
             verifyNoMoreInteractions(fetchManager);
 
             final InputStream part1 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, nextPart);
             assertThat(part1).hasBinaryContent(CHUNK_1);
             verify(fetchManager).partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, nextPart);
-            final InputStream cachedpart1 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, nextPart);
-            assertThat(cachedpart1).hasBinaryContent(CHUNK_1);
+            final InputStream cachedPart1 = fetchCache.partContent(SEGMENT_OBJECT_KEY, SEGMENT_MANIFEST, nextPart);
+            assertThat(cachedPart1).hasBinaryContent(CHUNK_1);
             verifyNoMoreInteractions(fetchManager);
 
             verifyNoInteractions(removalListener);