/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.jackrabbit.oak.plugins.index.lucene.directory;

import static org.apache.jackrabbit.guava.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.apache.jackrabbit.oak.plugins.index.IndexCommitCallback.IndexProgress.COMMIT_FAILED;
import static org.apache.jackrabbit.oak.plugins.index.IndexCommitCallback.IndexProgress.COMMIT_SUCCEDED;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assume.assumeFalse;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.PosixFilePermission;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import ch.qos.logback.classic.Level;
import org.apache.jackrabbit.guava.common.collect.Iterators;
import org.apache.jackrabbit.guava.common.collect.Lists;
import org.apache.commons.io.FileUtils;
import org.apache.jackrabbit.core.data.DataStoreException;
import org.apache.jackrabbit.oak.commons.CIHelper;
import org.apache.jackrabbit.oak.commons.junit.LogCustomizer;
import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.ActiveDeletedBlobCollector;
import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.ActiveDeletedBlobCollectorImpl;
import org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory.BlobDeletionCallback;
import org.apache.jackrabbit.oak.spi.blob.BlobOptions;
import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
import org.apache.jackrabbit.oak.stats.Clock;
import org.jetbrains.annotations.NotNull;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.internal.util.collections.Sets;

public class ActiveDeletedBlobCollectorTest {
    @Rule
    public TemporaryFolder blobCollectionRoot = new TemporaryFolder(new File("target"));

    private Clock clock;
    private ChunkDeletionTrackingBlobStore blobStore;
    private ActiveDeletedBlobCollector adbc;

    @Before
    public void setup() throws Exception {
        clock = new Clock.Virtual();
        blobStore = new ChunkDeletionTrackingBlobStore();
        createBlobCollector();
    }

    private void createBlobCollector() {
        adbc = new ActiveDeletedBlobCollectorImpl(clock,
                new File(blobCollectionRoot.getRoot(), "/a"), newDirectExecutorService());
    }

    @Test
    public void simpleCase() throws Exception {
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();

        bdc.deleted("blobId", Collections.singleton("/a"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted("blobId");
    }

    @Test
    public void noopDoesNothing() throws Exception {
        adbc = ActiveDeletedBlobCollectorFactory.NOOP;
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();

        bdc.deleted("blobId", Collections.singleton("/a"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted();
    }

    @Test
    public void blobTimestampMustBeBiggerThanFileTimestamp() throws Exception {
        BlobDeletionCallback bdc1 = adbc.getBlobDeletionCallback();
        bdc1.deleted("blobId1", Collections.singleton("/a"));
        bdc1.commitProgress(COMMIT_SUCCEDED);

        BlobDeletionCallback bdc2 = adbc.getBlobDeletionCallback();
        bdc2.deleted("blobId2", Collections.singleton("/b"));

        BlobDeletionCallback bdc3 = adbc.getBlobDeletionCallback();
        bdc3.deleted("blobId3", Collections.singleton("/c"));
        bdc3.commitProgress(COMMIT_SUCCEDED);

        long time = clock.getTimeIncreasing();
        clock.waitUntil(clock.getTime() + TimeUnit.MINUTES.toMillis(1));

        bdc2.commitProgress(COMMIT_SUCCEDED);

        adbc.purgeBlobsDeleted(time, blobStore);

        //blobId2 is committed later
        verifyBlobsDeleted("blobId1", "blobId3");
    }

    @Test
    public void uncommittedDeletionsMustNotBePurged() throws Exception {
        BlobDeletionCallback bdc1 = adbc.getBlobDeletionCallback();
        bdc1.deleted("blobId1", Collections.singleton("/a"));
        bdc1.commitProgress(COMMIT_FAILED);

        BlobDeletionCallback bdc2 = adbc.getBlobDeletionCallback();
        bdc2.deleted("blobId2", Collections.singleton("/b"));
        bdc2.commitProgress(COMMIT_SUCCEDED);

        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        //blobId2 is committed later
        verifyBlobsDeleted("blobId2");
    }

    @Test
    public void deleteBlobsDespiteFileExplicitlyPurgedBeforeRestart() throws Exception {
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId1", Collections.singleton("/a"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        clock.waitUntil(clock.getTime() + TimeUnit.MINUTES.toMillis(1));
        createBlobCollector();
        bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId2", Collections.singleton("/b"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        clock.waitUntil(clock.getTime() + TimeUnit.MINUTES.toMillis(1));
        createBlobCollector();
        bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId3", Collections.singleton("/c"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted("blobId1", "blobId2", "blobId3");
    }

    @Test
    public void multiThreadedCommits() throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        File rootDirectory = new File(blobCollectionRoot.getRoot(), "b");
        FileUtils.forceMkdir(rootDirectory);
        adbc = new ActiveDeletedBlobCollectorImpl(clock, rootDirectory, executorService);

        int numThreads = 4;
        int numBlobsPerThread = 500;

        List<Thread> threads = new ArrayList<>(numThreads);
        final AtomicInteger threadIndex = new AtomicInteger(0);
        for (; threadIndex.get() < numThreads; threadIndex.incrementAndGet()) {
            threads.add(new Thread(new Runnable() {
                private int thisThreadNum = threadIndex.get();
                @Override
                public void run() {
                    int blobCnt = 0;
                    while (blobCnt < numBlobsPerThread) {
                        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
                        for (; blobCnt < numBlobsPerThread;) {
                            String id = "Thread" + thisThreadNum + "Blob" + blobCnt;
                            bdc.deleted(id, Collections.singleton(id));
                            blobCnt++;
                            if (Math.random() > 0.5) {
                                break;
                            }
                        }
                        bdc.commitProgress(COMMIT_SUCCEDED);
                        try {
                            Thread.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }));
        }

        for (Thread t : threads) {
            t.start();
        }

        for (Thread t : threads) {
            t.join();
        }

        boolean timeout = executorService.awaitTermination(100, TimeUnit.MILLISECONDS);
        assertFalse(timeout);

        List<String> deletedChunks = new ArrayList<>(numThreads*numBlobsPerThread*2);
        for (int threadNum = 0; threadNum < numThreads; threadNum++) {
            for (int blobCnt = 0; blobCnt < numBlobsPerThread; blobCnt++) {
                String id = "Thread" + threadNum + "Blob" + blobCnt;
                Iterators.addAll(deletedChunks, blobStore.resolveChunks(id));
            }
        }

        // Blocking queue doesn't supply all the items immediately.
        // So, we'd push "MARKER*" blob ids and purge until some marker blob
        // gets purged. BUT, we'd time-out this activity in 3 seconds
        long until = Clock.SIMPLE.getTime() + TimeUnit.SECONDS.toMillis(3);
        List<String> markerChunks = Lists.newArrayList();
        int i = 0;
        while (Clock.SIMPLE.getTime() < until) {
            // Push commit with a marker blob-id and wait for it to be purged
            BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
            String markerBlobId = "MARKER-" + (i++);
            bdc.deleted(markerBlobId, Lists.newArrayList(markerBlobId));
            bdc.commitProgress(COMMIT_SUCCEDED);

            Iterators.addAll(markerChunks, blobStore.resolveChunks(markerBlobId));
            clock.waitUntil(clock.getTime() + TimeUnit.SECONDS.toMillis(5));
            adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

            if (blobStore.markerChunkDeleted) {
                break;
            }
        }

        assertTrue("Timed out while waiting for marker chunk to be purged", blobStore.markerChunkDeleted);

        // don't care how many marker blobs are purged
        blobStore.deletedChunkIds.removeAll(markerChunks);

        HashSet<String> list = new HashSet<>(deletedChunks);
        list.removeAll(blobStore.deletedChunkIds);
        assertTrue("size: " + list.size() + "; list: " + list.toString(), list.isEmpty());

        assertThat(blobStore.deletedChunkIds, containsInAnyOrder(deletedChunks.toArray()));
    }

    @Test
    public void inaccessibleWorkDirGivesNoop() throws Exception {
        assumeFalse(CIHelper.windows());

        File rootDir = blobCollectionRoot.getRoot();
        File unwritableExistingRootFolder = new File(rootDir, "existingRoot");
        FileUtils.forceMkdir(unwritableExistingRootFolder);
        File unwritableNonExistingRootFolder = new File(unwritableExistingRootFolder, "existingRoot");

        Path unwritableExistingPath = FileSystems.getDefault().getPath(unwritableExistingRootFolder.getPath());
        Files.setPosixFilePermissions(unwritableExistingPath,
                Sets.newSet(PosixFilePermission.OWNER_READ,
                        PosixFilePermission.GROUP_READ,
                        PosixFilePermission.OTHERS_READ));

        adbc = ActiveDeletedBlobCollectorFactory.newInstance(unwritableExistingRootFolder, newDirectExecutorService());
        assertEquals("Unwritable existing root folder must have NOOP active blob collector",
                ActiveDeletedBlobCollectorFactory.NOOP, adbc);

        adbc = ActiveDeletedBlobCollectorFactory.newInstance(unwritableNonExistingRootFolder, newDirectExecutorService());
        assertEquals("Unwritable non-existing root folder must have NOOP active blob collector",
                ActiveDeletedBlobCollectorFactory.NOOP, adbc);
    }

    @Test
    public void cancellablePurge() throws Exception {
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        for (int i = 0; i < 10; i++) {
            String id = "Blob" + i;
            bdc.deleted(id, Collections.singleton(id));
        }
        bdc.commitProgress(COMMIT_SUCCEDED);

        Semaphore purgeBlocker = new Semaphore(0);
        blobStore.callback = () -> purgeBlocker.acquireUninterruptibly();
        Thread purgeThread = new Thread(() -> {
            try {
                adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        purgeThread.setDaemon(true);
        purgeBlocker.release(10);//allow 5 deletes
        purgeThread.start();

        boolean deleted5 = waitFor(5000, () -> blobStore.deletedChunkIds.size() >= 10);
        assertTrue("Deleted " + blobStore.deletedChunkIds.size() + " chunks", deleted5);

        adbc.cancelBlobCollection();
        purgeBlocker.release(20);//release all that's there... this is more than needed, btw.

        boolean deleted6 = waitFor(5000, () -> blobStore.deletedChunkIds.size() >= 12);
        assertTrue("Haven't deleted another blob which was locked earlier.", deleted6);

        boolean cancelWorked = waitFor(5000, () -> !purgeThread.isAlive());
        assertTrue("Cancel didn't let go of purge thread in 2 seconds", cancelWorked);

        assertTrue("Cancelling purge must return asap", blobStore.deletedChunkIds.size() == 12);
    }

    @Test
    public void resumeCancelledPurge() throws Exception {
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        for (int i = 0; i < 10; i++) {
            String id = "Blob" + i;
            bdc.deleted(id, Collections.singleton(id));
        }
        bdc.commitProgress(COMMIT_SUCCEDED);

        Semaphore purgeBlocker = new Semaphore(0);
        blobStore.callback = () -> purgeBlocker.acquireUninterruptibly();
        Thread purgeThread = new Thread(() -> {
            try {
                adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        purgeThread.setDaemon(true);
        purgeBlocker.release(10);//allow 5 deletes
        purgeThread.start();

        waitFor(5000, () -> blobStore.deletedChunkIds.size() >= 10);

        adbc.cancelBlobCollection();
        purgeBlocker.release(22);//release all that's there... this is more than needed, btw.

        waitFor(5000, () -> blobStore.deletedChunkIds.size() >= 12);

        waitFor(5000, () -> !purgeThread.isAlive());

        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        // Resume can re-attempt to delete already deleted blobs. Hence, the need for for ">="
        assertEquals("All blobs must get deleted", 20, blobStore.deletedChunkIds.size());
    }

    @Test
    public void dontWarnWhileErrorsWhileDeletingBlobs() throws Exception {
        LogCustomizer warnLogCustomizer =
                LogCustomizer.forLogger(ActiveDeletedBlobCollectorFactory.class.getName()).enable(Level.WARN)
                        .contains("Exception occurred while ")
                        .create();

        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId1", Collections.singleton("/a"));
        bdc.deleted("blobId2", Collections.singleton("/b"));
        bdc.deleted("blobId3", Collections.singleton("/c"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        List<String> externallyDeletedChunks = Lists.newArrayList(blobStore.resolveChunks("blobId2"));
        blobStore.countDeleteChunks(externallyDeletedChunks, 0);

        warnLogCustomizer.starting();
        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);
        blobStore.deletedChunkIds.removeAll(externallyDeletedChunks);

        verifyBlobsDeleted("blobId1", "blobId3");

        assertEquals("No warn logs must show up: " + warnLogCustomizer.getLogs(), 0, warnLogCustomizer.getLogs().size());
        warnLogCustomizer.finished();

        bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId4", Collections.singleton("/d"));
        bdc.deleted("blobId5", Collections.singleton("/e"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        blobStore.resetLists();
        blobStore.failWithDSEForChunkIds.addAll(Lists.newArrayList(blobStore.resolveChunks("blobId4")));

        warnLogCustomizer.starting();
        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted("blobId3", "blobId5");

        assertEquals("No warn logs must show up", 0, warnLogCustomizer.getLogs().size());
        warnLogCustomizer.finished();

        bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId6", Collections.singleton("/f"));
        bdc.deleted("blobId7", Collections.singleton("/g"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        blobStore.resetLists();
        blobStore.failWithExceptionForChunkIds.addAll(Lists.newArrayList(blobStore.resolveChunks("blobId6")));

        warnLogCustomizer.starting();
        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted("blobId5", "blobId7");

        assertEquals("General exception must log a warn", 1, warnLogCustomizer.getLogs().size());
        warnLogCustomizer.finished();
    }

    @Test
    public void doDebugLogWhileErrorsWhileDeletingBlobs() throws Exception {
        LogCustomizer warnLogCustomizer =
                LogCustomizer.forLogger(ActiveDeletedBlobCollectorFactory.class.getName()).enable(Level.DEBUG)
                        .contains("Exception occurred while ")
                        .create();

        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId1", Collections.singleton("/a"));
        bdc.deleted("blobId2", Collections.singleton("/b"));
        bdc.deleted("blobId3", Collections.singleton("/c"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        List<String> externallyDeletedChunks = Lists.newArrayList(blobStore.resolveChunks("blobId2"));
        blobStore.countDeleteChunks(externallyDeletedChunks, 0);

        warnLogCustomizer.starting();
        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);
        blobStore.deletedChunkIds.removeAll(externallyDeletedChunks);

        verifyBlobsDeleted("blobId1", "blobId3");

        assertEquals("Should log on debug", 1, warnLogCustomizer.getLogs().size());
        warnLogCustomizer.finished();

        bdc = adbc.getBlobDeletionCallback();
        bdc.deleted("blobId4", Collections.singleton("/d"));
        bdc.deleted("blobId5", Collections.singleton("/e"));
        bdc.commitProgress(COMMIT_SUCCEDED);

        blobStore.resetLists();
        blobStore.failWithDSEForChunkIds.addAll(Lists.newArrayList(blobStore.resolveChunks("blobId4")));

        warnLogCustomizer.starting();
        adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);

        verifyBlobsDeleted("blobId3", "blobId5");

        assertEquals("Should log on debug", 1, warnLogCustomizer.getLogs().size());
        warnLogCustomizer.finished();
    }

    // OAK-6950
    @Test
    public void pauseMarkingDeletedBlobs() {
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        assertFalse("Active deletion should be safe by default", bdc.isMarkingForActiveDeletionUnsafe());

        adbc.flagActiveDeletionUnsafe(true);
        bdc = adbc.getBlobDeletionCallback();
        assertTrue("Active deletion should be unsafe", bdc.isMarkingForActiveDeletionUnsafe());

        adbc.flagActiveDeletionUnsafe(false);
        bdc = adbc.getBlobDeletionCallback();
        assertFalse("Active deletion should be safe after unpausing", bdc.isMarkingForActiveDeletionUnsafe());
    }

    // OAK-6950
    @Test
    public void pauseMarkingDeletedBlobsNOOP() {
        adbc = ActiveDeletedBlobCollectorFactory.NOOP;
        BlobDeletionCallback bdc = adbc.getBlobDeletionCallback();
        assertFalse("Active deletion should be safe by default", bdc.isMarkingForActiveDeletionUnsafe());

        adbc.flagActiveDeletionUnsafe(true);
        bdc = adbc.getBlobDeletionCallback();
        assertTrue("Active deletion should be unsafe", bdc.isMarkingForActiveDeletionUnsafe());

        adbc.flagActiveDeletionUnsafe(false);
        bdc = adbc.getBlobDeletionCallback();
        assertFalse("Active deletion should be safe after unpausing", bdc.isMarkingForActiveDeletionUnsafe());
    }

    private void verifyBlobsDeleted(String ... blobIds) throws IOException {
        List<String> chunkIds = new ArrayList<>();
        for (String blobId : blobIds) {
            chunkIds.addAll(Lists.newArrayList(blobStore.resolveChunks(blobId)));
        }

        assertThat(blobStore.deletedChunkIds, containsInAnyOrder(chunkIds.toArray()));
    }

    class ChunkDeletionTrackingBlobStore implements GarbageCollectableBlobStore {
        Set<String> deletedChunkIds = org.apache.jackrabbit.guava.common.collect.Sets.newLinkedHashSet();
        Set<String> failWithDSEForChunkIds = org.apache.jackrabbit.guava.common.collect.Sets.newLinkedHashSet();
        Set<String> failWithExceptionForChunkIds = org.apache.jackrabbit.guava.common.collect.Sets.newLinkedHashSet();
        Runnable callback = null;
        volatile boolean markerChunkDeleted = false;

        @Override
        public String writeBlob(InputStream in) throws IOException {
            return null;
        }

        @Override
        public String writeBlob(InputStream in, BlobOptions options) throws IOException {
            return null;
        }

        @Override
        public int readBlob(String blobId, long pos, byte[] buff, int off, int length) throws IOException {
            return 0;
        }

        @Override
        public long getBlobLength(String blobId) throws IOException {
            return 0;
        }

        @Override
        public InputStream getInputStream(String blobId) throws IOException {
            return null;
        }

        @Override
        public String getBlobId(@NotNull String reference) {
            return null;
        }

        @Override
        public String getReference(@NotNull String blobId) {
            return null;
        }

        @Override
        public void setBlockSize(int x) {

        }

        @Override
        public String writeBlob(String tempFileName) throws IOException {
            return null;
        }

        @Override
        public int sweep() throws IOException {
            return 0;
        }

        @Override
        public void startMark() throws IOException {

        }

        @Override
        public void clearInUse() {

        }

        @Override
        public void clearCache() {

        }

        @Override
        public long getBlockSizeMin() {
            return 0;
        }

        @Override
        public Iterator<String> getAllChunkIds(long maxLastModifiedTime) throws Exception {
            return null;
        }

        @Override
        public boolean deleteChunks(List<String> chunkIds, long maxLastModifiedTime) throws Exception {
            setMarkerChunkDeletedFlag(chunkIds);
            deletedChunkIds.addAll(chunkIds);
            return true;
        }

        @Override
        public long countDeleteChunks(List<String> chunkIds, long maxLastModifiedTime) throws Exception {
            setMarkerChunkDeletedFlag(chunkIds);

            long count = 0;
            for (String chunkId : chunkIds) {
                if (deletedChunkIds.contains(chunkId)) {
                    throw new DataStoreException("Already deleted chunk: " + chunkId);
                } else if (failWithDSEForChunkIds.contains(chunkId)) {
                    throw new DataStoreException("Synthetically failing with DSE for chunk: " + chunkId);
                } else if (failWithExceptionForChunkIds.contains(chunkId)) {
                    throw new Exception("Synthetically failing with Exception for chunk: " + chunkId);
                }
                deletedChunkIds.add(chunkId);
                count++;
            }
            return count;
        }

        private void setMarkerChunkDeletedFlag(List<String> deletedChunkIds) {
            if (!markerChunkDeleted) {
                for (String chunkId : deletedChunkIds) {
                    if (chunkId.startsWith("MARKER")) {
                        markerChunkDeleted = true;
                        break;
                    }

                    if (callback != null) {
                        callback.run();
                    }
                }
            }
        }

        @Override
        public Iterator<String> resolveChunks(String blobId) throws IOException {
            return Iterators.forArray(blobId + "-1", blobId + "-2");
        }

        @Override
        public void close() throws Exception {
        }

        private void resetLists() {
            deletedChunkIds.clear();
            failWithDSEForChunkIds.clear();
            failWithExceptionForChunkIds.clear();
        }
    }

    private interface Condition {
        boolean evaluate();
    }

    private boolean waitFor(long timeout, Condition c)
            throws InterruptedException {
        long end = System.currentTimeMillis() + timeout;
        long remaining = end - System.currentTimeMillis();
        while (remaining > 0) {
            if (c.evaluate()) {
                return true;
            }

            Thread.sleep(100);//The constant is exaggerated
            remaining = end - System.currentTimeMillis();
        }
        return c.evaluate();
    }
}
