You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/10/21 16:18:21 UTC
[L] Change in asterixdb[master]: [ASTERIXDB-3964][STO] Report failed GCS downloads
From Hussain Altowileb <hu...@gmail.com>:
Hussain Altowileb has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/22139?usp=email )
Change subject: [ASTERIXDB-3964][STO] Report failed GCS downloads
......................................................................
[ASTERIXDB-3964][STO] Report failed GCS downloads
GCSParallelDownloader.downloadFiles never checked the results of its
download jobs, so a failed download returned success; a missing object
left an empty local file behind a successful call.
Each attempt now checks every result and reports the files whose status
is not SUCCESS, with the result's exception as the cause, the same way
the S3 downloaders do: downloadFiles fails naming the failed files, and
downloadFilesWithRetry repeats only those. A GCS client error (4xx other
than 408 and 429), such as a missing object, is not retried, as for S3.
The transfer manager also strips the configured prefix from a blob name
as a regular expression, so a prefix with regex metacharacters sent the
file to another local path while the download reported success. The
prefix is now passed quoted.
Added test cases covering a failed download being reported, a missing
object not being retried, and a prefix with regex metacharacters.
Ext-ref: MB-74731
Generated-by: Claude Opus 5.5 <no...@anthropic.com>
Change-Id: I3ce9a61528d119698f132204cf2e8bdb848baac3
---
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/AbstractParallelDownloader.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
A asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloaderTest.java
3 files changed, 240 insertions(+), 16 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/39/22139/1
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/AbstractParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/AbstractParallelDownloader.java
index 740e58a..f32e3d5 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/AbstractParallelDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/AbstractParallelDownloader.java
@@ -37,6 +37,8 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import com.google.cloud.BaseServiceException;
+
import software.amazon.awssdk.core.exception.SdkServiceException;
public abstract class AbstractParallelDownloader implements IParallelDownloader {
@@ -161,13 +163,15 @@
/**
* Whether a failure is worth repeating: one {@link CloudRetryableRequestUtil#isRetryableFailure} accepts, unless
- * it is a client error the SDK itself would not retry (a 4xx status other than 408 and 429): a missing object or a
+ * it is a client error the SDK itself would not retry (an AWS or GCS 4xx status other than 408 and 429): a missing object or a
* refused credential does not change by waiting.
*/
private static boolean isRetryable(Throwable failure) {
Set<Throwable> seen = new HashSet<>();
for (Throwable cause = failure; cause != null && seen.add(cause); cause = cause.getCause()) {
- if (cause instanceof SdkServiceException && isClientError(((SdkServiceException) cause).statusCode())) {
+ if (cause instanceof SdkServiceException && isClientError(((SdkServiceException) cause).statusCode())
+ || cause instanceof BaseServiceException
+ && isClientError(((BaseServiceException) cause).getCode())) {
return false;
}
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
index 1df3da4..36f7dcb 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
@@ -29,6 +29,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.regex.Pattern;
import org.apache.asterix.cloud.clients.AbstractParallelDownloader;
import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
@@ -79,33 +80,54 @@
@Override
public void downloadFiles(Collection<FileReference> toDownload) throws HyracksDataException {
+ downloadOnce(toDownload, this::downloadFilesAndWait);
+ }
+
+ @Override
+ public void downloadFilesWithRetry(Collection<FileReference> toDownload) throws HyracksDataException {
+ retryFailedFiles(toDownload, this::downloadFilesAndWait);
+ }
+
+ private Map<FileReference, Throwable> downloadFilesAndWait(Collection<FileReference> toDownload)
+ throws IOException {
ParallelDownloadConfig.Builder downConfig =
- ParallelDownloadConfig.newBuilder().setBucketName(bucket).setStripPrefix(this.config.getPrefix());
+ ParallelDownloadConfig.newBuilder().setBucketName(bucket).setStripPrefix(stripPrefix());
Map<Path, List<BlobInfo>> pathListMap = new HashMap<>();
- try {
- for (FileReference fileReference : toDownload) {
- profiler.objectGet();
- FileUtils.createParentDirectories(fileReference.getFile());
- addToMap(pathListMap, fileReference.getDeviceHandle().getMount().toPath(), BlobInfo
- .newBuilder(BlobId.of(bucket, config.getPrefix() + fileReference.getRelativePath())).build());
- }
- } catch (IOException e) {
- throw HyracksDataException.create(e);
+ // per download directory: the same relative path on two IO devices names the same blob
+ Map<Path, Map<String, FileReference>> filesByBlobName = new HashMap<>();
+ for (FileReference fileReference : toDownload) {
+ profiler.objectGet();
+ FileUtils.createParentDirectories(fileReference.getFile());
+ Path mount = fileReference.getDeviceHandle().getMount().toPath();
+ String blobName = config.getPrefix() + fileReference.getRelativePath();
+ addToMap(pathListMap, mount, BlobInfo.newBuilder(BlobId.of(bucket, blobName)).build());
+ filesByBlobName.computeIfAbsent(mount, k -> new HashMap<>()).put(blobName, fileReference);
}
- List<DownloadJob> downloadJobs = new ArrayList<>(pathListMap.size());
+ Map<Path, DownloadJob> downloadJobs = new HashMap<>();
for (Map.Entry<Path, List<BlobInfo>> entry : pathListMap.entrySet()) {
- downloadJobs.add(transferManager.downloadBlobs(entry.getValue(),
+ downloadJobs.put(entry.getKey(), transferManager.downloadBlobs(entry.getValue(),
downConfig.setDownloadDirectory(entry.getKey()).build()));
}
- downloadJobs.forEach(DownloadJob::getDownloadResults);
+ Map<FileReference, Throwable> failed = new HashMap<>();
+ for (Map.Entry<Path, DownloadJob> job : downloadJobs.entrySet()) {
+ for (DownloadResult result : job.getValue().getDownloadResults()) {
+ if (result.getStatus() != TransferStatus.SUCCESS) {
+ String blobName = result.getInput().getName();
+ Exception cause = result.getException();
+ failed.put(filesByBlobName.get(job.getKey()).get(blobName), cause != null ? cause
+ : new IOException("download of " + blobName + " ended with status " + result.getStatus()));
+ }
+ }
+ }
+ return failed;
}
@Override
public Set<FileReference> downloadDirectories(Collection<FileReference> toDownload) {
Set<FileReference> failedFiles = new HashSet<>();
ParallelDownloadConfig.Builder config =
- ParallelDownloadConfig.newBuilder().setBucketName(bucket).setStripPrefix(this.config.getPrefix());
+ ParallelDownloadConfig.newBuilder().setBucketName(bucket).setStripPrefix(stripPrefix());
Map<Path, List<BlobInfo>> pathListMap = new HashMap<>();
for (FileReference fileReference : toDownload) {
@@ -134,6 +156,14 @@
return failedFiles;
}
+ /**
+ * The transfer manager removes the strip prefix from a blob name as a regular expression, while the prefix is a
+ * path, whose regex metacharacters must match themselves.
+ */
+ private String stripPrefix() {
+ return Pattern.quote(config.getPrefix());
+ }
+
@Override
protected String getPrefix() {
return config.getPrefix();
diff --git a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloaderTest.java b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloaderTest.java
new file mode 100644
index 0000000..7a01217
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloaderTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.asterix.cloud.clients.google.gcs;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.asterix.cloud.clients.profiler.NoOpRequestProfilerLimiter;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IODeviceHandle;
+import org.apache.hyracks.util.StorageUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class GCSParallelDownloaderTest {
+
+ private static final String KEY = "storage/partition_0/Default/dv/idx/0/idx/f0.dat";
+
+ private FakeGcs server;
+ private List<FileReference> files;
+
+ @Before
+ public void setUp() throws IOException {
+ server = new FakeGcs();
+ File dir = Files.createTempDirectory("gcs-download").toFile();
+ files = List.of(new FileReference(new IODeviceHandle(dir, "."), KEY));
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ server.close();
+ }
+
+ /**
+ * A download that fails is reported, not returned as a success: the transfer manager's results used to be
+ * ignored, so a missing object left an empty local file behind a successful call.
+ */
+ @Test(timeout = 120_000)
+ public void failedDownloadIsReported() throws Exception {
+ try (GCSParallelDownloader downloader = downloader("")) {
+ downloader.downloadFiles(files);
+ fail("a missing object was reported as downloaded");
+ } catch (HyracksDataException e) {
+ assertTrue(e.getMessage(), e.getMessage().contains(KEY));
+ }
+ }
+
+ /**
+ * A missing object is a client error: the retrying download fails at once instead of spending the budget.
+ */
+ @Test(timeout = 120_000)
+ public void missingObjectIsNotRetried() throws Exception {
+ long start = System.nanoTime();
+ try (GCSParallelDownloader downloader = downloader("")) {
+ downloader.downloadFilesWithRetry(files);
+ fail("a missing object was reported as downloaded");
+ } catch (HyracksDataException e) {
+ long ms = (System.nanoTime() - start) / 1_000_000;
+ assertTrue("requests " + server.requests.get(), server.requests.get() <= 3);
+ assertTrue("took " + ms + " ms", ms < 10_000);
+ }
+ }
+
+ /**
+ * The transfer manager strips the prefix as a regular expression; a prefix is a path, so regex metacharacters in
+ * it used to send the file under a directory named after the prefix, with the download reported as a success.
+ */
+ @Test(timeout = 120_000)
+ public void prefixIsAPathNotAPattern() throws Exception {
+ String prefix = "p+q/";
+ byte[] content = KEY.getBytes(StandardCharsets.UTF_8);
+ server.objects.put(prefix + KEY, content);
+ try (GCSParallelDownloader downloader = downloader(prefix)) {
+ downloader.downloadFiles(files);
+ }
+ assertArrayEquals(content, Files.readAllBytes(files.get(0).getFile().toPath()));
+ }
+
+ private GCSParallelDownloader downloader(String prefix) throws HyracksDataException {
+ GCSClientConfig config = new GCSClientConfig("", "http://127.0.0.1:" + server.port(), true, 0,
+ StorageUtil.getIntSizeInBytes(8, StorageUtil.StorageUnit.MEGABYTE), prefix);
+ return new GCSParallelDownloader("bucket", null, config, NoOpRequestProfilerLimiter.INSTANCE);
+ }
+
+ /**
+ * Serves the JSON API's object metadata and media GETs for the objects it holds, and 404 Not Found for any other.
+ */
+ private static final class FakeGcs implements AutoCloseable {
+ private final ServerSocket socket = new ServerSocket(0, 100);
+ private final AtomicInteger requests = new AtomicInteger();
+ private final Map<String, byte[]> objects = new ConcurrentHashMap<>();
+
+ FakeGcs() throws IOException {
+ Thread acceptor = new Thread(this::serve, "fake-gcs");
+ acceptor.setDaemon(true);
+ acceptor.start();
+ }
+
+ int port() {
+ return socket.getLocalPort();
+ }
+
+ private void serve() {
+ while (!socket.isClosed()) {
+ try (Socket connection = socket.accept()) {
+ BufferedReader in = new BufferedReader(
+ new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
+ String requestLine = in.readLine();
+ String header;
+ do {
+ header = in.readLine();
+ } while (header != null && !header.isEmpty());
+ requests.incrementAndGet();
+ respond(requestLine, connection.getOutputStream());
+ } catch (IOException e) {
+ if (socket.isClosed()) {
+ return;
+ }
+ }
+ }
+ }
+
+ private void respond(String requestLine, OutputStream out) throws IOException {
+ String target = requestLine == null ? "" : requestLine.split(" ")[1];
+ int query = target.indexOf('?');
+ String path = query < 0 ? target : target.substring(0, query);
+ int name = path.indexOf("/o/");
+ byte[] object =
+ name < 0 ? null : objects.get(URLDecoder.decode(path.substring(name + 3), StandardCharsets.UTF_8));
+ if (object == null) {
+ write(out, "404 Not Found", "application/json",
+ "{\"error\":{\"code\":404,\"message\":\"No such object\"}}".getBytes(StandardCharsets.UTF_8));
+ } else if (query >= 0 && target.substring(query).contains("alt=media")) {
+ write(out, "200 OK", "application/octet-stream", object);
+ } else {
+ write(out, "200 OK", "application/json",
+ ("{\"kind\":\"storage#object\",\"bucket\":\"bucket\",\"size\":\"" + object.length
+ + "\",\"generation\":\"1\",\"metageneration\":\"1\"}")
+ .getBytes(StandardCharsets.UTF_8));
+ }
+ out.flush();
+ }
+
+ private static void write(OutputStream out, String status, String contentType, byte[] body) throws IOException {
+ out.write(("HTTP/1.1 " + status + "\r\nContent-Length: " + body.length + "\r\nContent-Type: " + contentType
+ + "\r\nx-goog-generation: 1\r\nx-goog-metageneration: 1\r\nConnection: close\r\n\r\n")
+ .getBytes(StandardCharsets.UTF_8));
+ out.write(body);
+ }
+
+ @Override
+ public void close() throws IOException {
+ socket.close();
+ }
+ }
+}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/22139?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I3ce9a61528d119698f132204cf2e8bdb848baac3
Gerrit-Change-Number: 22139
Gerrit-PatchSet: 1
Gerrit-Owner: Hussain Altowileb <hu...@gmail.com>