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 to...@apache.org on 2018/03/20 10:54:10 UTC
svn commit: r1827292 [2/2] - in /jackrabbit/oak/trunk: ./ oak-segment-azure/
oak-segment-azure/src/ oak-segment-azure/src/main/
oak-segment-azure/src/main/java/ oak-segment-azure/src/main/java/org/
oak-segment-azure/src/main/java/org/apache/ oak-segmen...
Added: jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteAction.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteAction.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteAction.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteAction.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,62 @@
+/*
+ * 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.segment.azure.queue;
+
+import org.apache.jackrabbit.oak.segment.azure.AzureSegmentArchiveEntry;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+public class SegmentWriteAction {
+
+ private final AzureSegmentArchiveEntry indexEntry;
+
+ private final byte[] buffer;
+
+ private final int offset;
+
+ private final int length;
+
+ public SegmentWriteAction(AzureSegmentArchiveEntry indexEntry, byte[] buffer, int offset, int length) {
+ this.indexEntry = indexEntry;
+
+ this.buffer = new byte[length];
+ for (int i = 0; i < length; i++) {
+ this.buffer[i] = buffer[i + offset];
+ }
+ this.offset = 0;
+ this.length = length;
+ }
+
+ public UUID getUuid() {
+ return new UUID(indexEntry.getMsb(), indexEntry.getLsb());
+ }
+
+ public ByteBuffer toByteBuffer() {
+ return ByteBuffer.wrap(buffer, offset, length);
+ }
+
+ void passTo(SegmentWriteQueue.SegmentConsumer consumer) throws IOException {
+ consumer.consume(indexEntry, buffer, offset, length);
+ }
+
+ @Override
+ public String toString() {
+ return getUuid().toString();
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueue.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueue.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueue.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueue.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,277 @@
+/*
+ * 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.segment.azure.queue;
+
+import org.apache.jackrabbit.oak.segment.azure.AzureSegmentArchiveEntry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class SegmentWriteQueue implements Closeable {
+
+ public static final int THREADS = Integer.getInteger("oak.segment.azure.threads", 5);
+
+ private static final int QUEUE_SIZE = Integer.getInteger("oak.segment.org.apache.jackrabbit.oak.segment.azure.queue", 20);
+
+ private static final Logger log = LoggerFactory.getLogger(SegmentWriteQueue.class);
+
+ private final BlockingDeque<SegmentWriteAction> queue;
+
+ private final Map<UUID, SegmentWriteAction> segmentsByUUID;
+
+ private final ExecutorService executor;
+
+ private final ReadWriteLock flushLock;
+
+ private final SegmentConsumer writer;
+
+ private volatile boolean shutdown;
+
+ private final Object brokenMonitor = new Object();
+
+ private volatile boolean broken;
+
+ public SegmentWriteQueue(SegmentConsumer writer) {
+ this(writer, QUEUE_SIZE, THREADS);
+ }
+
+ SegmentWriteQueue(SegmentConsumer writer, int queueSize, int threadNo) {
+ this.writer = writer;
+ segmentsByUUID = new ConcurrentHashMap<>();
+ flushLock = new ReentrantReadWriteLock();
+
+ queue = new LinkedBlockingDeque<>(queueSize);
+ executor = Executors.newFixedThreadPool(threadNo + 1);
+ for (int i = 0; i < threadNo; i++) {
+ executor.submit(this::mainLoop);
+ }
+ executor.submit(this::emergencyLoop);
+ }
+
+ private void mainLoop() {
+ while (!shutdown) {
+ try {
+ waitWhileBroken();
+ if (shutdown) {
+ break;
+ }
+ consume();
+ } catch (SegmentConsumeException e) {
+ SegmentWriteAction segment = e.segment;
+ log.error("Can't persist the segment {}", segment.getUuid(), e.getCause());
+ try {
+ queue.put(segment);
+ } catch (InterruptedException e1) {
+ log.error("Can't re-add the segment {} to the queue. It'll be dropped.", segment.getUuid(), e1);
+ }
+ }
+ }
+ }
+
+ private void consume() throws SegmentConsumeException {
+ SegmentWriteAction segment = null;
+ try {
+ segment = queue.poll(100, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.error("Poll from queue interrupted", e);
+ }
+ if (segment != null) {
+ consume(segment);
+ }
+ }
+
+ private void consume(SegmentWriteAction segment) throws SegmentConsumeException {
+ try {
+ segment.passTo(writer);
+ } catch (IOException e) {
+ setBroken(true);
+ throw new SegmentConsumeException(segment, e);
+ }
+ synchronized (segmentsByUUID) {
+ segmentsByUUID.remove(segment.getUuid());
+ segmentsByUUID.notifyAll();
+ }
+ setBroken(false);
+ }
+
+ private void emergencyLoop() {
+ while (!shutdown) {
+ waitUntilBroken();
+ if (shutdown) {
+ break;
+ }
+
+ boolean success = false;
+ SegmentWriteAction segmentToRetry = null;
+ do {
+ try {
+ if (segmentToRetry == null) {
+ consume();
+ } else {
+ consume(segmentToRetry);
+ }
+ success = true;
+ } catch (SegmentConsumeException e) {
+ segmentToRetry = e.segment;
+ log.error("Can't persist the segment {}", segmentToRetry.getUuid(), e.getCause());
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e1) {
+ log.warn("Interrupted", e);
+ }
+ if (shutdown) {
+ log.error("Shutdown initiated. The segment {} will be dropped.", segmentToRetry.getUuid());
+ }
+ }
+ } while (!success && !shutdown);
+ }
+ }
+
+ public void addToQueue(AzureSegmentArchiveEntry indexEntry, byte[] data, int offset, int size) throws IOException {
+ waitWhileBroken();
+ if (shutdown) {
+ throw new IllegalStateException("Can't accept the new segment - shutdown in progress");
+ }
+
+ SegmentWriteAction action = new SegmentWriteAction(indexEntry, data, offset, size);
+ flushLock.readLock().lock();
+ try {
+ segmentsByUUID.put(action.getUuid(), action);
+ if (!queue.offer(action, 1, TimeUnit.MINUTES)) {
+ segmentsByUUID.remove(action.getUuid());
+ throw new IOException("Can't add segment to the queue");
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ } finally {
+ flushLock.readLock().unlock();
+ }
+ }
+
+ public void flush() throws IOException {
+ flushLock.writeLock().lock();
+ try {
+ synchronized (segmentsByUUID) {
+ long start = System.currentTimeMillis();
+ while (!segmentsByUUID.isEmpty()) {
+ segmentsByUUID.wait(100);
+ if (System.currentTimeMillis() - start > TimeUnit.MINUTES.toMillis(1)) {
+ log.error("Can't flush the queue in 1 minute. Queue: {}. Segment map: {}", queue, segmentsByUUID);
+ start = System.currentTimeMillis();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ } finally {
+ flushLock.writeLock().unlock();
+ }
+ }
+
+ public SegmentWriteAction read(UUID id) {
+ return segmentsByUUID.get(id);
+ }
+
+ public void close() throws IOException {
+ shutdown = true;
+ try {
+ executor.shutdown();
+ if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+ throw new IOException("The write wasn't able to shut down clearly");
+ }
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+ public boolean isEmpty() {
+ return segmentsByUUID.isEmpty();
+ }
+
+ boolean isBroken() {
+ return broken;
+ }
+
+ int getSize() {
+ return queue.size();
+ }
+
+ private void setBroken(boolean broken) {
+ synchronized (brokenMonitor) {
+ this.broken = broken;
+ brokenMonitor.notifyAll();
+ }
+ }
+
+ private void waitWhileBroken() {
+ if (!broken) {
+ return;
+ }
+ synchronized (brokenMonitor) {
+ while (broken && !shutdown) {
+ try {
+ brokenMonitor.wait(100);
+ } catch (InterruptedException e) {
+ log.warn("Interrupted", e);
+ }
+ }
+ }
+ }
+
+ private void waitUntilBroken() {
+ if (broken) {
+ return;
+ }
+ synchronized (brokenMonitor) {
+ while (!broken && !shutdown) {
+ try {
+ brokenMonitor.wait(100);
+ } catch (InterruptedException e) {
+ log.warn("Interrupted", e);
+ }
+ }
+ }
+ }
+
+ public interface SegmentConsumer {
+
+ void consume(AzureSegmentArchiveEntry indexEntry, byte[] data, int offset, int size) throws IOException;
+
+ }
+
+ public static class SegmentConsumeException extends Exception {
+
+ private final SegmentWriteAction segment;
+
+ public SegmentConsumeException(SegmentWriteAction segment, IOException cause) {
+ super(cause);
+ this.segment = segment;
+ }
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManagerTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManagerTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManagerTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManagerTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,74 @@
+/*
+ * 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.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveManager;
+import org.apache.jackrabbit.oak.segment.spi.monitor.FileStoreMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentArchiveWriter;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.UUID;
+
+import static com.google.common.collect.Lists.newArrayList;
+import static org.junit.Assert.assertEquals;
+
+public class AzureArchiveManagerTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ }
+
+ @Test
+ public void testRecovery() throws StorageException, URISyntaxException, IOException {
+ SegmentArchiveManager manager = new AzurePersistence(container.getDirectoryReference("oak")).createArchiveManager(false, new IOMonitorAdapter(), new FileStoreMonitorAdapter());
+ SegmentArchiveWriter writer = manager.create("data00000a.tar");
+
+ List<UUID> uuids = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ UUID u = UUID.randomUUID();
+ writer.writeSegment(u.getMostSignificantBits(), u.getLeastSignificantBits(), new byte[10], 0, 10, 0, 0, false);
+ uuids.add(u);
+ }
+
+ writer.flush();
+ writer.close();
+
+ container.getBlockBlobReference("oak/data00000a.tar/0005." + uuids.get(5).toString()).delete();
+
+ LinkedHashMap<UUID, byte[]> recovered = new LinkedHashMap<>();
+ manager.recoverEntries("data00000a.tar", recovered);
+ assertEquals(uuids.subList(0, 5), newArrayList(recovered.keySet()));
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureGCJournalTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureGCJournalTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureGCJournalTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureGCJournalTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,45 @@
+package org.apache.jackrabbit.oak.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentNodeStorePersistence;
+import org.apache.jackrabbit.oak.segment.file.GcJournalTest;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+public class AzureGCJournalTest extends GcJournalTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ }
+
+ @Override
+ protected SegmentNodeStorePersistence getPersistence() throws Exception {
+ return new AzurePersistence(container.getDirectoryReference("oak"));
+ }
+
+ @Test
+ @Ignore
+ @Override
+ public void testReadOak16GCLog() throws Exception {
+ super.testReadOak16GCLog();
+ }
+
+ @Test
+ @Ignore
+ @Override
+ public void testUpdateOak16GCLog() throws Exception {
+ super.testUpdateOak16GCLog();
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureJournalFileTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureJournalFileTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureJournalFileTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureJournalFileTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,74 @@
+/*
+ * 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.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.persistence.JournalFileReader;
+import org.apache.jackrabbit.oak.segment.spi.persistence.JournalFileWriter;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class AzureJournalFileTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ private AzureJournalFile journal;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ journal = new AzureJournalFile(container.getDirectoryReference("journal"), "journal.log", 10);
+ }
+
+ @Test
+ public void testSplitJournalFiles() throws IOException {
+ assertFalse(journal.exists());
+
+ JournalFileWriter writer = journal.openJournalWriter();
+ for (int i = 0; i < 100; i++) {
+ writer.writeLine("line " + i);
+ }
+
+ assertTrue(journal.exists());
+
+ writer = journal.openJournalWriter();
+ for (int i = 100; i < 200; i++) {
+ writer.writeLine("line " + i);
+ }
+
+ JournalFileReader reader = journal.openJournalReader();
+ for (int i = 199; i >= 0; i--) {
+ assertEquals("line " + i, reader.readLine());
+ }
+
+
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureManifestFileTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureManifestFileTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureManifestFileTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureManifestFileTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,44 @@
+package org.apache.jackrabbit.oak.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.persistence.ManifestFile;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class AzureManifestFileTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ }
+
+ @Test
+ public void testManifest() throws URISyntaxException, IOException {
+ ManifestFile manifestFile = new AzurePersistence(container.getDirectoryReference("oak")).getManifestFile();
+ assertFalse(manifestFile.exists());
+
+ Properties props = new Properties();
+ props.setProperty("xyz", "abc");
+ props.setProperty("version", "123");
+ manifestFile.save(props);
+
+ Properties loaded = manifestFile.load();
+ assertEquals(props, loaded);
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFileTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFileTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFileTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFileTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,62 @@
+/*
+ * 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.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.monitor.FileStoreMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.file.tar.TarFileTest;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+public class AzureTarFileTest extends TarFileTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ @Override
+ public void setUp() throws IOException {
+ try {
+ container = azurite.getContainer("oak-test");
+ archiveManager = new AzurePersistence(container.getDirectoryReference("oak")).createArchiveManager(true, new IOMonitorAdapter(), new FileStoreMonitorAdapter());
+ } catch (StorageException | InvalidKeyException | URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ protected long getWriteAndReadExpectedSize() {
+ return 45;
+ }
+
+ @Test
+ @Ignore
+ @Override
+ public void graphShouldBeTrimmedDownOnSweep() throws Exception {
+ super.graphShouldBeTrimmedDownOnSweep();
+ }
+}
\ No newline at end of file
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFilesTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFilesTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFilesTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarFilesTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,49 @@
+/*
+ * 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.segment.azure;
+
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.monitor.FileStoreMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
+import org.apache.jackrabbit.oak.segment.file.tar.TarFilesTest;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+public class AzureTarFilesTest extends TarFilesTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ container = azurite.getContainer("oak-test");
+ tarFiles = TarFiles.builder()
+ .withDirectory(folder.newFolder())
+ .withTarRecovery((id, data, recovery) -> {
+ // Intentionally left blank
+ })
+ .withIOMonitor(new IOMonitorAdapter())
+ .withFileStoreMonitor(new FileStoreMonitorAdapter())
+ .withMaxFileSize(MAX_FILE_SIZE)
+ .withPersistence(new AzurePersistence(container.getDirectoryReference("oak")))
+ .build();
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarWriterTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarWriterTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarWriterTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzureTarWriterTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,48 @@
+/*
+ * 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.segment.azure;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitorAdapter;
+import org.apache.jackrabbit.oak.segment.file.tar.TarWriterTest;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+public class AzureTarWriterTest extends TarWriterTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ @Override
+ public void setUp() throws IOException {
+ try {
+ monitor = new TestFileStoreMonitor();
+ container = azurite.getContainer("oak-test");
+ archiveManager = new AzurePersistence(container.getDirectoryReference("oak")).createArchiveManager(true, new IOMonitorAdapter(), monitor);
+ } catch (StorageException | InvalidKeyException | URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzuriteDockerRule.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzuriteDockerRule.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzuriteDockerRule.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/AzuriteDockerRule.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,73 @@
+/*
+ * 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.segment.azure;
+
+import com.arakelian.docker.junit.DockerRule;
+import com.arakelian.docker.junit.model.ImmutableDockerConfig;
+import com.microsoft.azure.storage.CloudStorageAccount;
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.spotify.docker.client.DefaultDockerClient;
+import org.junit.Assume;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+public class AzuriteDockerRule implements TestRule {
+
+ private final DockerRule wrappedRule;
+
+ public AzuriteDockerRule() {
+ wrappedRule = new DockerRule(ImmutableDockerConfig.builder()
+ .image("trekawek/azurite")
+ .name("oak-test-azurite")
+ .ports("10000")
+ .addStartedListener(container -> {
+ container.waitForPort("10000/tcp");
+ container.waitForLog("Azure Blob Storage Emulator listening on port 10000");
+ })
+ .addContainerConfigurer(builder -> builder.env("executable=blob"))
+ .alwaysRemoveContainer(true)
+ .build());
+
+ }
+
+ public CloudBlobContainer getContainer(String name) throws URISyntaxException, StorageException, InvalidKeyException {
+ int mappedPort = wrappedRule.getContainer().getPortBinding("10000/tcp").getPort();
+ CloudStorageAccount cloud = CloudStorageAccount.parse("DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:" + mappedPort + "/devstoreaccount1;");
+ CloudBlobContainer container = cloud.createCloudBlobClient().getContainerReference(name);
+ container.deleteIfExists();
+ container.create();
+ return container;
+ }
+
+ @Override
+ public Statement apply(Statement statement, Description description) {
+ try {
+ DefaultDockerClient client = DefaultDockerClient.fromEnv().connectTimeoutMillis(5000L).readTimeoutMillis(20000L).build();
+ client.ping();
+ client.close();
+ } catch (Exception e) {
+ Assume.assumeNoException(e);
+ }
+
+ return wrappedRule.apply(statement, description);
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/fixture/SegmentAzureFixture.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/fixture/SegmentAzureFixture.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/fixture/SegmentAzureFixture.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/fixture/SegmentAzureFixture.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,111 @@
+/*
+ * 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.segment.azure.fixture;
+
+import com.google.common.io.Files;
+import com.microsoft.azure.storage.CloudStorageAccount;
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.microsoft.azure.storage.blob.CloudBlobDirectory;
+import org.apache.jackrabbit.oak.fixture.NodeStoreFixture;
+import org.apache.jackrabbit.oak.segment.SegmentNodeStoreBuilders;
+import org.apache.jackrabbit.oak.segment.azure.AzurePersistence;
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.FileStoreBuilder;
+import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class SegmentAzureFixture extends NodeStoreFixture {
+
+ private static final String AZURE_CONNECTION_STRING = System.getProperty("oak.segment.azure.connection", "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;");
+
+ private static final String AZURE_CONTAINER = System.getProperty("oak.segment.azure.container", "oak");
+
+ private static final String AZURE_ROOT_PATH = System.getProperty("oak.segment.azure.rootPath", "/oak");
+
+ private Map<NodeStore, FileStore> fileStoreMap = new HashMap<>();
+
+ private Map<NodeStore, CloudBlobContainer> containerMap = new HashMap<>();
+
+ @Override
+ public NodeStore createNodeStore() {
+ AzurePersistence persistence;
+ CloudBlobContainer container;
+ try {
+ CloudStorageAccount cloud = CloudStorageAccount.parse(AZURE_CONNECTION_STRING);
+
+ int i = 1;
+ while (true) {
+ String containerName;
+ if (i == 1) {
+ containerName = AZURE_CONTAINER;
+ } else {
+ containerName = AZURE_CONTAINER + "_" + i;
+ }
+ container = cloud.createCloudBlobClient().getContainerReference(containerName);
+ if (!container.exists()) {
+ container.create();
+ break;
+ }
+ i++;
+ }
+ CloudBlobDirectory directory = container.getDirectoryReference(AZURE_ROOT_PATH);
+ persistence = new AzurePersistence(directory);
+ } catch (StorageException | URISyntaxException | InvalidKeyException e) {
+ throw new RuntimeException(e);
+ }
+
+ try {
+ FileStore fileStore = FileStoreBuilder.fileStoreBuilder(Files.createTempDir()).withCustomPersistence(persistence).build();
+ NodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build();
+ fileStoreMap.put(nodeStore, fileStore);
+ containerMap.put(nodeStore, container);
+ return nodeStore;
+ } catch (IOException | InvalidFileStoreVersionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void dispose(NodeStore nodeStore) {
+ FileStore fs = fileStoreMap.remove(nodeStore);
+ if (fs != null) {
+ fs.close();
+ }
+ try {
+ CloudBlobContainer container = containerMap.remove(nodeStore);
+ if (container != null) {
+ container.deleteIfExists();
+ }
+ } catch (StorageException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "SegmentAzure";
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureJournalReaderTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureJournalReaderTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureJournalReaderTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureJournalReaderTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,55 @@
+/*
+ * 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.segment.azure.journal;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudAppendBlob;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.azure.AzuriteDockerRule;
+import org.apache.jackrabbit.oak.segment.file.JournalReader;
+import org.apache.jackrabbit.oak.segment.file.JournalReaderTest;
+import org.apache.jackrabbit.oak.segment.azure.AzureJournalFile;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+public class AzureJournalReaderTest extends JournalReaderTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ }
+
+ protected JournalReader createJournalReader(String s) throws IOException {
+ try {
+ CloudAppendBlob blob = container.getAppendBlobReference("journal/journal.log.001");
+ blob.createOrReplace();
+ blob.appendText(s);
+ return new JournalReader(new AzureJournalFile(container.getDirectoryReference("journal"), "journal.log"));
+ } catch (StorageException | URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureTarRevisionsTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureTarRevisionsTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureTarRevisionsTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/AzureTarRevisionsTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,51 @@
+/*
+ * 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.segment.azure.journal;
+
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.spi.persistence.SegmentNodeStorePersistence;
+import org.apache.jackrabbit.oak.segment.azure.AzuriteDockerRule;
+import org.apache.jackrabbit.oak.segment.file.TarRevisionsTest;
+import org.apache.jackrabbit.oak.segment.azure.AzurePersistence;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+
+public class AzureTarRevisionsTest extends TarRevisionsTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws Exception {
+ container = azurite.getContainer("oak-test");
+ super.setup();
+ }
+
+ @Override
+ protected SegmentNodeStorePersistence getPersistence() throws IOException {
+ try {
+ return new AzurePersistence(container.getDirectoryReference("oak"));
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/ReverseFileReaderTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/ReverseFileReaderTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/ReverseFileReaderTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/journal/ReverseFileReaderTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,114 @@
+/*
+ * 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.segment.azure.journal;
+
+import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.blob.CloudAppendBlob;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import org.apache.jackrabbit.oak.segment.azure.AzuriteDockerRule;
+import org.apache.jackrabbit.oak.segment.azure.ReverseFileReader;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+public class ReverseFileReaderTest {
+
+ @ClassRule
+ public static AzuriteDockerRule azurite = new AzuriteDockerRule();
+
+ private CloudBlobContainer container;
+
+ @Before
+ public void setup() throws StorageException, InvalidKeyException, URISyntaxException {
+ container = azurite.getContainer("oak-test");
+ getBlob().createOrReplace();
+ }
+
+ private CloudAppendBlob getBlob() throws URISyntaxException, StorageException {
+ return container.getAppendBlobReference("test-blob");
+ }
+
+ @Test
+ public void testReverseReader() throws IOException, URISyntaxException, StorageException {
+ List<String> entries = createFile( 1024, 80);
+ ReverseFileReader reader = new ReverseFileReader(getBlob(), 256);
+ assertEquals(entries, reader);
+ }
+
+ @Test
+ public void testEmptyFile() throws IOException, URISyntaxException, StorageException {
+ List<String> entries = createFile( 0, 80);
+ ReverseFileReader reader = new ReverseFileReader(getBlob(), 256);
+ assertEquals(entries, reader);
+ }
+
+ @Test
+ public void test1ByteBlock() throws IOException, URISyntaxException, StorageException {
+ List<String> entries = createFile( 10, 16);
+ ReverseFileReader reader = new ReverseFileReader(getBlob(), 1);
+ assertEquals(entries, reader);
+ }
+
+
+ private List<String> createFile(int lines, int maxLineLength) throws IOException, URISyntaxException, StorageException {
+ Random random = new Random();
+ List<String> entries = new ArrayList<>();
+ CloudAppendBlob blob = getBlob();
+ for (int i = 0; i < lines; i++) {
+ int entrySize = random.nextInt(maxLineLength) + 1;
+ String entry = randomString(entrySize);
+ try {
+ blob.appendText(entry + '\n');
+ } catch (StorageException e) {
+ throw new IOException(e);
+ }
+ entries.add(entry);
+ }
+
+ entries.add("");
+ Collections.reverse(entries);
+ return entries;
+ }
+
+ private static void assertEquals(List<String> entries, ReverseFileReader reader) throws IOException {
+ int i = entries.size();
+ for (String e : entries) {
+ Assert.assertEquals("line " + (--i), e, reader.readLine());
+ }
+ Assert.assertNull(reader.readLine());
+ }
+
+ private static String randomString(int entrySize) {
+ Random r = new Random();
+
+ StringBuilder result = new StringBuilder();
+ for (int i = 0; i < entrySize; i++) {
+ result.append((char) ('a' + r.nextInt('z' - 'a')));
+ }
+
+ return result.toString();
+ }
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueueTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueueTest.java?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueueTest.java (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/src/test/java/org/apache/jackrabbit/oak/segment/azure/queue/SegmentWriteQueueTest.java Tue Mar 20 10:54:09 2018
@@ -0,0 +1,231 @@
+/*
+ * 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.segment.azure.queue;
+
+import org.apache.jackrabbit.oak.segment.azure.AzureSegmentArchiveEntry;
+import org.junit.After;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class SegmentWriteQueueTest {
+
+ private static final byte[] EMPTY_DATA = new byte[0];
+
+ private SegmentWriteQueue queue;
+
+ @After
+ public void shutdown() throws IOException {
+ if (queue != null) {
+ queue.close();
+ }
+ }
+
+ @Test
+ public void testQueue() throws IOException, InterruptedException {
+ Set<UUID> added = Collections.synchronizedSet(new HashSet<>());
+ Semaphore semaphore = new Semaphore(0);
+ queue = new SegmentWriteQueue((tarEntry, data, offset, size) -> {
+ try {
+ semaphore.acquire();
+ } catch (InterruptedException e) {
+ }
+ added.add(new UUID(tarEntry.getMsb(), tarEntry.getLsb()));
+ });
+
+ for (int i = 0; i < 10; i++) {
+ queue.addToQueue(tarEntry(i), EMPTY_DATA, 0, 0);
+ }
+
+ for (int i = 0; i < 10; i++) {
+ assertNotNull("Segments should be available for read", queue.read(uuid(i)));
+ }
+ assertFalse("Queue shouldn't be empty", queue.isEmpty());
+
+ semaphore.release(Integer.MAX_VALUE);
+ while (!queue.isEmpty()) {
+ Thread.sleep(10);
+ }
+
+ assertEquals("There should be 10 segments consumed",10, added.size());
+ for (int i = 0; i < 10; i++) {
+ assertTrue("Missing consumed segment", added.contains(uuid(i)));
+ }
+ }
+
+ @Test(timeout = 1000)
+ public void testFlush() throws IOException, InterruptedException {
+ Set<UUID> added = Collections.synchronizedSet(new HashSet<>());
+ Semaphore semaphore = new Semaphore(0);
+ queue = new SegmentWriteQueue((tarEntry, data, offset, size) -> {
+ try {
+ semaphore.acquire();
+ } catch (InterruptedException e) {
+ }
+ added.add(new UUID(tarEntry.getMsb(), tarEntry.getLsb()));
+ });
+
+ for (int i = 0; i < 3; i++) {
+ queue.addToQueue(tarEntry(i), EMPTY_DATA, 0, 0);
+ }
+
+ AtomicBoolean flushFinished = new AtomicBoolean(false);
+ Set<UUID> addedAfterFlush = new HashSet<>();
+ new Thread(() -> {
+ try {
+ queue.flush();
+ flushFinished.set(true);
+ addedAfterFlush.addAll(added);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }).start();
+
+ Thread.sleep(100);
+ assertFalse("Flush should be blocked", flushFinished.get());
+
+ AtomicBoolean addFinished = new AtomicBoolean(false);
+ new Thread(() -> {
+ try {
+ queue.addToQueue(tarEntry(10), EMPTY_DATA, 0, 0);
+ addFinished.set(true);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }).start();
+
+ Thread.sleep(100);
+ assertFalse("Adding segments should be blocked until the flush is finished", addFinished.get());
+
+ semaphore.release(Integer.MAX_VALUE);
+
+ while (!addFinished.get()) {
+ Thread.sleep(10);
+ }
+ assertTrue("Flush should be finished once the ", flushFinished.get());
+ assertTrue("Adding segments should be blocked until the flush is finished", addFinished.get());
+
+ for (int i = 0; i < 3; i++) {
+ assertTrue(addedAfterFlush.contains(uuid(i)));
+ }
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testClose() throws IOException, InterruptedException {
+ queue = new SegmentWriteQueue((tarEntry, data, offset, size) -> {});
+ queue.close();
+ queue.addToQueue(tarEntry(10), EMPTY_DATA, 0, 0);
+ }
+
+ @Test
+ public void testRecoveryMode() throws IOException, InterruptedException {
+ Set<UUID> added = Collections.synchronizedSet(new HashSet<>());
+ Semaphore semaphore = new Semaphore(0);
+ AtomicBoolean doBreak = new AtomicBoolean(true);
+ List<Long> writeAttempts = Collections.synchronizedList(new ArrayList<>());
+ queue = new SegmentWriteQueue((tarEntry, data, offset, size) -> {
+ writeAttempts.add(System.currentTimeMillis());
+ try {
+ semaphore.acquire();
+ } catch (InterruptedException e) {
+ }
+ if (doBreak.get()) {
+ throw new IOException();
+ }
+ added.add(new UUID(tarEntry.getMsb(), tarEntry.getLsb()));
+ });
+
+ for (int i = 0; i < 10; i++) {
+ queue.addToQueue(tarEntry(i), EMPTY_DATA, 0, 0);
+ }
+
+ semaphore.release(Integer.MAX_VALUE);
+ Thread.sleep(100);
+
+ assertTrue(queue.isBroken());
+ assertEquals(9, queue.getSize()); // the 10th segment is handled by the recovery thread
+
+ writeAttempts.clear();
+ while (writeAttempts.size() < 5) {
+ Thread.sleep(100);
+ }
+ long lastAttempt = writeAttempts.get(0);
+ for (int i = 1; i < 5; i++) {
+ long delay = writeAttempts.get(i) - lastAttempt;
+ assertTrue("The delay between attempts to persist segment should be larger than 1s. Actual: " + delay, delay >= 1000);
+ lastAttempt = writeAttempts.get(i);
+ }
+
+ AtomicBoolean addFinished = new AtomicBoolean(false);
+ new Thread(() -> {
+ try {
+ queue.addToQueue(tarEntry(10), EMPTY_DATA, 0, 0);
+ addFinished.set(true);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }).start();
+
+ Thread.sleep(100);
+ assertFalse("Adding segments should be blocked until the recovery mode is finished", addFinished.get());
+
+ doBreak.set(false);
+ while (queue.isBroken()) {
+ Thread.sleep(10);
+ }
+ assertFalse("Queue shouldn't be broken anymore", queue.isBroken());
+
+ while (added.size() < 11) {
+ Thread.sleep(10);
+ }
+ assertEquals("All segments should be consumed",11, added.size());
+ for (int i = 0; i < 11; i++) {
+ assertTrue("All segments should be consumed", added.contains(uuid(i)));
+ }
+
+ int i = writeAttempts.size() - 10;
+ lastAttempt = writeAttempts.get(i);
+ for (; i < writeAttempts.size(); i++) {
+ long delay = writeAttempts.get(i) - lastAttempt;
+ assertTrue("Segments should be persisted immediately", delay < 1000);
+ lastAttempt = writeAttempts.get(i);
+ }
+ }
+
+ private static AzureSegmentArchiveEntry tarEntry(long i) {
+ return new AzureSegmentArchiveEntry(0, i, 0, 0, 0, 0, false);
+ }
+
+ private static UUID uuid(long i) {
+ return new UUID(0, i);
+ }
+
+}
Added: jackrabbit/oak/trunk/oak-segment-azure/start-azurite.sh
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-azure/start-azurite.sh?rev=1827292&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-segment-azure/start-azurite.sh (added)
+++ jackrabbit/oak/trunk/oak-segment-azure/start-azurite.sh Tue Mar 20 10:54:09 2018
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+docker run -e executable=blob --rm -t -p 10000:10000 trekawek/azurite
Propchange: jackrabbit/oak/trunk/oak-segment-azure/start-azurite.sh
------------------------------------------------------------------------------
svn:executable = *
Modified: jackrabbit/oak/trunk/pom.xml
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/pom.xml?rev=1827292&r1=1827291&r2=1827292&view=diff
==============================================================================
--- jackrabbit/oak/trunk/pom.xml (original)
+++ jackrabbit/oak/trunk/pom.xml Tue Mar 20 10:54:09 2018
@@ -70,6 +70,7 @@
<module>oak-examples</module>
<module>oak-it</module>
<module>oak-segment-tar</module>
+ <module>oak-segment-azure</module>
<module>oak-benchmarks</module>
</modules>