class PurgeObjectGroupPreparationHandler extends ActionHandler {

    @Override
    public ItemStatus execute(WorkerParameters param, HandlerIO handler)
        throws ProcessingException {

        try {

            File objectGroupsToDeleteFile = handler.getNewLocalFile(OBJECT_GROUPS_TO_DELETE_FILE);
            File objectGroupsToDetachFile = handler.getNewLocalFile(OBJECT_GROUPS_TO_DETACH_FILE);

            try (
                OutputStream objectGroupsToDeleteStream = new FileOutputStream(objectGroupsToDeleteFile);
                OutputStream objectGroupsToDetachStream = new FileOutputStream(objectGroupsToDetachFile);
                JsonLineWriter objectGroupsToDeleteWriter = new JsonLineWriter(objectGroupsToDeleteStream);
                JsonLineWriter objectGroupsToDetachWriter = new JsonLineWriter(objectGroupsToDetachStream);
                CloseableIterator<String> iterator =
                    purgeReportService.exportDistinctObjectGroups(param.getContainerName())) {

                Iterator<List<String>> bulkIterator = Iterators.partition(iterator, objectGroupBulkSize);

                while (bulkIterator.hasNext()) {
                    List<String> objectGroupIds = bulkIterator.next();
                    process(new HashSet<>(objectGroupIds), objectGroupsToDeleteWriter, objectGroupsToDetachWriter,
                        param.getContainerName());
                }

            } catch (IOException | InvalidParseOperationException e) {
                throw new ProcessingStatusException(StatusCode.FATAL, "Could not generate object group distribution",
                    e);
            }

            handler.transferFileToWorkspace(OBJECT_GROUPS_TO_DELETE_FILE, objectGroupsToDeleteFile, true, false);
            handler.transferFileToWorkspace(OBJECT_GROUPS_TO_DETACH_FILE, objectGroupsToDetachFile, true, false);

            LOGGER.info("Purge of object group preparation succeeded");
            return buildItemStatus(actionId, StatusCode.OK, null);

        } catch (ProcessingStatusException e) {
            LOGGER.error(
                String.format("Purge of object group preparation failed with status [%s]", e.getStatusCode()),
                e);
            return buildItemStatus(actionId, e.getStatusCode(), e.getEventDetails());
        }
    }

    public  PurgeObjectGroupPreparationHandler(String actionId);
    @VisibleForTesting protected  PurgeObjectGroupPreparationHandler(
        String actionId, MetaDataClientFactory metaDataClientFactory,
        PurgeReportService purgeReportService,
        int objectGroupBulkSize);

}