class CdcDeleteLostSegmentsJob extends VisorJob<CdcDeleteLostSegmentLinksCommandArg, Void> {
        /** */
        private static final long serialVersionUID = 0L;

        /** Injected logger. */
        @LoggerResource
        protected IgniteLogger log;

        /**
         * Create job with specified argument.
         *
         * @param arg   Job argument.
         * @param debug Flag indicating whether debug information should be printed into node log.
         */
        protected CdcDeleteLostSegmentsJob(CdcDeleteLostSegmentLinksCommandArg arg, boolean debug) {
            super(arg, debug);
        }

        /** {@inheritDoc} */
        @Override protected Void run(CdcDeleteLostSegmentLinksCommandArg arg) throws IgniteException {<FILL_FUNCTION_BODY>}
    }

            FileWriteAheadLogManager wal = (FileWriteAheadLogManager)ignite.context().cache().context().wal(true);

            File walCdcDir = wal.walCdcDirectory();

            if (walCdcDir == null)
                throw new IgniteException("CDC is not configured.");

            CdcFileLockHolder lock = new CdcFileLockHolder(walCdcDir.getAbsolutePath(), "Delete lost segments job", log);

            try {
                lock.tryLock(1);

                try (Stream<Path> cdcFiles = Files.list(walCdcDir.toPath())) {
                    Set<File> delete = new HashSet<>();

                    AtomicLong lastSgmnt = new AtomicLong(-1);

                    cdcFiles
                        .filter(p -> WAL_SEGMENT_FILE_FILTER.accept(p.toFile()))
                        .sorted(Comparator.comparingLong(FileWriteAheadLogManager::segmentIndex)
                            .reversed()) // Sort by segment index.
                        .forEach(path -> {
                            long idx = FileWriteAheadLogManager.segmentIndex(path);

                            if (lastSgmnt.get() == -1 || lastSgmnt.get() - idx == 1) {
                                lastSgmnt.set(idx);

                                return;
                            }

                            delete.add(path.toFile());
                        });

                    if (delete.isEmpty()) {
                        log.info("Lost segment CDC links were not found.");

                        return null;
                    }

                    log.info("Found lost segment CDC links. The following links will be deleted: " + delete);

                    delete.forEach(file -> {
                        if (!file.delete()) {
                            throw new IgniteException("Failed to delete lost segment CDC link [file=" +
                                file.getAbsolutePath() + ']');
                        }

                        log.info("Segment CDC link deleted [file=" + file.getAbsolutePath() + ']');
                    });

                    Path stateDir = walCdcDir.toPath().resolve(STATE_DIR);

                    if (stateDir.toFile().exists()) {
                        File walState = stateDir.resolve(WAL_STATE_FILE_NAME).toFile();

                        if (walState.exists() && !walState.delete()) {
                            throw new IgniteException("Failed to delete wal state file [file=" +
                                walState.getAbsolutePath() + ']');
                        }
                    }
                }
                catch (IOException e) {
                    throw new RuntimeException("Failed to delete lost segment CDC links.", e);
                }
            }
            catch (IgniteCheckedException e) {
                throw new RuntimeException("Failed to delete lost segment CDC links. " +
                    "Unable to acquire lock to lock CDC folder. Make sure a CDC app is shut down " +
                    "[dir=" + walCdcDir.getAbsolutePath() + ", reason=" + e.getMessage() + ']');
            }
            finally {
                U.closeQuiet(lock);
            }

            return null;
        
<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.cdc.CdcDeleteLostSegmentLinksCommandArg>) ,public final java.lang.Void 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.cdc.CdcDeleteLostSegmentLinksCommandArg taskArg