diff --git a/ehcache-core/src/main/java/org/ehcache/core/spi/service/LocalPersistenceService.java b/ehcache-core/src/main/java/org/ehcache/core/spi/service/LocalPersistenceService.java
index e068f152f9..7ce2fe1d0c 100644
--- a/ehcache-core/src/main/java/org/ehcache/core/spi/service/LocalPersistenceService.java
+++ b/ehcache-core/src/main/java/org/ehcache/core/spi/service/LocalPersistenceService.java
@@ -75,4 +75,16 @@ interface SafeSpaceIdentifier {
      */
     File getRoot();
   }
+
+  /**
+   * Return the cleanliness of the state stored in this service.
+   * <p>
+   * Stored state is assumed to be clean if the service detects
+   * that the last started instantiation of this service was shutdown
+   * successfully.
+   *
+   * @return {@code true} if the state is clean
+   * @throws IllegalStateException if the service is not started
+   */
+  boolean isClean();
 }
diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultDiskResourceService.java b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultDiskResourceService.java
index 2d3e273a19..4916903fb1 100644
--- a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultDiskResourceService.java
+++ b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultDiskResourceService.java
@@ -60,6 +60,10 @@ private boolean isStarted() {
   @Override
   public void start(final ServiceProvider<Service> serviceProvider) {
     innerStart(serviceProvider);
+    if (!persistenceService.isClean()) {
+      destroyAll();
+      LOGGER.info("Probably unclean shutdown was done, so deleted root directory.");
+    }
   }
 
   /**
@@ -89,7 +93,7 @@ public void stop() {
    */
   @Override
   public boolean handlesResourceType(ResourceType<?> resourceType) {
-    return persistenceService != null && ResourceType.Core.DISK.equals(resourceType);
+    return ResourceType.Core.DISK.equals(resourceType);
   }
 
   /**
@@ -97,9 +101,6 @@ public boolean handlesResourceType(ResourceType<?> resourceType) {
    */
   @Override
   public PersistenceSpaceIdentifier<DiskResourceService> getPersistenceSpaceIdentifier(String name, CacheConfiguration<?, ?> config) throws CachePersistenceException {
-    if (persistenceService == null) {
-      return null;
-    }
     boolean persistent = config.getResourcePools().getPoolForResource(ResourceType.Core.DISK).isPersistent();
     while (true) {
       PersistenceSpace persistenceSpace = knownPersistenceSpaces.get(name);
@@ -173,10 +174,6 @@ private void checkStarted() {
   public void destroy(String name) {
     checkStarted();
 
-    if(persistenceService == null) {
-      return;
-    }
-
     PersistenceSpace space = knownPersistenceSpaces.remove(name);
     SafeSpaceIdentifier identifier = (space == null) ?
       persistenceService.createSafeSpaceIdentifier(PERSISTENCE_SPACE_OWNER, name) : space.identifier.persistentSpaceId;
@@ -190,10 +187,6 @@ public void destroy(String name) {
   public void destroyAll() {
     checkStarted();
 
-    if(persistenceService == null) {
-      return;
-    }
-
     persistenceService.destroyAll(PERSISTENCE_SPACE_OWNER);
   }
 
diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultLocalPersistenceService.java b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultLocalPersistenceService.java
index 81f05f4554..00f8871811 100644
--- a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultLocalPersistenceService.java
+++ b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/DefaultLocalPersistenceService.java
@@ -37,6 +37,7 @@
 import static org.ehcache.impl.persistence.FileUtils.safeIdentifier;
 import static org.ehcache.impl.persistence.FileUtils.tryRecursiveDelete;
 import static org.ehcache.impl.persistence.FileUtils.validateName;
+import static org.ehcache.impl.persistence.FileUtils.isDirectoryEmpty;
 
 /**
  * Implements the local persistence service that provides individual sub-spaces for different
@@ -48,10 +49,12 @@ public class DefaultLocalPersistenceService implements LocalPersistenceService {
 
   private final File rootDirectory;
   private final File lockFile;
+  private final File cleanFile;
 
   private FileLock lock;
   private RandomAccessFile rw;
   private boolean started;
+  private boolean clean;
 
   /**
    * Creates a new service instance using the provided configuration.
@@ -65,6 +68,7 @@ public DefaultLocalPersistenceService(final DefaultPersistenceConfiguration pers
       throw new NullPointerException("DefaultPersistenceConfiguration cannot be null");
     }
     lockFile = new File(rootDirectory, ".lock");
+    cleanFile = new File(rootDirectory, ".clean");
   }
 
   /**
@@ -82,7 +86,15 @@ public synchronized void startForMaintenance(ServiceProvider<? super Maintainabl
 
   private void internalStart() {
     if (!started) {
+      clean = false;
       createLocationIfRequiredAndVerify(rootDirectory);
+      try {
+        if (isDirectoryEmpty(rootDirectory.toPath())) {
+          clean = true;
+        }
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
       try {
         rw = new RandomAccessFile(lockFile, "rw");
       } catch (FileNotFoundException e) {
@@ -104,6 +116,19 @@ private void internalStart() {
       if (lock == null) {
         throw new RuntimeException("Persistence directory already locked by another process: " + rootDirectory.getAbsolutePath());
       }
+
+      if (cleanFile.exists()) {
+        try {
+          LOGGER.debug("clean file exists, trying to delete the file.");
+          Files.delete(cleanFile.toPath());
+          clean = true;
+          LOGGER.debug("clean file is deleted.");
+        } catch (IOException e) {
+          LOGGER.debug("clean file is not deleted {}.", cleanFile.getPath());
+          throw new RuntimeException(e);
+        }
+      }
+
       started = true;
       LOGGER.debug("RootDirectory Locked");
     }
@@ -115,6 +140,19 @@ private void internalStart() {
   @Override
   public synchronized void stop() {
     if (started) {
+      try {
+        if (cleanFile.createNewFile()) {
+          LOGGER.debug("clean file is created.");
+        } else {
+          LOGGER.warn("clean file already exists. The file didn't got deleted, may be due to network issue or file permission on directory." +
+            "\n Hint: clean file exists on service start-up, indicates service was stopped cleanly last time. It gets created while the service is stopped and it should be deleted while the service is started." +
+            "\n Action: Please verify there permission to delete the file and delete the root directory prior to start the service again.");
+        }
+      } catch (IOException e) {
+        LOGGER.warn("clean file is not created. Reason: " + e.getMessage() +
+          "\n Hint: clean file exists on service start-up, indicates service was stopped cleanly last time. It gets created while the service is stopped and it should be deleted while the service is started." +
+          "\n Action: Do resolve the exception received. Prior to start the service again, please delete the root directory.");
+      }
       try {
         lock.release();
         // Closing RandomAccessFile so that files gets deleted on windows and
@@ -129,6 +167,7 @@ public synchronized void stop() {
       } catch (IOException e) {
         throw new RuntimeException("Couldn't unlock rootDir: " + rootDirectory.getAbsolutePath(), e);
       }
+
       started = false;
       LOGGER.debug("RootDirectory Unlocked");
     }
@@ -198,6 +237,15 @@ public void destroyAll(String owner) {
     }
   }
 
+  @Override
+  public final synchronized boolean isClean() {
+    if (started) {
+      return clean;
+    } else {
+      throw new IllegalStateException("Service is not running");
+    }
+  }
+
   private void destroy(SafeSpace ss, boolean verbose) {
     if (verbose) {
       LOGGER.debug("Destroying file based persistence context for {}", ss.identifier);
@@ -209,7 +257,6 @@ private void destroy(SafeSpace ss, boolean verbose) {
     }
   }
 
-
   private SafeSpace createSafeSpaceLogical(String owner, String identifier) {
     File ownerDirectory = new File(rootDirectory, owner);
     File directory = new File(ownerDirectory, safeIdentifier(identifier));
diff --git a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/FileUtils.java b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/FileUtils.java
index 4a1ff25b34..67e64426e2 100644
--- a/ehcache-impl/src/main/java/org/ehcache/impl/persistence/FileUtils.java
+++ b/ehcache-impl/src/main/java/org/ehcache/impl/persistence/FileUtils.java
@@ -24,12 +24,15 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Path;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
 import java.util.HashSet;
 import java.util.Locale;
 import java.util.Set;
+import java.util.stream.Stream;
 
 import static java.lang.Integer.toHexString;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -156,4 +159,12 @@ private static MessageDigest getSha1Digest() {
       throw new AssertionError("All JDKs must have SHA-1");
     }
   }
+
+  static boolean isDirectoryEmpty(Path path) throws IOException {
+    try (Stream<Path> entries = java.nio.file.Files.list(path)) {
+      return !entries.findFirst().isPresent();
+    } catch (UncheckedIOException e) {
+      throw e.getCause();
+    }
+  }
 }
diff --git a/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultDiskResourceServiceTest.java b/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultDiskResourceServiceTest.java
index dc1fda510a..7e60cf019e 100644
--- a/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultDiskResourceServiceTest.java
+++ b/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultDiskResourceServiceTest.java
@@ -15,38 +15,30 @@
  */
 package org.ehcache.impl.persistence;
 
-import org.ehcache.CachePersistenceException;
 import org.ehcache.config.ResourceType;
+import org.ehcache.core.spi.ServiceLocator;
 import org.ehcache.core.spi.service.LocalPersistenceService;
-import org.ehcache.spi.service.Service;
-import org.ehcache.spi.service.ServiceProvider;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.junit.experimental.runners.Enclosed;
-import org.junit.runner.RunWith;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.ehcache.test.MockitoUtil.uncheckedGenericMock;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
 /**
  * @author Henri Tremblay
  */
-@RunWith(Enclosed.class)
 public class DefaultDiskResourceServiceTest {
 
-  public static abstract class AbstractDefaultDiskResourceServiceTest {
-
-    protected DefaultDiskResourceService service = new DefaultDiskResourceService();
-    protected ServiceProvider<Service> serviceProvider = uncheckedGenericMock(ServiceProvider.class);
+    DefaultDiskResourceService service = new DefaultDiskResourceService();
+    LocalPersistenceService persistenceService = mock(LocalPersistenceService.class);
 
     @Before
     public void setup() {
-      service.start(serviceProvider);
+      ServiceLocator serviceLocator = ServiceLocator.dependencySet().with(service).with(persistenceService).build();
+      service.start(serviceLocator);
     }
 
     @After
@@ -54,18 +46,6 @@ public void tearDown() {
       service.stop();
     }
 
-  }
-
-  public static class WithPersistenceService extends AbstractDefaultDiskResourceServiceTest {
-
-    LocalPersistenceService persistenceService = mock(LocalPersistenceService.class);
-
-    @Before
-    public void setup() {
-      when(serviceProvider.getService(LocalPersistenceService.class)).thenReturn(persistenceService);
-      super.setup();
-    }
-
     @Test
     public void testHandlesResourceType() {
       assertThat(service.handlesResourceType(ResourceType.Core.DISK)).isTrue();
@@ -74,58 +54,12 @@ public void testHandlesResourceType() {
     @Test
     public void testDestroyAll() {
       service.destroyAll();
-      verify(persistenceService).destroyAll(DefaultDiskResourceService.PERSISTENCE_SPACE_OWNER);
-    }
-
-    @Test
-    public void testDestroy() throws CachePersistenceException {
-      service.destroy("test"); // should do nothing
-    }
-
-    // Some tests still missing here
-  }
-
-  public static class WithoutPersistenceService extends AbstractDefaultDiskResourceServiceTest {
-
-    @Test
-    public void testHandlesResourceType() {
-      assertThat(service.handlesResourceType(ResourceType.Core.DISK)).isFalse();
-    }
-
-    @Test
-    public void testDestroyAll() {
-      service.destroyAll(); // should do nothing
+      verify(persistenceService, times(2)).destroyAll(DefaultDiskResourceService.PERSISTENCE_SPACE_OWNER);
     }
 
     @Test
-    public void testDestroy() throws CachePersistenceException {
+    public void testDestroy() {
       service.destroy("test"); // should do nothing
     }
 
-    @Test
-    public void testCreatePersistenceContextWithin() throws CachePersistenceException {
-      assertThatThrownBy(() -> service.createPersistenceContextWithin(null, "test"))
-        .isInstanceOf(CachePersistenceException.class).withFailMessage("Unknown space: null");
-    }
-
-    @Test
-    public void testGetPersistenceSpaceIdentifier() throws CachePersistenceException {
-      assertThat(service.getPersistenceSpaceIdentifier("test", null)).isNull();
-    }
-
-
-    @Test
-    public void testGetStateRepositoryWithin() throws CachePersistenceException {
-      assertThatThrownBy(() -> service.getStateRepositoryWithin(null, "test"))
-        .isInstanceOf(CachePersistenceException.class).withFailMessage("Unknown space: null");
-    }
-
-    @Test
-    public void testReleasePersistenceSpaceIdentifier() throws CachePersistenceException {
-      assertThatThrownBy(() -> service.getStateRepositoryWithin(null, "test"))
-        .isInstanceOf(CachePersistenceException.class).withFailMessage("Unknown space: null");
-    }
-
-  }
-
 }
diff --git a/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultLocalPersistenceServiceTest.java b/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultLocalPersistenceServiceTest.java
index 4281cc9345..070038d620 100644
--- a/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultLocalPersistenceServiceTest.java
+++ b/ehcache-impl/src/test/java/org/ehcache/impl/persistence/DefaultLocalPersistenceServiceTest.java
@@ -35,6 +35,8 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeTrue;
 
@@ -140,4 +142,50 @@ public void testExclusiveLock() throws IOException {
     RuntimeException thrown = assertThrows(RuntimeException.class, () -> service2.start(null));
     assertThat(thrown, hasProperty("message", is("Persistence directory already locked by this process: " + testFolder.getAbsolutePath())));
   }
+
+  @Test
+  public void testServiceShutdownWithEmptyDirectory() throws IOException {
+    File f = folder.newFolder("testServiceShutdownWithEmptyDirectory");
+    final DefaultLocalPersistenceService service = new DefaultLocalPersistenceService(new DefaultPersistenceConfiguration(f));
+    service.start(null);
+    assertTrue(service.isClean());
+    service.stop();
+    service.start(null);
+    assertTrue(service.isClean());
+    service.stop();
+  }
+
+  @Test
+  public void testServiceShutdownWithNonEmptyDirectory() throws IOException {
+    File f = folder.newFolder("testServiceShutdownWithNonEmptyDirectory");
+    final DefaultLocalPersistenceService service = new DefaultLocalPersistenceService(new DefaultPersistenceConfiguration(f));
+    new File(f, "dummy.txt").createNewFile();
+    new File(f, ".clean").createNewFile();
+    service.start(null);
+    assertTrue(service.isClean());
+    service.stop();
+  }
+
+  @Test
+  public void testServiceShutdownUnexpectedly() throws IOException {
+    // Service shutdown unexpectedly means directory exists with some data but without .clean file.
+    File f = folder.newFolder("testServiceShutdownUnexpectedly");
+    final DefaultLocalPersistenceService service = new DefaultLocalPersistenceService(new DefaultPersistenceConfiguration(f));
+    new File(f, "dummy.txt").createNewFile();
+    service.start(null);
+    assertFalse(service.isClean());
+    service.stop();
+  }
+
+  @Test
+  public void testServiceShutdownStatusIfServiceIsNotRunning() throws IOException {
+    File f = folder.newFolder("testServiceShutdownStatusIfServiceIsNotRunning");
+    final DefaultLocalPersistenceService service = new DefaultLocalPersistenceService(new DefaultPersistenceConfiguration(f));
+    try {
+      service.isClean();
+      fail("Expected IllegalStateException");
+    } catch(IllegalStateException e) {
+      assertThat(e.getMessage(), equalTo("Service is not running"));
+    }
+  }
 }