private Runnable deleteFiles(final SimpleMeta simpleMeta,
                                 final TaskContext parentTaskContext,
                                 final Queue<Long> successfulMetaIdDeleteQueue,
                                 final Progress progress,
                                 final Map<Path, Path> dirToVolPathMap) {

        final DataStoreServiceConfig dataStoreServiceConfig = dataStoreServiceConfigProvider.get();
        return taskContextFactory.childContext(
                parentTaskContext,
                "Deleting files",
                taskContext -> {
                    try {
                        if (Thread.interrupted()) {
                            throw new InterruptedException();
                        }
                        // This all needs to be re-runnable so if we come back again with partially deleted
                        // state it will cope
                        if (progress.hasBreachedThreshold()) {
                            LOGGER.warn("{} - Skipping file deletion for stream {} as failure threshold exceeded {}. " +
                                        "See property {}",
                                    TASK_NAME,
                                    simpleMeta.getId(),
                                    dataStoreServiceConfig.getDeleteFailureThreshold(),
                                    dataStoreServiceConfig.getFullPath(
                                            DataStoreServiceConfig.PROP_NAME_DELETE_FAILURE_THRESHOLD));
                        } else {
                            info(() -> LogUtil.message(
                                    "Physically deleting everything associated with stream: {}, " +
                                    "feed: {}, type: {}, statusTime: {}",
                                    simpleMeta.getId(), simpleMeta.getFeedName(),
                                    simpleMeta.getTypeName(),
                                    LogUtil.instant(simpleMeta.getStatusMs())));

                            final DataVolume dataVolume = dataVolumeDao.findDataVolume(simpleMeta.getId());
                            final boolean isSuccessful;
                            if (dataVolume == null) {
                                LOGGER.warn(() -> TASK_NAME + " - Unable to find any volume for " + simpleMeta);
                                isSuccessful = true;
                            } else {
                                switch (dataVolume.getVolume().getVolumeType()) {
                                    case STANDARD -> {
                                        final Path volumePath = pathCreator.toAppPath(dataVolume.getVolume().getPath());
                                        final Path file = fileSystemStreamPathHelper.getRootPath(
                                                volumePath,
                                                simpleMeta,
                                                simpleMeta.getTypeName());
                                        final Path dir = file.getParent();
                                        String baseName = file.getFileName().toString();
                                        baseName = baseName.substring(0, baseName.indexOf("."));

                                        if (Files.isDirectory(dir)) {
                                            isSuccessful = fsFileDeleter.deleteFilesByBaseName(
                                                    simpleMeta.getId(), dir, baseName, progress::addFileDeletes);

                                            dirToVolPathMap.put(dir, volumePath);
                                        } else {
                                            isSuccessful = true;
                                            LOGGER.warn(TASK_NAME + " - Directory does not exist '" +
                                                        FileUtil.getCanonicalPath(dir) +
                                                        "'");
                                        }
                                    }
                                    case S3 -> {
                                        // FIXME: Add something.
                                        isSuccessful = false;
                                    }
                                    default -> {
                                        isSuccessful = false;
                                    }
                                }

                            }

                            if (isSuccessful) {
                                successfulMetaIdDeleteQueue.add(simpleMeta.getId());
                            }
                            progress.recordMetaFileDeleteSuccess(isSuccessful);
                        }
                    } catch (final InterruptedException e) {
                        LOGGER.trace(e::getMessage, e);
                        // Keep interrupting this thread.
                        Thread.currentThread().interrupt();
                    }
                });
    }
Deletes a batch of simpleMetas from the file system and the database.

@param deleteThresholdEpoch The threshold for statusTime, i.e. records <= deleteThresholdEpoch
@param progress             Count of failures that spans multiple batches
@return Those simpleMeta items that could not be deleted for some reason.
stroom-data/stroom-data-store-impl-fs/src/main/java/stroom/data/store/impl/fs/PhysicalDeleteExecutor.java