diff --git a/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/ChunkManagerFactoryConfig.java b/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/ChunkManagerFactoryConfig.java
index 2f044e69e..ad7689beb 100644
--- a/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/ChunkManagerFactoryConfig.java
+++ b/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/ChunkManagerFactoryConfig.java
@@ -49,6 +49,7 @@ public ChunkManagerFactoryConfig(final Map<?, ?> originals) {
         super(CONFIG, originals);
     }
 
+    @SuppressWarnings("unchecked")
     public Class<ChunkCache<?>> cacheClass() {
         return (Class<ChunkCache<?>>) getClass(CHUNK_CACHE_CONFIG);
     }
diff --git a/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfig.java b/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfig.java
index 387adc900..3d76a4bea 100644
--- a/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfig.java
+++ b/core/src/main/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfig.java
@@ -28,7 +28,8 @@
 
 public class DiskBasedChunkCacheConfig extends ChunkCacheConfig {
     private static final String CACHE_PATH_CONFIG = "path";
-    private static final String CACHE_PATH_DOC = "Cache directory";
+    private static final String CACHE_PATH_DOC = "Cache base directory. "
+        + "It is required to exist and be writable prior to the execution of the plugin.";
 
     public static final String TEMP_CACHE_DIRECTORY = "temp";
     public static final String CACHE_DIRECTORY = "cache";
@@ -47,21 +48,26 @@ private static ConfigDef configDef() {
 
     public DiskBasedChunkCacheConfig(final Map<String, ?> props) {
         super(configDef(), props);
+        final var baseCachePath = baseCachePath();
+        if (!Files.isDirectory(baseCachePath) || !Files.isWritable(baseCachePath)) {
+            throw new ConfigException(CACHE_PATH_CONFIG, baseCachePath,
+                baseCachePath + " must exists and be a writable directory");
+        }
         // Cleaning the cache directory since there is no way so far
         // to reuse previously cached files after broker restart.
         resetCacheDirectory();
     }
 
     private void resetCacheDirectory() {
-        if (Files.exists(baseCachePath())) {
-            try {
-                FileUtils.deleteDirectory(baseCachePath().toFile());
-                Files.createDirectories(cachePath());
-                Files.createDirectories(tempCachePath());
-            } catch (final IOException e) {
-                throw new ConfigException(CACHE_PATH_CONFIG, baseCachePath(),
-                        "Failed to reset cache directory, please empty the directory, reason: " + e.getMessage());
-            }
+        final var baseCachePath = baseCachePath();
+        try {
+            FileUtils.cleanDirectory(baseCachePath.toFile());
+            Files.createDirectories(cachePath());
+            Files.createDirectories(tempCachePath());
+        } catch (final IOException e) {
+            // printing e.toString instead of e.getMessage as some message have no context without exception type
+            throw new ConfigException(CACHE_PATH_CONFIG, baseCachePath,
+                "Failed to reset cache directory, please empty the directory, reason: " + e);
         }
     }
 
diff --git a/core/src/test/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfigTest.java b/core/src/test/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfigTest.java
index 0662b8001..a7e992a9c 100644
--- a/core/src/test/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfigTest.java
+++ b/core/src/test/java/io/aiven/kafka/tieredstorage/chunkmanager/cache/DiskBasedChunkCacheConfigTest.java
@@ -54,33 +54,30 @@ void setUp() throws IOException {
 
     @Test
     void validConfig() {
-        final String emptyCachePath = "/path/";
-        final DiskBasedChunkCacheConfig diskBasedCachingChunkManagerConfig
-                = new DiskBasedChunkCacheConfig(
-                Map.of(
-                        "size", "-1",
-                        "path", emptyCachePath
-                )
+        final var config = new DiskBasedChunkCacheConfig(
+            Map.of(
+                "size", "-1",
+                "path", path.toString()
+            )
         );
-        assertThat(diskBasedCachingChunkManagerConfig.baseCachePath()).isEqualTo(Path.of(emptyCachePath));
+        assertThat(config.baseCachePath()).isEqualTo(path);
     }
 
     @Test
-    void noneEmptyCachePath() throws IOException {
+    void nonEmptyCachePath() throws IOException {
         Files.createFile(tempCachePath.resolve("temp-file"));
         Files.createFile(cachePath.resolve("cached-file"));
 
-        final DiskBasedChunkCacheConfig diskBasedCachingChunkManagerConfig
-                = new DiskBasedChunkCacheConfig(
-                Map.of(
-                        "size", "-1",
-                        "path", path.toString()
-                )
+        final var config = new DiskBasedChunkCacheConfig(
+            Map.of(
+                "size", "-1",
+                "path", path.toString()
+            )
         );
-        assertThat(diskBasedCachingChunkManagerConfig.baseCachePath())
-                .isEqualTo(path)
-                .isDirectoryContaining(cp -> cp.equals(cachePath))
-                .isDirectoryContaining(tcp -> tcp.equals(tempCachePath));
+        assertThat(config.baseCachePath())
+            .isEqualTo(path)
+            .isDirectoryContaining(cp -> cp.equals(cachePath))
+            .isDirectoryContaining(tcp -> tcp.equals(tempCachePath));
         assertThat(cachePath).isEmptyDirectory();
         assertThat(tempCachePath).isEmptyDirectory();
     }
@@ -91,18 +88,30 @@ void failedToResetCachePath() throws IOException {
         final var file = Files.createFile(cachePath.resolve("cached-file"));
 
         try (final var filesMockedStatic = mockStatic(FileUtils.class, CALLS_REAL_METHODS)) {
-            filesMockedStatic.when(() -> FileUtils.deleteDirectory(eq(path.toFile())))
-                    .thenThrow(new IOException("Failed to delete file " + file));
+            filesMockedStatic.when(() -> FileUtils.cleanDirectory(eq(path.toFile())))
+                .thenThrow(new IOException("Failed to delete file " + file));
             assertThat(path).exists();
             assertThatThrownBy(() -> new DiskBasedChunkCacheConfig(
-                    Map.of(
-                            "size", "-1",
-                            "path", path.toString()
-                    )
+                Map.of(
+                    "size", "-1",
+                    "path", path.toString()
+                )
             )).isInstanceOf(ConfigException.class)
-                    .hasMessage("Invalid value " + path + " for configuration path: "
-                            + "Failed to reset cache directory, please empty the directory, reason: "
-                            + "Failed to delete file " + file);
+                .hasMessage("Invalid value " + path + " for configuration path: "
+                    + "Failed to reset cache directory, please empty the directory, reason: "
+                    + "java.io.IOException: Failed to delete file " + file);
         }
     }
+
+    @Test
+    void failWhenNonExistingBaseDir() {
+        assertThatThrownBy(() -> new DiskBasedChunkCacheConfig(
+            Map.of(
+                "size", "-1",
+                "path", "non-existing"
+            )))
+            .isInstanceOf(ConfigException.class)
+            .hasMessage("Invalid value non-existing for configuration path: "
+                + "non-existing must exists and be a writable directory");
+    }
 }