class OldDeletedEntriesCleanupThread extends Thread
        implements MapClosable, Predicate<ReplicableEntry> {

    /**
     * Don't store a strong ref to a map in order to avoid it's leaking, if the user forgets to close() map, from where this thread is shut down
     * explicitly. Dereference map within a single method, {@link #cleanupSegment()}. The map has a chance to be collected by GC when this thread is
     * sleeping after cleaning up a segment.
     */
    private final WeakReference<ReplicatedChronicleMap<?, ?, ?>> mapRef;
    /**
     * {@code cleanupTimeout}, {@link #cleanupTimeoutUnit} and {@link #segments} are parts of the
     * cleaned Map's state, extracted in order to minimize accesses to the map.
     *
     * @see ChronicleHashBuilderPrivateAPI#removedEntryCleanupTimeout(long, TimeUnit)
     */
    private final long cleanupTimeout;
    private final TimeUnit cleanupTimeoutUnit;
    private final int segments;

    /**
     * {@code segmentsPermutation} and {@link #inverseSegmentsPermutation} determine random order,
     * in which segments are cleaned up.
     */
    private final int[] segmentsPermutation;
    private final int[] inverseSegmentsPermutation;

    /**
     * This object is used to determine that this thread is parked from {@link #sleepMillis(long)}
     * or {@link #sleepNanos(long)}, not somewhere inside ChronicleMap logic, to interrupt()
     * selectively in {@link #close()}.
     */
    private final Object cleanupSleepingHandle = new Object();

    private volatile boolean shutdown;

    private long prevSegment0ScanStart = -1;
    private long removedCompletely;
    private long startTime = System.currentTimeMillis();

    OldDeletedEntriesCleanupThread(ReplicatedChronicleMap<?, ?, ?> map) {
        super("Cleanup Thread for " + map.toIdentityString());
        setDaemon(true);
        this.mapRef = new WeakReference<>(map);
        cleanupTimeout = map.cleanupTimeout;
        cleanupTimeoutUnit = map.cleanupTimeoutUnit;
        segments = map.segments();

        segmentsPermutation = randomPermutation(map.segments());
        inverseSegmentsPermutation = inversePermutation(segmentsPermutation);
    }

    private static int[] randomPermutation(int n) {
        int[] a = new int[n];
        for (int i = 0; i < n; i++) {
            a[i] = i;
        }
        shuffle(a);
        return a;
    }

    // Implementing Fisher–Yates shuffle
    private static void shuffle(int[] a) {
        SecureRandom rnd = new SecureRandom();
        for (int i = a.length - 1; i > 0; i--) {
            int index = rnd.nextInt(i + 1);
            int e = a[index];
            a[index] = a[i];
            a[i] = e;
        }
    }

    private static int[] inversePermutation(int[] permutation) {
        int n = permutation.length;
        int[] inverse = new int[n];
        for (int i = 0; i < n; i++) {
            inverse[permutation[i]] = i;
        }
        return inverse;
    }

    @Override
    public void run() {
        throwExceptionIfClosed();

        if (System.currentTimeMillis() - startTime < 1_000)
            return;

        while (!shutdown) {
            int nextSegmentIndex;
            try {
                nextSegmentIndex = cleanupSegment();
            } catch (Exception e) {
                if (shutdown)
                    break;
                throw e;
            }
            if (nextSegmentIndex == -1)
                return;
            if (nextSegmentIndex == 0) {
                long currentTime = currentTime();
                long mapScanTime = systemTimeIntervalBetween(
                        prevSegment0ScanStart, currentTime, cleanupTimeoutUnit);
                Jvm.debug().on(getClass(), "Old deleted entries scan time: " + mapScanTime + " " + cleanupTimeoutUnit);
                if (mapScanTime < cleanupTimeout) {
                    long timeToSleep = cleanupTimeoutUnit.toMillis(cleanupTimeout - mapScanTime);
                    if (timeToSleep > 0) {
                        sleepMillis(timeToSleep);
                    } else {
                        sleepNanos(cleanupTimeoutUnit.toNanos(cleanupTimeout - mapScanTime));
                    }
                }
            }
        }
    }

    /**
     * @return next segment index to cleanup, or -1 if cleanup thread should be shut down
     */
    private int cleanupSegment() {<FILL_FUNCTION_BODY>}

    @Override
    public boolean test(ReplicableEntry e) {
        throwExceptionIfClosed();

        if (shutdown)
            return false;
        if (e instanceof MapAbsentEntry) {
            long deleteTimeout = systemTimeIntervalBetween(
                    e.originTimestamp(), currentTime(), cleanupTimeoutUnit);
            if (deleteTimeout > cleanupTimeout && !e.isChanged()) {
                e.doRemoveCompletely();
                removedCompletely++;
            }
        }
        return true;
    }

    private void sleepMillis(long millis) {
        long deadline = System.currentTimeMillis() + millis;
        while (System.currentTimeMillis() < deadline && !shutdown)
            LockSupport.parkUntil(cleanupSleepingHandle, deadline);
    }

    private void sleepNanos(long nanos) {
        long deadline = System.nanoTime() + nanos;
        while (System.nanoTime() < deadline && !shutdown)
            LockSupport.parkNanos(cleanupSleepingHandle, deadline);
    }

    @Override
    public void close() {
        shutdown = true;
        // this means blocked in sleepMillis() or sleepNanos()
        if (LockSupport.getBlocker(this) == cleanupSleepingHandle)
            this.interrupt(); // unblock
    }

    private int nextSegmentIndex(int segmentIndex) {
        int permutationIndex = inverseSegmentsPermutation[segmentIndex];
        int nextPermutationIndex = (permutationIndex + 1) % segments;
        return segmentsPermutation[nextPermutationIndex];
    }
}

        ReplicatedChronicleMap<?, ?, ?> map = mapRef.get();
        if (map == null)
            return -1;
        int segmentIndex = map.globalMutableState().getCurrentCleanupSegmentIndex();
        int nextSegmentIndex;
        try (MapSegmentContext<?, ?, ?> context = map.segmentContext(segmentIndex)) {
            if (segmentIndex == 0)
                prevSegment0ScanStart = currentTime();
            removedCompletely = 0;
            if (((ReplicatedHashSegmentContext<?, ?>) context)
                    .forEachSegmentReplicableEntryWhile(this)) {
                Jvm.debug().on(getClass(),
                        "Removed " + removedCompletely + " old deleted entries " +
                                "in the segment " + segmentIndex);
                nextSegmentIndex = nextSegmentIndex(segmentIndex);
                map.globalMutableState().setCurrentCleanupSegmentIndex(nextSegmentIndex);
                return nextSegmentIndex;
            } else {
                // forEachWhile returned false => interrupted => shutdown = true
                assert shutdown;
                return -1;
            }
        }
    
<methods>public void <init>() ,public void <init>(java.lang.Runnable) ,public void <init>(java.lang.String) ,public void <init>(java.lang.ThreadGroup, java.lang.Runnable) ,public void <init>(java.lang.ThreadGroup, java.lang.String) ,public void <init>(java.lang.Runnable, java.lang.String) ,public void <init>(java.lang.ThreadGroup, java.lang.Runnable, java.lang.String) ,public void <init>(java.lang.ThreadGroup, java.lang.Runnable, java.lang.String, long) ,public void <init>(java.lang.ThreadGroup, java.lang.Runnable, java.lang.String, long, boolean) ,public static int activeCount() ,public final void checkAccess() ,public int countStackFrames() ,public static native java.lang.Thread currentThread() ,public static void dumpStack() ,public static int enumerate(java.lang.Thread[]) ,public static Map<java.lang.Thread,java.lang.StackTraceElement[]> getAllStackTraces() ,public java.lang.ClassLoader getContextClassLoader() ,public static java.lang.Thread.UncaughtExceptionHandler getDefaultUncaughtExceptionHandler() ,public long getId() ,public final java.lang.String getName() ,public final int getPriority() ,public java.lang.StackTraceElement[] getStackTrace() ,public java.lang.Thread.State getState() ,public final java.lang.ThreadGroup getThreadGroup() ,public java.lang.Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() ,public static native boolean holdsLock(java.lang.Object) ,public void interrupt() ,public static boolean interrupted() ,public final boolean isAlive() ,public final boolean isDaemon() ,public boolean isInterrupted() ,public final void join() throws java.lang.InterruptedException,public final synchronized void join(long) throws java.lang.InterruptedException,public final synchronized void join(long, int) throws java.lang.InterruptedException,public static void onSpinWait() ,public final void resume() ,public void run() ,public void setContextClassLoader(java.lang.ClassLoader) ,public final void setDaemon(boolean) ,public static void setDefaultUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler) ,public final synchronized void setName(java.lang.String) ,public final void setPriority(int) ,public void setUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler) ,public static native void sleep(long) throws java.lang.InterruptedException,public static void sleep(long, int) throws java.lang.InterruptedException,public synchronized void start() ,public final void stop() ,public final void suspend() ,public java.lang.String toString() ,public static native void yield() <variables>private static final java.lang.StackTraceElement[] EMPTY_STACK_TRACE,public static final int MAX_PRIORITY,public static final int MIN_PRIORITY,public static final int NORM_PRIORITY,private volatile sun.nio.ch.Interruptible blocker,private final java.lang.Object blockerLock,private java.lang.ClassLoader contextClassLoader,private boolean daemon,private static volatile java.lang.Thread.UncaughtExceptionHandler defaultUncaughtExceptionHandler,private volatile long eetop,private java.lang.ThreadGroup group,java.lang.ThreadLocal.ThreadLocalMap inheritableThreadLocals,private java.security.AccessControlContext inheritedAccessControlContext,private volatile boolean interrupted,private volatile java.lang.String name,volatile java.lang.Object parkBlocker,private int priority,private final long stackSize,private boolean stillborn,private java.lang.Runnable target,private static int threadInitNumber,int threadLocalRandomProbe,int threadLocalRandomSecondarySeed,long threadLocalRandomSeed,java.lang.ThreadLocal.ThreadLocalMap threadLocals,private static long threadSeqNumber,private volatile int threadStatus,private final long tid,private volatile java.lang.Thread.UncaughtExceptionHandler uncaughtExceptionHandler