class WalJob extends VisorJob<WalDeleteCommandArg, Collection<String>> {
        /** */
        private static final long serialVersionUID = 0L;

        /** Auto injected logger */
        @LoggerResource
        private transient IgniteLogger log;

        /**
         *  @param arg WAL task argument.
         *  @param debug Debug flag.
         */
        public WalJob(WalDeleteCommandArg arg, boolean debug) {
            super(arg, debug);
        }

        /** {@inheritDoc} */
        @Nullable @Override protected Collection<String> run(@Nullable WalDeleteCommandArg arg) throws IgniteException {<FILL_FUNCTION_BODY>}

        /**
         * Get unused wal segments.
         *
         * @param  wal Database manager.
         * @return {@link Collection<String>} of absolute paths of unused WAL segments.
         * @throws IgniteCheckedException if failed.
         */
        Collection<String> getUnusedWalSegments(
            GridCacheDatabaseSharedManager dbMgr,
            FileWriteAheadLogManager wal
        ) throws IgniteCheckedException {
            WALPointer lowBoundForTruncate = dbMgr.checkpointHistory().firstCheckpointPointer();

            if (lowBoundForTruncate == null)
                return Collections.emptyList();

            int maxIdx = resolveMaxReservedIndex(wal, lowBoundForTruncate);

            File[] walFiles = getWalArchiveDir().listFiles(WAL_ARCHIVE_FILE_FILTER);

            Collection<String> res = new ArrayList<>(walFiles != null && walFiles.length > 0 ? walFiles.length - 1 : 0);

            if (walFiles != null && walFiles.length > 0) {
                sortWalFiles(walFiles);

                // Obtain index of last archived WAL segment, it will not be deleted.
                long lastArchIdx = getIndex(walFiles[walFiles.length - 1]);

                for (File f : walFiles) {
                    long fileIdx = getIndex(f);

                    if (fileIdx < maxIdx && fileIdx < lastArchIdx)
                        res.add(f.getAbsolutePath());
                    else
                        break;
                }
            }

            return res;
        }

        /**
         * Delete unused wal segments.
         *
         * @param dbMgr Database manager.
         * @return {@link Collection<String>} of deleted WAL segment's files.
         * @throws IgniteCheckedException if failed.
         */
        Collection<String> deleteUnusedWalSegments(
            GridCacheDatabaseSharedManager dbMgr,
            FileWriteAheadLogManager wal
        ) throws IgniteCheckedException {
            WALPointer lowBoundForTruncate = dbMgr.checkpointHistory().firstCheckpointPointer();

            if (lowBoundForTruncate == null)
                return Collections.emptyList();

            int maxIdx = resolveMaxReservedIndex(wal, lowBoundForTruncate);

            File[] walFiles = getWalArchiveDir().listFiles(WAL_ARCHIVE_FILE_FILTER);

            dbMgr.onWalTruncated(lowBoundForTruncate);

            int num = wal.truncate(lowBoundForTruncate);

            if (walFiles != null) {
                sortWalFiles(walFiles);

                Collection<String> res = new ArrayList<>(num);

                for (File walFile: walFiles) {
                    if (getIndex(walFile) < maxIdx && num > 0)
                        res.add(walFile.getAbsolutePath());
                    else
                        break;

                    num--;
                }

                return res;
            }
            else
                return Collections.emptyList();

        }

        /**
         *
         */
        private int resolveMaxReservedIndex(FileWriteAheadLogManager wal, WALPointer lowBoundForTruncate) {
            int resCnt = wal.reserved(null, lowBoundForTruncate);

            long highIdx = lowBoundForTruncate.index();

            return (int)(highIdx - resCnt + 1);
        }

        /**
         * Get WAL archive directory from configuration.
         *
         * @return WAL archive directory.
         * @throws IgniteCheckedException if failed.
         */
        private File getWalArchiveDir() throws IgniteCheckedException {
            IgniteConfiguration igCfg = ignite.context().config();

            DataStorageConfiguration dsCfg = igCfg.getDataStorageConfiguration();

            PdsFolderSettings resFldrs = ignite.context().pdsFolderResolver().resolveFolders();

            String consId = resFldrs.folderName();

            File dir;

            if (dsCfg.getWalArchivePath() != null) {
                File workDir0 = new File(dsCfg.getWalArchivePath());

                dir = workDir0.isAbsolute() ?
                        new File(workDir0, consId) :
                        new File(U.resolveWorkDirectory(igCfg.getWorkDirectory(), dsCfg.getWalArchivePath(), false),
                                consId);
            }
            else
                dir = new File(U.resolveWorkDirectory(igCfg.getWorkDirectory(),
                        DataStorageConfiguration.DFLT_WAL_ARCHIVE_PATH, false), consId);

            if (!dir.exists())
                throw new IgniteCheckedException("WAL archive directory does not exists" + dir.getAbsolutePath());

            return dir;
        }

        /**
         * Sort WAL files according their indices.
         *
         * @param files Array of WAL segment files.
         */
        private void sortWalFiles(File[] files) {
            Arrays.sort(files, new Comparator<File>() {
                @Override public int compare(File o1, File o2) {
                    return Long.compare(getIndex(o1), getIndex(o2));
                }
            });
        }
    }

            try {
                GridKernalContext cctx = ignite.context();

                GridCacheDatabaseSharedManager dbMgr = (GridCacheDatabaseSharedManager)cctx.cache().context().database();
                FileWriteAheadLogManager wal = (FileWriteAheadLogManager)cctx.cache().context().wal();

                if (dbMgr == null || arg == null || wal == null)
                    return null;

                if (arg instanceof WalPrintCommandArg)
                    return getUnusedWalSegments(dbMgr, wal);
                else
                    return deleteUnusedWalSegments(dbMgr, wal);
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Failed to perform WAL task", e);

                throw new IgniteException("Failed to perform WAL task", e);
            }
        
<methods>public non-sealed void <init>() ,public Map<? extends org.apache.ignite.compute.ComputeJob,org.apache.ignite.cluster.ClusterNode> map(List<org.apache.ignite.cluster.ClusterNode>, VisorTaskArgument<org.apache.ignite.internal.management.wal.WalDeleteCommandArg>) ,public final org.apache.ignite.internal.management.wal.WalTaskResult reduce(List<org.apache.ignite.compute.ComputeJobResult>) ,public org.apache.ignite.compute.ComputeJobResultPolicy result(org.apache.ignite.compute.ComputeJobResult, List<org.apache.ignite.compute.ComputeJobResult>) <variables>protected boolean debug,protected transient org.apache.ignite.internal.IgniteEx ignite,protected long start,protected org.apache.ignite.internal.management.wal.WalDeleteCommandArg taskArg