| /* |
| * 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.blob; |
| |
| import java.io.BufferedWriter; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import ch.qos.logback.classic.Level; |
| import com.google.common.base.Charsets; |
| import com.google.common.base.Optional; |
| import com.google.common.collect.Iterators; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.io.Closeables; |
| import com.google.common.io.Closer; |
| import com.google.common.io.Files; |
| import com.google.common.util.concurrent.Futures; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import com.google.common.util.concurrent.ListeningExecutorService; |
| import com.google.common.util.concurrent.MoreExecutors; |
| import com.google.common.util.concurrent.SettableFuture; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.jackrabbit.core.data.DataStoreException; |
| import org.apache.jackrabbit.oak.commons.FileIOUtils; |
| import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser; |
| import org.apache.jackrabbit.oak.commons.junit.LogCustomizer; |
| import org.apache.jackrabbit.oak.stats.DefaultStatisticsProvider; |
| import org.apache.jackrabbit.oak.stats.StatisticsProvider; |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.rules.TemporaryFolder; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import static org.hamcrest.CoreMatchers.containsString; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertThat; |
| import static org.junit.Assert.assertTrue; |
| |
| /** |
| * Tests for {@link UploadStagingCache}. |
| */ |
| public class UploadStagingCacheTest extends AbstractDataStoreCacheTest { |
| private static final Logger LOG = LoggerFactory.getLogger(UploadStagingCacheTest.class); |
| private static final String ID_PREFIX = "12345"; |
| |
| @Rule |
| public TemporaryFolder folder = new TemporaryFolder(new File("target")); |
| |
| private final Closer closer = Closer.create(); |
| private TestStagingUploader uploader; |
| private File root; |
| private CountDownLatch taskLatch; |
| private CountDownLatch callbackLatch; |
| private CountDownLatch afterExecuteLatch; |
| private TestExecutor executor; |
| private UploadStagingCache stagingCache; |
| private StatisticsProvider statsProvider; |
| private ScheduledExecutorService removeExecutor; |
| |
| @Before |
| public void setup() throws IOException { |
| root = folder.newFolder(); |
| init(0); |
| } |
| |
| private void init(int i) throws IOException { |
| init(i, new TestStagingUploader(folder.newFolder()), null); |
| } |
| |
| private void init(int i, TestStagingUploader testUploader, File homeDir) { |
| // uploader |
| uploader = testUploader; |
| |
| // create executor |
| taskLatch = new CountDownLatch(1); |
| callbackLatch = new CountDownLatch(1); |
| afterExecuteLatch = new CountDownLatch(i); |
| executor = new TestExecutor(1, taskLatch, callbackLatch, afterExecuteLatch); |
| |
| // stats |
| ScheduledExecutorService statsExecutor = Executors.newSingleThreadScheduledExecutor(); |
| closer.register(new ExecutorCloser(statsExecutor, 500, TimeUnit.MILLISECONDS)); |
| statsProvider = new DefaultStatisticsProvider(statsExecutor); |
| |
| removeExecutor = Executors.newSingleThreadScheduledExecutor(); |
| closer.register(new ExecutorCloser(removeExecutor, 500, TimeUnit.MILLISECONDS)); |
| |
| //cache instance |
| stagingCache = |
| UploadStagingCache.build(root, homeDir, 1/*threads*/, 8 * 1024 /* bytes */, |
| uploader, null/*cache*/, statsProvider, executor, null, 3000, 6000); |
| closer.register(stagingCache); |
| } |
| |
| @After |
| public void tear() throws IOException { |
| closer.close(); |
| } |
| |
| @Test |
| public void testZeroCache() throws IOException { |
| stagingCache = |
| UploadStagingCache.build(root, null, 1/*threads*/, 0 /* bytes */, |
| uploader, null/*cache*/, statsProvider, executor, null, 3000, 6000); |
| closer.register(stagingCache); |
| |
| File f = copyToFile(randomStream(0, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future = stagingCache.put(ID_PREFIX + 0, f); |
| assertFalse(future.isPresent()); |
| |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 0)); |
| assertEquals(0, Iterators.size(stagingCache.getAllIdentifiers())); |
| assertEquals(0, stagingCache.getStats().getMaxTotalWeight()); |
| } |
| |
| @Test |
| public void testDefaultStatsProvider() throws Exception { |
| stagingCache = |
| UploadStagingCache.build(root, null, 1/*threads*/, 8 * 1024 /* bytes */, |
| uploader, null/*cache*/, null, executor, null, 3000, 6000); |
| |
| // add load |
| File f = copyToFile(randomStream(0, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future = stagingCache.put(ID_PREFIX + 0, f); |
| assertTrue(future.isPresent()); |
| |
| assertNotNull(stagingCache.getIfPresent(ID_PREFIX + 0)); |
| |
| assertCacheStats(stagingCache, 1, 4 * 1024, 1, 1); |
| } |
| |
| /** |
| * Stage file successful upload. |
| * @throws Exception |
| */ |
| @Test |
| public void testAdd() throws Exception { |
| // add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| |
| assertFuture(futures, 0); |
| assertCacheStats(stagingCache, 0, 0, 1, 1); |
| } |
| |
| /** |
| * Stage file unsuccessful upload. |
| * @throws Exception |
| */ |
| @Test |
| public void testAddUploadException() throws Exception { |
| final AtomicInteger count = new AtomicInteger(0); |
| TestStagingUploader secondTimeUploader = new TestStagingUploader(folder.newFolder()) { |
| @Override |
| public void write(String id, File f) throws DataStoreException { |
| if (count.get() == 0) { |
| throw new DataStoreException("Error in writing blob"); |
| } |
| super.write(id, f); |
| } |
| }; |
| |
| // initialize staging cache using the mocked uploader |
| init(2, secondTimeUploader, null); |
| |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| waitFinish(futures); |
| |
| // assert file retrieved from staging cache |
| File ret = stagingCache.getIfPresent(ID_PREFIX + 0); |
| assertTrue(Files.equal(copyToFile(randomStream(0, 4 * 1024), folder.newFile()), ret)); |
| |
| assertEquals(1, stagingCache.getStats().getLoadCount()); |
| assertEquals(1, stagingCache.getStats().getLoadSuccessCount()); |
| assertCacheStats(stagingCache, 1, 4 * 1024, 1, 1); |
| |
| // Retry upload and wait for finish |
| count.incrementAndGet(); |
| ScheduledFuture<?> scheduledFuture = |
| removeExecutor.schedule(stagingCache.new RetryJob(), 0, TimeUnit.MILLISECONDS); |
| scheduledFuture.get(); |
| afterExecuteLatch.await(); |
| |
| // Now uploaded |
| ret = stagingCache.getIfPresent(ID_PREFIX + 0); |
| assertNull(ret); |
| assertTrue(Files.equal(copyToFile(randomStream(0, 4 * 1024), folder.newFile()), |
| secondTimeUploader.read(ID_PREFIX + 0))); |
| } |
| |
| /** |
| * Retrieve without adding. |
| * @throws Exception |
| */ |
| @Test |
| public void testGetNoAdd() throws Exception { |
| File ret = stagingCache.getIfPresent(ID_PREFIX + 0); |
| |
| // assert no file |
| assertNull(ret); |
| assertEquals(1, stagingCache.getStats().getLoadCount()); |
| assertCacheStats(stagingCache, 0, 0, 0, 0); |
| } |
| |
| /** |
| * GetAllIdentifiers without adding. |
| * @throws Exception |
| */ |
| @Test |
| public void testGetAllIdentifiersNoAdd() throws Exception { |
| Iterator<String> ids = stagingCache.getAllIdentifiers(); |
| assertFalse(ids.hasNext()); |
| } |
| |
| /** |
| * Invalidate without adding. |
| * @throws Exception |
| */ |
| @Test |
| public void testInvalidateNoAdd() throws Exception { |
| stagingCache.invalidate(ID_PREFIX + 0); |
| assertCacheStats(stagingCache, 0, 0, 0, 0); |
| } |
| |
| /** |
| * Error in putting file to stage. |
| * @throws Exception |
| */ |
| @Test |
| public void testPutMoveFileError() throws Exception { |
| File empty = new File(folder.getRoot(), String.valueOf(System.currentTimeMillis())); |
| assertFalse(empty.exists()); |
| Optional<SettableFuture<Integer>> future = stagingCache.put(ID_PREFIX + 0, empty); |
| // assert no file |
| assertFalse(future.isPresent()); |
| assertEquals(1, stagingCache.getStats().getMissCount()); |
| assertCacheStats(stagingCache, 0, 0, 0, 1); |
| } |
| |
| /** |
| * Put and retrieve different files concurrently. |
| * @throws Exception |
| */ |
| @Test |
| public void testGetAddDifferent() throws Exception { |
| //add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Create an async retrieve task |
| final SettableFuture<File> retFuture = SettableFuture.create(); |
| Thread t = new Thread(new Runnable() { |
| @Override public void run() { |
| retFuture.set(stagingCache.getIfPresent(ID_PREFIX + 1)); |
| } |
| }); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| t.start(); |
| |
| //assert no file retrieve |
| assertNull(retFuture.get()); |
| assertEquals(1, stagingCache.getStats().getLoadCount()); |
| |
| assertFuture(futures, 0); |
| assertCacheStats(stagingCache, 0, 0, 1, 1); |
| } |
| |
| /** |
| * Stage file when cache full. |
| * @throws Exception |
| */ |
| @Test |
| public void testCacheFullAdd() throws Exception { |
| // initialize cache to have restricted size |
| stagingCache = |
| UploadStagingCache.build(root, null, 1/*threads*/, 4 * 1024 /* bytes */, |
| uploader, null/*cache*/, statsProvider, executor, null, 3000, 6000); |
| closer.register(stagingCache); |
| |
| // add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Add another load |
| File f2 = copyToFile(randomStream(1, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future2 = stagingCache.put(ID_PREFIX + 1, f2); |
| assertFalse(future2.isPresent()); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| assertFuture(futures, 0); |
| |
| // Try 2nd upload again |
| Optional<SettableFuture<Integer>> future = stagingCache.put(ID_PREFIX + 1, f2); |
| futures = Lists.newArrayList(); |
| if (future.isPresent()) { |
| futures.add(future.get()); |
| } |
| assertFuture(futures, 1); |
| |
| assertCacheStats(stagingCache, 0, 0, 2, 3); |
| } |
| |
| /** |
| * GetAllIdentifiers after staging before upload. |
| * @throws Exception |
| */ |
| @Test |
| public void testGetAllIdentifiers() throws Exception { |
| // add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Check getAllIdentifiers |
| Iterator<String> idsIter = stagingCache.getAllIdentifiers(); |
| assertEquals(ID_PREFIX + 0, Iterators.getOnlyElement(idsIter)); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| assertFuture(futures, 0); |
| |
| assertCacheStats(stagingCache, 0, 0, 1, 1); |
| |
| // Should not return anything |
| idsIter = stagingCache.getAllIdentifiers(); |
| assertEquals(0, Iterators.size(idsIter)); |
| } |
| |
| /** |
| * Invalidate after staging before upload. |
| * @throws Exception |
| */ |
| @Test |
| public void testInvalidate() throws Exception { |
| // add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Check invalidate |
| stagingCache.invalidate(ID_PREFIX + 0); |
| File file = stagingCache.getIfPresent(ID_PREFIX + 0); |
| assertNull(file); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| waitFinish(futures); |
| |
| assertCacheStats(stagingCache, 0, 0, 1, 1); |
| |
| // Should not return anything |
| file = stagingCache.getIfPresent(ID_PREFIX + 0); |
| assertNull(file); |
| } |
| |
| /** |
| * Stage same file concurrently. |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentSameAdd() throws Exception { |
| LOG.info("Starting testConcurrentSameAdd"); |
| |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| File f = copyToFile(randomStream(0, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future2 = stagingCache.put(ID_PREFIX + 0, f); |
| assertTrue(future2.isPresent()); |
| assertEquals(future2.get().get().intValue(), 0); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| assertFuture(futures, 0); |
| |
| assertCacheStats(stagingCache, 0, 0, 1, 2); |
| |
| LOG.info("Finished testConcurrentSameAdd"); |
| } |
| |
| /** |
| * Stage request same file concurrently. |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentSameAddRequest() throws Exception { |
| LOG.info("Starting testConcurrentSameAddRequest"); |
| |
| closer.close(); |
| ListeningExecutorService executorService = |
| MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); |
| |
| List<ListenableFuture<Integer>> futures = Lists.newArrayList(); |
| CountDownLatch moveLatch = new CountDownLatch(1); |
| init(1, new TestStagingUploader(folder.newFolder(), moveLatch), null); |
| |
| //1st request |
| ListenableFuture<Boolean> resultReq1 = putThread(folder, executorService, futures); |
| Thread.sleep(100); |
| |
| //2nd Request |
| ListenableFuture<Boolean> resultReq2 = putThread(folder, executorService, futures); |
| Thread.sleep(200); |
| |
| // Allow any thread to start moving |
| moveLatch.countDown(); |
| |
| assertTrue(resultReq1.get()); |
| assertTrue(resultReq2.get()); |
| |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| |
| assertFuture(futures, 0); |
| assertCacheStats(stagingCache, 0, 0, 1, 2); |
| |
| LOG.info("Finished testConcurrentSameAddRequest"); |
| } |
| |
| /** |
| * Stage different files concurrently |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentDifferentAdd() throws Exception { |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Add diff load |
| File f2 = copyToFile(randomStream(1, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future2 = stagingCache.put(ID_PREFIX + 1, f2); |
| if (future2.isPresent()) { |
| futures.add(future2.get()); |
| } |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| assertFuture(futures, 0, 1); |
| |
| assertCacheStats(stagingCache, 0, 0, 2, 2); |
| } |
| |
| /** |
| * Concurrently retrieve after stage but before upload. |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentGetDelete() throws Exception { |
| ListeningExecutorService executorService = |
| MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); |
| closer.register(new ExecutorCloser(executorService)); |
| |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // Get a handle to the file and open stream |
| File file = stagingCache.getIfPresent(ID_PREFIX + 0); |
| final InputStream fStream = Files.asByteSource(file).openStream(); |
| |
| // task to copy the steam to a file simulating read from the stream |
| File temp = folder.newFile(); |
| CountDownLatch copyThreadLatch = new CountDownLatch(1); |
| SettableFuture<File> future1 = |
| copyStreamThread(executorService, fStream, temp, copyThreadLatch); |
| |
| //start |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| |
| waitFinish(futures); |
| |
| // trying copying now |
| copyThreadLatch.countDown(); |
| future1.get(); |
| |
| assertTrue(Files.equal(temp, uploader.read(ID_PREFIX + 0))); |
| } |
| |
| /** |
| * Concurrently stage and trigger delete after upload for same file. |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentPutDeleteSame() throws Exception { |
| testConcurrentPutDelete(0); |
| } |
| |
| /** |
| * Concurrently stage and trigger delete after upload for different file. |
| * @throws Exception |
| */ |
| @Test |
| public void testConcurrentPutDeleteDifferent() throws Exception { |
| testConcurrentPutDelete(1); |
| } |
| |
| private void testConcurrentPutDelete(int diff) throws Exception { |
| ListeningExecutorService executorService = |
| MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); |
| closer.register(new ExecutorCloser(executorService)); |
| //start immediately |
| taskLatch.countDown(); |
| |
| // Add immediately |
| List<ListenableFuture<Integer>> futures = put(folder); |
| |
| // New task to put another file |
| File f2 = copyToFile(randomStream(diff, 4 * 1024), folder.newFile()); |
| CountDownLatch putThreadLatch = new CountDownLatch(1); |
| CountDownLatch triggerLatch = new CountDownLatch(1); |
| SettableFuture<Optional<SettableFuture<Integer>>> future1 = |
| putThread(executorService, diff, f2, stagingCache, putThreadLatch, triggerLatch); |
| putThreadLatch.countDown(); |
| |
| // wait for put thread to go ahead |
| callbackLatch.countDown(); |
| ScheduledFuture<?> scheduledFuture = |
| removeExecutor.schedule(stagingCache.new RemoveJob(), 0, TimeUnit.MILLISECONDS); |
| triggerLatch.await(); |
| if (future1.get().isPresent()) { |
| futures.add(future1.get().get()); |
| } |
| |
| ListenableFuture<List<Integer>> listListenableFuture = Futures.successfulAsList(futures); |
| try { |
| listListenableFuture.get(); |
| scheduledFuture.get(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| |
| assertTrue(Files.equal(copyToFile(randomStream(0, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 0))); |
| assertTrue(Files.equal(copyToFile(randomStream(diff, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + diff))); |
| } |
| |
| /** |
| * Test build on start. |
| * @throws Exception |
| */ |
| @Test |
| public void testBuild() throws Exception { |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| // Close before uploading finished |
| closer.close(); |
| |
| // Start again |
| init(1); |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| afterExecuteLatch.await(); |
| |
| waitFinish(futures); |
| |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 0)); |
| assertTrue(Files.equal(copyToFile(randomStream(0, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 0))); |
| assertCacheStats(stagingCache, 0, 0, 1, 1); |
| } |
| |
| /** |
| * Test build on start with more files available in terms of total size in the upload cache. |
| * @throws Exception |
| */ |
| @Test |
| public void testBuildMoreThanCacheSize() throws Exception { |
| closer.close(); |
| |
| // create load greater than the cache size upgrades or cache size changes |
| File f1 = copyToFile(randomStream(1, 4 * 1024), |
| DataStoreCacheUtils.getFile(ID_PREFIX + "1", new File(root, "upload"))); |
| File f2 = copyToFile(randomStream(2, 4 * 1024), |
| DataStoreCacheUtils.getFile(ID_PREFIX + "2", new File(root, "upload"))); |
| File f3 = copyToFile(randomStream(3, 4 * 1024), |
| DataStoreCacheUtils.getFile(ID_PREFIX + "3", new File(root, "upload"))); |
| // Directly add files to staging dir simulating an upgrade scenario |
| |
| // Start staging cache |
| init(3); |
| |
| List<ListenableFuture<Integer>> futures = put(folder); |
| // Not staged as already full |
| assertTrue(futures.isEmpty()); |
| |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| afterExecuteLatch.await(); |
| |
| waitFinish(futures); |
| |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 1)); |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 2)); |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 3)); |
| |
| // Initial files should have been uploaded |
| assertTrue(Files.equal(copyToFile(randomStream(1, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 1))); |
| assertTrue(Files.equal(copyToFile(randomStream(2, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 2))); |
| assertTrue(Files.equal(copyToFile(randomStream(3, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 3))); |
| assertCacheStats(stagingCache, 0, 0, 3, 4); |
| } |
| |
| /** |
| * Test upgrade with build on start. |
| * @throws Exception |
| */ |
| @Test |
| public void testUpgrade() throws Exception { |
| // Add load |
| List<ListenableFuture<Integer>> futures = put(folder); |
| // Close before uploading finished |
| closer.close(); |
| |
| // Create pre-upgrade load |
| File home = folder.newFolder(); |
| File pendingUploadsFile = new File(home, DataStoreCacheUpgradeUtils.UPLOAD_MAP); |
| createUpgradeLoad(home, pendingUploadsFile); |
| |
| // Start again |
| init(2, new TestStagingUploader(folder.newFolder()), home); |
| |
| taskLatch.countDown(); |
| callbackLatch.countDown(); |
| afterExecuteLatch.await(); |
| |
| waitFinish(futures); |
| |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 0)); |
| assertTrue(Files.equal(copyToFile(randomStream(0, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 0))); |
| |
| assertUpgrade(pendingUploadsFile); |
| |
| assertCacheStats(stagingCache, 0, 0, 2, 2); |
| } |
| |
| @Test |
| public void testUpgradeCompromisedSerializedMap() throws IOException { |
| // Close the init setup |
| closer.close(); |
| |
| // Create pre-upgrade load |
| File home = folder.newFolder(); |
| File pendingUploadsFile = new File(home, DataStoreCacheUpgradeUtils.UPLOAD_MAP); |
| createGibberishLoad(home, pendingUploadsFile); |
| |
| LogCustomizer lc = LogCustomizer.forLogger(DataStoreCacheUpgradeUtils.class.getName()) |
| .filter(Level.WARN) |
| .enable(Level.WARN) |
| .create(); |
| lc.starting(); |
| // Start |
| init(2, new TestStagingUploader(folder.newFolder()), home); |
| assertThat(lc.getLogs().toString(), containsString("Error in reading pending uploads map")); |
| } |
| |
| /** -------------------- Helper methods ----------------------------------------------------**/ |
| |
| private void createUpgradeLoad(File home, File pendingUploadFile) throws IOException { |
| String id = ID_PREFIX + 1; |
| copyToFile(randomStream(1, 4 * 1024), getFile(id, root)); |
| String name = id.substring(0, 2) + "/" + id.substring(2, 4) + "/" + id; |
| Map<String, Long> pendingUploads = Maps.newHashMap(); |
| pendingUploads.put(name, System.currentTimeMillis()); |
| serializeMap(pendingUploads, pendingUploadFile); |
| } |
| |
| |
| private void createGibberishLoad(File home, File pendingUploadFile) throws IOException { |
| BufferedWriter writer = null; |
| try { |
| writer = Files.newWriter(pendingUploadFile, Charsets.UTF_8); |
| FileIOUtils.writeAsLine(writer, "jerhgiuheirghoeoorqehgsjlwjpfkkwpkf", false); |
| } finally { |
| Closeables.close(writer, true); |
| } |
| } |
| |
| private void assertUpgrade(File pendingUploadFile) throws IOException { |
| assertNull(stagingCache.getIfPresent(ID_PREFIX + 1)); |
| assertTrue(Files.equal(copyToFile(randomStream(1, 4 * 1024), folder.newFile()), |
| uploader.read(ID_PREFIX + 1))); |
| assertFalse(pendingUploadFile.exists()); |
| } |
| |
| private static SettableFuture<File> copyStreamThread(ListeningExecutorService executor, |
| final InputStream fStream, final File temp, final CountDownLatch start) { |
| final SettableFuture<File> future = SettableFuture.create(); |
| executor.submit(new Runnable() { |
| @Override public void run() { |
| try { |
| LOG.info("Waiting for start of copying"); |
| start.await(); |
| LOG.info("Starting copy of [{}]", temp); |
| FileUtils.copyInputStreamToFile(fStream, temp); |
| LOG.info("Finished retrieve"); |
| future.set(temp); |
| } catch (Exception e) { |
| LOG.info("Exception in get", e); |
| } |
| } |
| }); |
| return future; |
| } |
| |
| private static SettableFuture<Optional<SettableFuture<Integer>>> putThread( |
| ListeningExecutorService executor, final int seed, final File f, final UploadStagingCache cache, |
| final CountDownLatch start, final CountDownLatch trigger) { |
| final SettableFuture<Optional<SettableFuture<Integer>>> future = SettableFuture.create(); |
| executor.submit(new Runnable() { |
| @Override public void run() { |
| try { |
| LOG.info("Waiting for start to put"); |
| start.await(); |
| LOG.info("Starting put"); |
| trigger.countDown(); |
| Optional<SettableFuture<Integer>> opt = cache.put(ID_PREFIX + seed, f); |
| LOG.info("Finished put"); |
| future.set(opt); |
| } catch (Exception e) { |
| LOG.info("Exception in get", e); |
| } |
| } |
| }); |
| return future; |
| } |
| |
| private void waitFinish(List<ListenableFuture<Integer>> futures) { |
| ListenableFuture<List<Integer>> listListenableFuture = Futures.successfulAsList(futures); |
| try { |
| listListenableFuture.get(); |
| ScheduledFuture<?> scheduledFuture = |
| removeExecutor.schedule(stagingCache.new RemoveJob(), 0, TimeUnit.MILLISECONDS); |
| scheduledFuture.get(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| |
| private ListenableFuture<Boolean> putThread(TemporaryFolder folder, ListeningExecutorService executorService, List<ListenableFuture<Integer>> futures) { |
| closer.register(new ExecutorCloser(executorService)); |
| |
| ListenableFuture<Boolean> result = executorService.submit(new Callable<Boolean>() { |
| @Override public Boolean call() { |
| try { |
| LOG.info("Starting put"); |
| futures.addAll(put(folder)); |
| LOG.info("Finished put"); |
| File f = stagingCache.getIfPresent(ID_PREFIX + 0); |
| LOG.info("Retrieved file {}, {}", f, f.exists()); |
| return f != null && f.exists(); |
| } catch (Exception e) { |
| LOG.info("Exception in get", e); |
| } |
| return false; |
| } |
| }); |
| |
| return result; |
| } |
| |
| private List<ListenableFuture<Integer>> put(TemporaryFolder folder) |
| throws IOException { |
| File f = copyToFile(randomStream(0, 4 * 1024), folder.newFile()); |
| Optional<SettableFuture<Integer>> future = stagingCache.put(ID_PREFIX + 0, f); |
| List<ListenableFuture<Integer>> futures = Lists.newArrayList(); |
| if (future.isPresent()) { |
| futures.add(future.get()); |
| } |
| return futures; |
| } |
| |
| private void assertFuture(List<ListenableFuture<Integer>> futures, int... seeds) |
| throws Exception { |
| waitFinish(futures); |
| |
| for (int i = 0; i < seeds.length; i++) { |
| File upload = uploader.read(ID_PREFIX + seeds[i]); |
| assertFile(upload, seeds[i], folder); |
| } |
| } |
| |
| private void assertFile(File f, int seed, TemporaryFolder folder) throws IOException { |
| assertTrue(f.exists()); |
| File temp = copyToFile(randomStream(seed, 4 * 1024), folder.newFile()); |
| assertTrue("Uploaded file content differs", FileUtils.contentEquals(temp, f)); |
| } |
| |
| private static void assertCacheStats(UploadStagingCache cache, long elems, long weight, |
| long hits, long count) { |
| assertEquals(elems, cache.getStats().getElementCount()); |
| assertEquals(weight, cache.getStats().estimateCurrentWeight()); |
| assertEquals(hits, cache.getStats().getHitCount()); |
| assertEquals(count, cache.getStats().getRequestCount()); |
| } |
| } |
