You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by ca...@apache.org on 2017/05/28 22:20:19 UTC
svn commit: r1796552 - in /jackrabbit/oak/trunk/oak-lucene/src:
main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/
test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/
Author: catholicon
Date: Sun May 28 22:20:19 2017
New Revision: 1796552
URL: http://svn.apache.org/viewvc?rev=1796552&view=rev
Log:
OAK-2808: Active deletion of 'deleted' Lucene index files from DataStore without relying on full scale Blob GC
Add active deleted blob collector implementation and test
Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java (with props)
jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java (with props)
Added: jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java?rev=1796552&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java (added)
+++ jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java Sun May 28 22:20:19 2017
@@ -0,0 +1,373 @@
+/*
+ * 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 com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.apache.commons.io.filefilter.RegexFileFilter;
+import org.apache.jackrabbit.oak.commons.benchmark.PerfLogger;
+import org.apache.jackrabbit.oak.plugins.index.IndexCommitCallback;
+import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
+import org.apache.jackrabbit.oak.stats.Clock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+public class ActiveDeletedBlobCollectorFactory {
+ public interface ActiveDeletedBlobCollector {
+ BlobDeletionCallback getBlobDeletionCallback();
+ void purgeBlobsDeleted(long before, GarbageCollectableBlobStore blobStore);
+ }
+// LOG.info("Added {} to delete.", info);
+
+ public static ActiveDeletedBlobCollector NOOP = new ActiveDeletedBlobCollector() {
+ @Override
+ public BlobDeletionCallback getBlobDeletionCallback() {
+ return BlobDeletionCallback.NOOP;
+ }
+
+ @Override
+ public void purgeBlobsDeleted(long before, GarbageCollectableBlobStore blobStore) {
+
+ }
+ };
+
+ public interface BlobDeletionCallback extends IndexCommitCallback {
+ void deleted(String blobId, Iterable<String> ids);
+ BlobDeletionCallback NOOP = new BlobDeletionCallback() {
+ @Override
+ public void deleted(String blobId, Iterable<String> ids) {
+ }
+
+ @Override
+ public void commitProgress(IndexProgress indexProgress) {
+ }
+ };
+ }
+
+ public static ActiveDeletedBlobCollector newInstance(@Nonnull File rootDirectory,
+ ExecutorService executorService) {
+ try {
+ FileUtils.forceMkdir(rootDirectory);
+ } catch (IOException ioe) {
+ ActiveDeletedBlobCollectorImpl.LOG.warn("Disabling active blob collector as we couldn't not create folder: "
+ + rootDirectory, ioe);
+ return NOOP;
+ }
+ if(!rootDirectory.canRead() || !rootDirectory.canWrite() || !rootDirectory.canExecute()) {
+ ActiveDeletedBlobCollectorImpl.LOG.warn("Insufficient access in directory - {}. Disabling active blob collector",
+ rootDirectory);
+ return NOOP;
+ }
+ return new ActiveDeletedBlobCollectorImpl(rootDirectory, executorService);
+ }
+
+ /**
+ * Blob collector which takes *no* guarantees about checking whether the
+ * blob might be referred by paths other than one for which it is notified
+ * due deleted blob
+ */
+ static class ActiveDeletedBlobCollectorImpl implements ActiveDeletedBlobCollector {
+
+ private static PerfLogger PERF_LOG = new PerfLogger(
+ LoggerFactory.getLogger(ActiveDeletedBlobCollectorImpl.class.getName() + ".perf"));
+ private static Logger LOG = LoggerFactory.getLogger(ActiveDeletedBlobCollectorImpl.class.getName());
+
+ private final Clock clock;
+
+ private final File rootDirectory;
+
+ private final ExecutorService executorService;
+
+ private static final String BLOB_FILE_PATTERN_PREFIX = "blobs-";
+ private static final String BLOB_FILE_PATTERN_SUFFIX = ".txt";
+ private static final String BLOB_FILE_PATTERN = BLOB_FILE_PATTERN_PREFIX + "%s" + BLOB_FILE_PATTERN_SUFFIX;
+ private static final IOFileFilter blobFileNameFilter = new RegexFileFilter("blobs-.*\\.txt");
+
+ private final BlockingQueue<BlobIdInfoStruct> deletedBlobs;
+ private final DeletedBlobsFileWriter deletedBlobsFileWriter;
+
+ /**
+ * @param rootDirectory directory that may be used by this instance to
+ * keep temporary data (e.g. reported deleted blob-ids).
+ * @param executorService executor service to asynchronously flush deleted blobs
+ * to a file.
+ */
+ ActiveDeletedBlobCollectorImpl(@Nonnull File rootDirectory, @Nonnull ExecutorService executorService) {
+ this(Clock.SIMPLE, rootDirectory, executorService);
+ }
+
+ ActiveDeletedBlobCollectorImpl(Clock clock, @Nonnull File rootDirectory,
+ @Nonnull ExecutorService executorService) {
+ this.clock = clock;
+ this.rootDirectory = rootDirectory;
+ this.executorService = executorService;
+ this.deletedBlobs = new ArrayBlockingQueue<>(1000); //1000 items should be ok for async index commits
+ this.deletedBlobsFileWriter = new DeletedBlobsFileWriter();
+ }
+
+ public void purgeBlobsDeleted(long before, @Nonnull GarbageCollectableBlobStore blobStore) {
+ long numBlobsDeleted = 0;
+ long numChunksDeleted = 0;
+
+ long lastCheckedBlobTimestamp = readLastCheckedBlobTimestamp();
+ deletedBlobsFileWriter.releaseInUseFile();
+ for (File deletedBlobListFile : FileUtils.listFiles(rootDirectory, blobFileNameFilter, null)) {
+ if (deletedBlobListFile.getName().equals(deletedBlobsFileWriter.inUseFileName)) {
+ continue;
+ }
+ LOG.debug("Purging blobs from {}", deletedBlobListFile);
+ long timestamp;
+ try {
+ timestamp = getTimestampFromBlobFileName(deletedBlobListFile.getName());
+ } catch (IllegalArgumentException iae) {
+ LOG.warn("Couldn't extract timestamp from filename - " + deletedBlobListFile, iae);
+ continue;
+ }
+ if (timestamp < before) {
+ try {
+ for (String deletedBlobLine : FileUtils.readLines(deletedBlobListFile, (String) null)) {
+ String[] parsedDeletedBlobIdLine = deletedBlobLine.split("\\|", 3);
+ if (parsedDeletedBlobIdLine.length != 3) {
+ LOG.warn("Unparseable line ({}) in file {}. It won't be retried.",
+ parsedDeletedBlobIdLine, deletedBlobListFile);
+ } else {
+ String deletedBlobId = parsedDeletedBlobIdLine[0];
+ try {
+ long blobDeletionTimestamp = Long.valueOf(parsedDeletedBlobIdLine[1]);
+
+ if (blobDeletionTimestamp < lastCheckedBlobTimestamp) {
+ continue;
+ }
+
+ if (blobDeletionTimestamp >= before) {
+ break;
+ }
+
+ long deleted = blobStore.countDeleteChunks(
+ Lists.newArrayList(blobStore.resolveChunks(deletedBlobId)), 0);
+ if (deleted < 1) {
+ LOG.warn("Blob {} in file {} not deleted", deletedBlobId, deletedBlobListFile);
+ } else {
+ numBlobsDeleted++;
+ numChunksDeleted += deleted;
+ }
+ } catch (NumberFormatException nfe) {
+ LOG.warn("Couldn't parse blobTimestamp(" + parsedDeletedBlobIdLine[1] +
+ "). deletedBlobLine - " + deletedBlobLine +
+ "; file - " + deletedBlobListFile.getName(), nfe);
+ } catch (Exception e) {
+ LOG.warn("Exception occurred while attempting to delete blob " + deletedBlobId, e);
+ }
+ }
+ }
+ } catch (IOException ioe) {
+ //log error and continue
+ LOG.warn("Couldn't read deleted blob list file - " + deletedBlobListFile, ioe);
+ }
+
+ if (!deletedBlobListFile.delete()) {
+ LOG.warn("File {} couldn't be deleted while all blobs listed in it have been purged", deletedBlobListFile);
+ }
+ } else {
+ LOG.debug("Skipping {} as its timestamp is newer than {}", deletedBlobListFile.getName(), before);
+ }
+ }
+ LOG.info("Deleted {} blobs contained in {} chunks", numBlobsDeleted, numChunksDeleted);
+ writeOutLastCheckedBlobTimestamp(before);
+ }
+
+ private long readLastCheckedBlobTimestamp() {
+ File blobCollectorInfoFile = new File(rootDirectory, "collection-info.txt");
+ InputStream is = null;
+ Properties p;
+ try {
+ is = new BufferedInputStream(new FileInputStream(blobCollectorInfoFile));
+ p = new Properties();
+ p.load(is);
+ } catch (IOException e) {
+ LOG.info("Couldn't read last checked blob timestamp... would do a bit more scan");
+ return -1;
+ } finally {
+ org.apache.commons.io.IOUtils.closeQuietly(is);
+ }
+
+ String resString = p.getProperty("last-checked-blob-timestamp");
+ if (resString == null) {
+ LOG.warn("Couldn't fine last checked blob timestamp property in collection-info.txt");
+ return -1;
+ }
+
+ try {
+ return Long.valueOf(resString);
+ } catch (NumberFormatException nfe) {
+ LOG.warn("Couldn't read last checked blob timestamp '" + resString + "' as long", nfe);
+ return -1;
+ }
+ }
+
+ private void writeOutLastCheckedBlobTimestamp(long timestamp) {
+ Properties p = new Properties();
+ p.setProperty("last-checked-blob-timestamp", String.valueOf(timestamp));
+ File blobCollectorInfoFile = new File(rootDirectory, "collection-info.txt");
+ OutputStream os = null;
+ try {
+ os = new BufferedOutputStream(new FileOutputStream(blobCollectorInfoFile));
+ p.store(os, "Last checked blob timestamp");
+ } catch (IOException e) {
+ LOG.warn("Couldn't write out last checked blob timestamp(" + timestamp + ")", e);
+ } finally {
+ org.apache.commons.io.IOUtils.closeQuietly(os);
+ }
+
+ }
+
+ public BlobDeletionCallback getBlobDeletionCallback() throws IllegalStateException {
+ return new DeletedBlobCollector();
+ }
+
+ static long getTimestampFromBlobFileName(String filename) throws IllegalArgumentException {
+ checkArgument(filename.startsWith(BLOB_FILE_PATTERN_PREFIX),
+ "Filename(%s) must start with %s", filename, BLOB_FILE_PATTERN_PREFIX);
+ checkArgument(filename.endsWith(BLOB_FILE_PATTERN_SUFFIX),
+ "Filename(%s) must end with %s", filename, BLOB_FILE_PATTERN_SUFFIX);
+ String timestampStr = filename.substring(
+ BLOB_FILE_PATTERN_PREFIX.length(),
+ filename.length() - BLOB_FILE_PATTERN_SUFFIX.length());
+
+ return Long.parseLong(timestampStr);
+ }
+
+ private void addDeletedBlobs(Collection<BlobIdInfoStruct> deletedBlobs) {
+ for (BlobIdInfoStruct info : deletedBlobs) {
+ try {
+ if (!this.deletedBlobs.offer(info, 1, TimeUnit.SECONDS)) {
+ LOG.warn("Timed out while offer-ing {} into queue.", info);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while adding " + info, e);
+ }
+ }
+ LOG.debug("Added {} to be flushed", deletedBlobs.size());
+ deletedBlobsFileWriter.scheduleFileFlushIfNeeded();
+ }
+
+ private class DeletedBlobsFileWriter implements Runnable {
+ private final AtomicBoolean fileFlushScheduled = new AtomicBoolean(false);
+
+ private volatile String inUseFileName = null;
+
+ private synchronized void flushDeletedBlobs() {
+ List<BlobIdInfoStruct> localDeletedBlobs = new LinkedList<>();
+ while (deletedBlobs.peek() != null) {
+ localDeletedBlobs.add(deletedBlobs.poll());
+ }
+ File outFile = new File(rootDirectory, getBlobFileName());
+ try {
+ long start = PERF_LOG.start();
+ FileUtils.writeLines(outFile, localDeletedBlobs, true);
+ PERF_LOG.end(start, 1, "Flushing deleted blobs");
+ } catch (IOException e) {
+ LOG.error("Couldn't write out to " + outFile, e);
+ }
+ }
+
+ private void scheduleFileFlushIfNeeded() {
+ if (fileFlushScheduled.compareAndSet(false, true)) {
+ executorService.submit(this);
+ }
+ }
+
+ private synchronized void releaseInUseFile() {
+ inUseFileName = null;
+ }
+
+ @Override
+ public void run() {
+ flushDeletedBlobs();
+ fileFlushScheduled.set(false);
+ }
+
+ private String getBlobFileName() {
+ if (inUseFileName == null) {
+ inUseFileName = String.format(BLOB_FILE_PATTERN, clock.getTime());
+ }
+ return inUseFileName;
+ }
+ }
+
+ private class DeletedBlobCollector implements BlobDeletionCallback {
+ List<BlobIdInfoStruct> deletedBlobs = new ArrayList<>();
+
+ @Override
+ public void deleted(String blobId, Iterable<String> ids) {
+ deletedBlobs.add(new BlobIdInfoStruct(blobId, ids));
+ }
+
+ @Override
+ public void commitProgress(IndexProgress indexProgress) {
+ if (indexProgress != IndexProgress.COMMIT_SUCCEDED && indexProgress != IndexProgress.COMMIT_FAILED) {
+ LOG.debug("We only care for commit success/failure");
+ return;
+ }
+ if (indexProgress == IndexProgress.COMMIT_SUCCEDED) {
+ addDeletedBlobs(deletedBlobs);
+ }
+
+ deletedBlobs.clear();
+ }
+ }
+
+ private class BlobIdInfoStruct {
+ final String blobId;
+ final Iterable<String> ids;
+
+ BlobIdInfoStruct(String blobId, Iterable<String> ids) {
+ this.blobId = blobId;
+ this.ids = ids;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s|%s|%s", blobId, clock.getTime(), Joiner.on("|").join(ids));
+ }
+ }
+ }
+}
Propchange: jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java?rev=1796552&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java (added)
+++ jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java Sun May 28 22:20:19 2017
@@ -0,0 +1,358 @@
+/*
+ * 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 com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import org.apache.commons.io.FileUtils;
+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.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.internal.util.collections.Sets;
+
+import javax.annotation.Nonnull;
+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.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
+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.assertThat;
+import static org.junit.Assert.assertTrue;
+
+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"), sameThreadExecutor());
+ }
+
+ @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 {
+ clock = Clock.SIMPLE;
+ ExecutorService executorService = Executors.newFixedThreadPool(3);
+ adbc = ActiveDeletedBlobCollectorFactory.newInstance(
+ new File(blobCollectionRoot.getRoot(), "b"), 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();
+ }
+
+ // Push one more commit to flush out any remaining ones
+ adbc.getBlobDeletionCallback().commitProgress(COMMIT_SUCCEDED);
+
+ executorService.awaitTermination(100, TimeUnit.MILLISECONDS);
+
+ 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));
+ }
+ }
+
+ adbc.purgeBlobsDeleted(clock.getTimeIncreasing(), blobStore);
+
+ assertThat(blobStore.deletedChunkIds, containsInAnyOrder(deletedChunks.toArray()));
+ }
+
+ @Test
+ public void inaccessibleWorkDirGivesNoop() throws Exception {
+ 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, sameThreadExecutor());
+ assertEquals("Unwritable existing root folder must have NOOP active blob collector",
+ ActiveDeletedBlobCollectorFactory.NOOP, adbc);
+
+ adbc = ActiveDeletedBlobCollectorFactory.newInstance(unwritableNonExistingRootFolder, sameThreadExecutor());
+ assertEquals("Unwritable non-existing root folder must have NOOP active blob collector",
+ ActiveDeletedBlobCollectorFactory.NOOP, adbc);
+ }
+
+ 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 {
+ List<String> deletedChunkIds = Lists.newArrayList();
+
+ @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(@Nonnull String reference) {
+ return null;
+ }
+
+ @Override
+ public String getReference(@Nonnull 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 {
+ deletedChunkIds.addAll(chunkIds);
+ return true;
+ }
+
+ @Override
+ public long countDeleteChunks(List<String> chunkIds, long maxLastModifiedTime) throws Exception {
+ deletedChunkIds.addAll(chunkIds);
+ return chunkIds.size();
+ }
+
+ @Override
+ public Iterator<String> resolveChunks(String blobId) throws IOException {
+ return Iterators.forArray(blobId + "-1", blobId + "-2");
+ }
+ }
+}
Propchange: jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ActiveDeletedBlobCollectorTest.java
------------------------------------------------------------------------------
svn:eol-style = native