refactor: reuse cache config on chunk cache

Extend cache config with chunk cache configs (default and disk-based config) to reuse the configuration definitions.

core/src/main/java/io/aiven/kafka/tieredstorage/config/CacheConfig.java

@@ -31,10 +31,14 @@ public class CacheConfig extends AbstractConfig {
     private static final String CACHE_RETENTION_CONFIG = "retention.ms";
     private static final String CACHE_RETENTION_DOC = "Cache retention time ms, "
         + "where \"-1\" represents infinite retention";
-    private static final long DEFAULT_CACHE_RETENTION_MS = 600_000;
 
-    private static ConfigDef configDef(final Object defaultSize, final long defaultRetentionMs) {
-        final var configDef = new ConfigDef();
+    static final long CACHE_RETENTION_MS_DEFAULT = 600_000;
+
+    private static ConfigDef configDef(
+        final ConfigDef configDef,
+        final Object defaultSize,
+        final long defaultRetentionMs
+    ) {
         configDef.define(
             CACHE_SIZE_CONFIG,
             ConfigDef.Type.LONG,
@@ -54,12 +58,13 @@ private static ConfigDef configDef(final Object defaultSize, final long defaultR
         return configDef;
     }
 
-    private CacheConfig(
+    CacheConfig(
+        final ConfigDef configDef,
         final Map<String, ?> props,
         final Object defaultSize,
         final long defaultRetentionMs
     ) {
-        super(configDef(defaultSize, defaultRetentionMs), props);
+        super(configDef(configDef, defaultSize, defaultRetentionMs), props);
     }
 
     public static Builder newBuilder(final Map<String, ?> configs) {
@@ -84,7 +89,7 @@ public Optional<Duration> cacheRetention() {
 
     public static class Builder {
         private final Map<String, ?> props;
-        private long defaultRetentionMs = DEFAULT_CACHE_RETENTION_MS;
+        private long defaultRetentionMs = CACHE_RETENTION_MS_DEFAULT;
         private Object maybeDefaultSize = NO_DEFAULT_VALUE;
 
         public Builder(final Map<String, ?> props) {
@@ -102,7 +107,7 @@ public Builder withDefaultRetentionMs(final long retentionMs) {
         }
 
         public CacheConfig build() {
-            return new CacheConfig(props, maybeDefaultSize, defaultRetentionMs);
+            return new CacheConfig(new ConfigDef(), props, maybeDefaultSize, defaultRetentionMs);
         }
     }
 }

core/src/main/java/io/aiven/kafka/tieredstorage/config/ChunkCacheConfig.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2023 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.tieredstorage.config;
+
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigDef;
+
+public class ChunkCacheConfig extends CacheConfig {
+    private static final String CACHE_PREFETCH_MAX_SIZE_CONFIG = "prefetch.max.size";
+    private static final String CACHE_PREFETCH_MAX_SIZE_DOC =
+        "The amount of data that should be eagerly prefetched and cached";
+    private static final int CACHE_PREFETCHING_SIZE_DEFAULT = 0; //TODO find out what it should be
+
+    private static ConfigDef addCacheConfigs(final ConfigDef configDef) {
+        configDef.define(
+            CACHE_PREFETCH_MAX_SIZE_CONFIG,
+            ConfigDef.Type.INT,
+            CACHE_PREFETCHING_SIZE_DEFAULT,
+            ConfigDef.Range.between(0, Integer.MAX_VALUE),
+            ConfigDef.Importance.MEDIUM,
+            CACHE_PREFETCH_MAX_SIZE_DOC
+        );
+        return configDef;
+    }
+
+    public ChunkCacheConfig(final ConfigDef configDef, final Map<String, ?> props) {
+        super(addCacheConfigs(configDef), props, ConfigDef.NO_DEFAULT_VALUE, CacheConfig.CACHE_RETENTION_MS_DEFAULT);
+    }
+
+    public int cachePrefetchingSize() {
+        return getInt(CACHE_PREFETCH_MAX_SIZE_CONFIG);
+    }
+}

core/src/main/java/io/aiven/kafka/tieredstorage/fetch/cache/DiskChunkCacheConfig.java → core/src/main/java/io/aiven/kafka/tieredstorage/config/DiskChunkCacheConfig.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.aiven.kafka.tieredstorage.fetch.cache;
+package io.aiven.kafka.tieredstorage.config;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -75,11 +75,11 @@ final Path baseCachePath() {
         return Path.of(getString(CACHE_PATH_CONFIG));
     }
 
-    final Path cachePath() {
+    public final Path cachePath() {
         return baseCachePath().resolve(CACHE_DIRECTORY);
     }
 
-    final Path tempCachePath() {
+    public final Path tempCachePath() {
         return baseCachePath().resolve(TEMP_CACHE_DIRECTORY);
     }
 }

core/src/main/java/io/aiven/kafka/tieredstorage/fetch/cache/ChunkCache.java

@@ -29,6 +29,7 @@
 
 import org.apache.kafka.common.Configurable;
 
+import io.aiven.kafka.tieredstorage.config.ChunkCacheConfig;
 import io.aiven.kafka.tieredstorage.fetch.ChunkKey;
 import io.aiven.kafka.tieredstorage.fetch.ChunkManager;
 import io.aiven.kafka.tieredstorage.manifest.SegmentManifest;

core/src/main/java/io/aiven/kafka/tieredstorage/fetch/cache/DiskChunkCache.java

@@ -24,6 +24,7 @@
 
 import org.apache.kafka.common.utils.Time;
 
+import io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig;
 import io.aiven.kafka.tieredstorage.fetch.ChunkKey;
 import io.aiven.kafka.tieredstorage.fetch.ChunkManager;
 

core/src/main/java/io/aiven/kafka/tieredstorage/fetch/cache/MemoryChunkCache.java

@@ -23,6 +23,7 @@
 
 import org.apache.kafka.common.config.ConfigDef;
 
+import io.aiven.kafka.tieredstorage.config.ChunkCacheConfig;
 import io.aiven.kafka.tieredstorage.fetch.ChunkKey;
 import io.aiven.kafka.tieredstorage.fetch.ChunkManager;
 

core/src/test/java/io/aiven/kafka/tieredstorage/fetch/cache/ChunkCacheConfigTest.java → core/src/test/java/io/aiven/kafka/tieredstorage/config/ChunkCacheConfigTest.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.aiven.kafka.tieredstorage.fetch.cache;
+package io.aiven.kafka.tieredstorage.config;
 
 import java.time.Duration;
 import java.util.Map;

core/src/test/java/io/aiven/kafka/tieredstorage/fetch/cache/DiskChunkCacheConfigTest.java → core/src/test/java/io/aiven/kafka/tieredstorage/config/DiskChunkCacheConfigTest.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.aiven.kafka.tieredstorage.fetch.cache;
+package io.aiven.kafka.tieredstorage.config;
 
 import java.io.File;
 import java.io.IOException;
@@ -37,8 +37,8 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import static io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCacheConfig.CACHE_DIRECTORY;
-import static io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCacheConfig.TEMP_CACHE_DIRECTORY;
+import static io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig.CACHE_DIRECTORY;
+import static io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig.TEMP_CACHE_DIRECTORY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.params.provider.Arguments.arguments;

core/src/test/java/io/aiven/kafka/tieredstorage/fetch/cache/DiskChunkCacheTest.java

@@ -22,6 +22,7 @@
 import java.nio.file.Path;
 import java.util.Map;
 
+import io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig;
 import io.aiven.kafka.tieredstorage.fetch.ChunkKey;
 import io.aiven.kafka.tieredstorage.fetch.ChunkManager;
 
@@ -36,8 +37,8 @@
 import org.mockito.MockedStatic;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import static io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCacheConfig.CACHE_DIRECTORY;
-import static io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCacheConfig.TEMP_CACHE_DIRECTORY;
+import static io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig.CACHE_DIRECTORY;
+import static io.aiven.kafka.tieredstorage.config.DiskChunkCacheConfig.TEMP_CACHE_DIRECTORY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;