This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2c11a5fbc04 Fail fast in artifact staging when storing an artifact
fails (#39367)
2c11a5fbc04 is described below
commit 2c11a5fbc043a1b82dc4f236f4fef34c812d11ef
Author: Elia Liu <[email protected]>
AuthorDate: Tue Sep 1 00:20:46 2026 +1000
Fail fast in artifact staging when storing an artifact fails (#39367)
* Propagate artifact storage failures to the staging client instead of
hanging
A StoreArtifact task that failed with an unchecked exception, such as
InvalidPathException on Windows, was never reported: the chunk producer
blocked forever and the reverse-retrieval stream was never terminated,
so clients hung in ArtifactStagingService.offer until their timeout.
Report every failure to the pending-bytes semaphore, make the producer
observe it promptly, and error the stream with the root cause.
---
.../artifact/ArtifactStagingService.java | 34 ++++++++---
.../artifact/ArtifactStagingServiceTest.java | 68 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 7 deletions(-)
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java
index 8b403f2f25f..f6c4b4d4f34 100644
---
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java
@@ -223,15 +223,19 @@ public class ArtifactStagingService
}
synchronized void aquire(int permits) throws Exception {
- while (usedPermits >= totalPermits) {
- if (exception != null) {
- throw exception;
- }
+ while (exception == null && usedPermits >= totalPermits) {
this.wait();
}
+ checkException();
usedPermits += permits;
}
+ synchronized void checkException() throws Exception {
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
synchronized void release(int permits) {
usedPermits -= permits;
this.notifyAll();
@@ -282,13 +286,21 @@ public class ArtifactStagingService
.setTypeUrn(dest.getTypeUrn())
.setTypePayload(dest.getTypePayload())
.build();
- } catch (IOException | InterruptedException exn) {
+ } catch (Exception exn) {
// As this thread will no longer be draining the queue, we don't want
to get stuck writing
- // to it.
+ // to it. This must happen for unchecked exceptions as well:
getDestination can throw e.g.
+ // InvalidPathException, and leaving the error unset would block the
producer forever.
totalPendingBytes.setException(exn);
+ // Free a producer already blocked in put; its next aquire observes
the exception.
+ bytesQueue.clear();
LOG.error("Exception staging artifacts", exn);
+ if (exn instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
if (exn instanceof IOException) {
throw (IOException) exn;
+ } else if (exn instanceof RuntimeException) {
+ throw (RuntimeException) exn;
} else {
throw new RuntimeException(exn);
}
@@ -421,8 +433,16 @@ public class ArtifactStagingService
}
}
} catch (Exception exn) {
- LOG.error("Error submitting.", exn);
+ if (exn instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
onError(exn);
+ // Terminate the stream towards the client, which would
otherwise wait forever.
+ responseObserver.onError(
+ Status.INTERNAL
+ .withDescription("Error staging artifacts: " + exn)
+ .withCause(exn)
+ .asException());
}
break;
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java
index 5610e4f5bb5..abdce59458d 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingServiceTest.java
@@ -18,7 +18,10 @@
package org.apache.beam.runners.fnexecution.artifact;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import java.nio.file.InvalidPathException;
import java.nio.file.Path;
import java.util.Iterator;
import java.util.List;
@@ -135,6 +138,22 @@ public class ArtifactStagingServiceTest {
}
}
+ /** Streams each artifact one byte per chunk, to exercise the service's
chunk buffering. */
+ private static class OneBytePerChunkArtifactRetrievalService
+ extends FakeArtifactRetrievalService {
+ @Override
+ public void getArtifact(
+ ArtifactApi.GetArtifactRequest request,
+ StreamObserver<ArtifactApi.GetArtifactResponse> responseObserver) {
+ ByteString data = request.getArtifact().getTypePayload();
+ for (int i = 0; i < data.size(); i++) {
+ responseObserver.onNext(
+
ArtifactApi.GetArtifactResponse.newBuilder().setData(data.substring(i, i +
1)).build());
+ }
+ responseObserver.onCompleted();
+ }
+ }
+
private String getArtifact(RunnerApi.ArtifactInformation artifact) {
ByteString all = ByteString.EMPTY;
Iterator<ArtifactApi.GetArtifactResponse> response =
@@ -166,6 +185,55 @@ public class ArtifactStagingServiceTest {
checkArtifacts(contentsList, staged.get("env2"));
}
+ @SuppressWarnings("InlineMeInliner") // inline `Strings.repeat()` - Java 11+
API only
+ @Test(timeout = 60_000)
+ public void testDestinationFailureFailsOfferInsteadOfHanging() throws
Exception {
+ // Resolving the destination of a staged artifact can throw an unchecked
exception, e.g.
+ // InvalidPathException on Windows where the generated filename may
contain characters that
+ // are illegal in paths (https://github.com/apache/beam/issues/39364).
This must fail the
+ // offering client instead of stalling the transfer forever.
+ ArtifactStagingService failingStagingService =
+ new ArtifactStagingService(
+ new ArtifactStagingService.ArtifactDestinationProvider() {
+ @Override
+ public ArtifactStagingService.ArtifactDestination getDestination(
+ String stagingToken, String name) {
+ throw new InvalidPathException(name, "Illegal char simulated");
+ }
+
+ @Override
+ public void removeStagedArtifacts(String stagingToken) {}
+ });
+ grpcCleanup.register(
+ InProcessServerBuilder.forName("failing-server")
+ .directExecutor()
+ .addService(failingStagingService)
+ .build()
+ .start());
+ ManagedChannel failingChannel =
+
grpcCleanup.register(InProcessChannelBuilder.forName("failing-server").build());
+ ArtifactStagingServiceGrpc.ArtifactStagingServiceStub failingStub =
+ ArtifactStagingServiceGrpc.newStub(failingChannel);
+
+ // More chunks than the service buffers per artifact, so staging cannot
run to completion
+ // before the destination failure is observed.
+ String contents = Strings.repeat("x", 300);
+ failingStagingService.registerJob(
+ "failingToken",
+ ImmutableMap.of(
+ "env1",
ImmutableList.of(FakeArtifactRetrievalService.resolvedArtifact(contents))));
+
+ ExecutionException exn =
+ assertThrows(
+ ExecutionException.class,
+ () ->
+ ArtifactStagingService.offer(
+ new OneBytePerChunkArtifactRetrievalService(),
failingStub, "failingToken"));
+ assertTrue(
+ "Expected the destination failure, got: " + exn.getCause(),
+ exn.getCause().getMessage().contains("Illegal char simulated"));
+ }
+
private void checkArtifacts(
List<String> expectedContents, List<RunnerApi.ArtifactInformation>
staged) {
assertEquals(
