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 js...@apache.org on 2025/08/14 14:31:09 UTC
(jackrabbit-oak) branch jsedding/OAK-11860-segment-azure-parallel-io created (now 565cceaf57)
This is an automated email from the ASF dual-hosted git repository.
jsedding pushed a change to branch jsedding/OAK-11860-segment-azure-parallel-io
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
at 565cceaf57 OAK-11860 - oak-segment-azure should perform I/O in parallel
This branch includes the following new commits:
new 565cceaf57 OAK-11860 - oak-segment-azure should perform I/O in parallel
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(jackrabbit-oak) 01/01: OAK-11860 - oak-segment-azure should perform I/O in parallel
Posted by js...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jsedding pushed a commit to branch jsedding/OAK-11860-segment-azure-parallel-io
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit 565cceaf57d19d45ecbd999e861c8ba91a9e82bd
Author: Julian Sedding <js...@apache.org>
AuthorDate: Thu Aug 14 16:20:50 2025 +0200
OAK-11860 - oak-segment-azure should perform I/O in parallel
---
.../oak/segment/azure/AzureArchiveManager.java | 59 ++++++++++------------
.../oak/segment/azure/AzureUtilities.java | 18 ++++---
2 files changed, 38 insertions(+), 39 deletions(-)
diff --git a/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManager.java b/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManager.java
index a2d6c4e908..6611a3a5d2 100644
--- a/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManager.java
+++ b/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureArchiveManager.java
@@ -38,21 +38,21 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Objects;
import java.util.UUID;
import java.util.Set;
import java.time.Duration;
+import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static com.azure.storage.blob.models.BlobType.BLOCK_BLOB;
import static org.apache.jackrabbit.oak.commons.conditions.Validate.checkArgument;
-import static org.apache.jackrabbit.oak.segment.azure.AzureUtilities.getName;
public class AzureArchiveManager implements SegmentArchiveManager {
@@ -206,28 +206,21 @@ public class AzureArchiveManager implements SegmentArchiveManager {
@Override
public void recoverEntries(String archiveName, LinkedHashMap<UUID, byte[]> entries) throws IOException {
Pattern pattern = Pattern.compile(RemoteUtilities.SEGMENT_FILE_NAME_PATTERN);
- List<RecoveredEntry> entryList = new ArrayList<>();
-
- for (BlobItem b : getBlobs(archiveName)) {
- String name = getName(b);
- Matcher m = pattern.matcher(name);
- if (!m.matches()) {
- continue;
- }
- int position = Integer.parseInt(m.group(1), 16);
- UUID uuid = UUID.fromString(m.group(2));
- long length = b.getProperties().getContentLength();
- if (length > 0) {
- byte[] data;
- try {
- data = readBlobContainerClient.getBlobClient(b.getName()).downloadContent().toBytes();
- } catch (BlobStorageException e) {
- throw new IOException(e);
- }
- entryList.add(new RecoveredEntry(position, uuid, data, name));
- }
- }
- Collections.sort(entryList);
+ List<RecoveredEntry> entryList = getBlobs(archiveName)
+ .filter(b -> b.getProperties().getContentLength() > 0)
+ .map(BlobItem::getName)
+ .map(blobPath -> {
+ String blobName = AzureUtilities.getFileName(blobPath);
+ Matcher m = pattern.matcher(blobName);
+ return !m.matches() ? null : new RecoveredEntry(
+ Integer.parseInt(m.group(1), 16),
+ UUID.fromString(m.group(2)),
+ readBlobContainerClient.getBlobClient(blobPath).downloadContent().toBytes(),
+ blobName);
+ })
+ .filter(Objects::nonNull)
+ .sorted()
+ .collect(Collectors.toUnmodifiableList());
int i = 0;
for (RecoveredEntry e : entryList) {
@@ -243,12 +236,14 @@ public class AzureArchiveManager implements SegmentArchiveManager {
private void delete(String archiveName, Set<UUID> recoveredEntries) throws IOException {
getBlobs(archiveName + "/")
- .forEach(blobItem -> {
- if (!recoveredEntries.contains(RemoteUtilities.getSegmentUUID(getName(blobItem)))) {
+ .filter(Predicate.not(BlobItem::isPrefix))
+ .map(BlobItem::getName)
+ .forEach(blobName -> {
+ if (!recoveredEntries.contains(RemoteUtilities.getSegmentUUID(AzureUtilities.getFileName(blobName)))) {
try {
- writeBlobContainerClient.getBlobClient(blobItem.getName()).delete();
+ writeBlobContainerClient.getBlobClient(blobName).delete();
} catch (BlobStorageException e) {
- log.error("Can't delete segment {}", blobItem.getName(), e);
+ log.error("Can't delete segment {}", blobName, e);
}
}
});
@@ -269,12 +264,12 @@ public class AzureArchiveManager implements SegmentArchiveManager {
return String.format("%s/%s", rootPrefix, archiveName);
}
- private List<BlobItem> getBlobs(String archiveName) throws IOException {
+ private Stream<BlobItem> getBlobs(String archiveName) throws IOException {
String archivePath = getDirectory(archiveName);
ListBlobsOptions listBlobsOptions = new ListBlobsOptions();
listBlobsOptions.setPrefix(archivePath);
- return AzureUtilities.getBlobs(readBlobContainerClient, listBlobsOptions);
+ return AzureUtilities.getParallelBlobStream(readBlobContainerClient, listBlobsOptions);
}
private void renameBlob(BlobItem blob, String newParent) throws IOException {
@@ -291,7 +286,7 @@ public class AzureArchiveManager implements SegmentArchiveManager {
BlockBlobClient sourceBlobClient = readBlobContainerClient.getBlobClient(blob.getName()).getBlockBlobClient();
- String destinationBlob = String.format("%s/%s", newParent, AzureUtilities.getName(blob));
+ String destinationBlob = String.format("%s/%s", newParent, AzureUtilities.getFileName(blob.getName()));
BlockBlobClient destinationBlobClient = writeBlobContainerClient.getBlobClient(destinationBlob).getBlockBlobClient();
PollResponse<BlobCopyInfo> response = destinationBlobClient.beginCopy(sourceBlobClient.getBlobUrl(), Duration.ofMillis(100)).waitForCompletion();
diff --git a/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java b/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java
index c3a7349a0e..a948c3946b 100644
--- a/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java
+++ b/oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/AzureUtilities.java
@@ -21,7 +21,6 @@ import com.azure.storage.blob.models.BlobItem;
import com.azure.storage.blob.models.BlobListDetails;
import com.azure.storage.blob.models.BlobStorageException;
import com.azure.storage.blob.models.ListBlobsOptions;
-import com.azure.storage.blob.specialized.AppendBlobClient;
import com.azure.storage.blob.specialized.BlockBlobClient;
import org.apache.jackrabbit.oak.commons.Buffer;
import org.apache.jackrabbit.oak.segment.spi.RepositoryNotReachableException;
@@ -35,6 +34,7 @@ import java.io.OutputStream;
import java.nio.file.Paths;
import java.util.List;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
public final class AzureUtilities {
@@ -50,15 +50,13 @@ public final class AzureUtilities {
}
public static String getName(BlobItem blob) {
- return Paths.get(blob.getName()).getFileName().toString();
+ return getFileName(blob.getName());
}
- public static String getName(AppendBlobClient blob) {
- return Paths.get(blob.getBlobName()).getFileName().toString();
+ public static @NotNull String getFileName(String blobName) {
+ return Paths.get(blobName).getFileName().toString();
}
-
-
public static List<BlobItem> getBlobs(BlobContainerClient blobContainerClient, ListBlobsOptions listOptions) {
if (listOptions != null) {
listOptions.setDetails(new BlobListDetails().setRetrieveMetadata(true));
@@ -66,6 +64,12 @@ public final class AzureUtilities {
return blobContainerClient.listBlobs(listOptions, null).stream().collect(Collectors.toList());
}
+ public static Stream<BlobItem> getParallelBlobStream(BlobContainerClient blobContainerClient, ListBlobsOptions listOptions) {
+ // collecting into a list, as done by #getBlobs(), is necessary in order to allow creation
+ // of a parallel stream (without a more complex implementation)
+ return getBlobs(blobContainerClient, listOptions).stream().unordered().parallel();
+ }
+
public static void readBufferFully(BlockBlobClient blob, Buffer buffer) throws IOException {
try {
blob.downloadStream(new ByteBufferOutputStream(buffer));
@@ -80,7 +84,7 @@ public final class AzureUtilities {
}
public static void deleteAllEntries(BlobContainerClient blobContainerClient, ListBlobsOptions listBlobsOptions) {
- getBlobs(blobContainerClient, listBlobsOptions).forEach(b -> {
+ getParallelBlobStream(blobContainerClient, listBlobsOptions).forEach(b -> {
try {
blobContainerClient.getBlobClient(b.getName()).deleteIfExists();
} catch (BlobStorageException e) {