[
https://issues.apache.org/jira/browse/BEAM-4071?focusedWorklogId=95277&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-95277
]
ASF GitHub Bot logged work on BEAM-4071:
----------------------------------------
Author: ASF GitHub Bot
Created on: 25/Apr/18 22:09
Start Date: 25/Apr/18 22:09
Worklog Time Spent: 10m
Work Description: jkff closed pull request #5150: [BEAM-4071] Add
Portable Runner Job API shim
URL: https://github.com/apache/beam/pull/5150
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ArtifactServiceStager.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ArtifactServiceStager.java
index 7319b8f30d1..63cc50810c3 100644
---
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ArtifactServiceStager.java
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ArtifactServiceStager.java
@@ -47,6 +47,7 @@
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.ArtifactChunk;
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.ArtifactMetadata;
import
org.apache.beam.model.jobmanagement.v1.ArtifactApi.CommitManifestRequest;
+import
org.apache.beam.model.jobmanagement.v1.ArtifactApi.CommitManifestResponse;
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.Manifest;
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.PutArtifactRequest;
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.PutArtifactResponse;
@@ -87,26 +88,33 @@ private ArtifactServiceStager(Channel channel, int
bufferSize) {
this.bufferSize = bufferSize;
}
- public void stage(Iterable<File> files) throws IOException,
InterruptedException {
- final Map<File, CompletionStage<ArtifactMetadata>> futures = new
HashMap<>();
- for (File file : files) {
+ /**
+ * Stages the given artifact files to the staging service.
+ *
+ * @return The artifact staging token returned by the service
+ */
+ public String stage(Iterable<StagedFile> files) throws IOException,
InterruptedException {
+ final Map<StagedFile, CompletionStage<ArtifactMetadata>> futures = new
HashMap<>();
+ for (StagedFile file : files) {
futures.put(file, MoreFutures.supplyAsync(new StagingCallable(file),
executorService));
}
CompletionStage<StagingResult> stagingResult =
MoreFutures.allAsList(futures.values())
.thenApply(ignored -> new
ExtractStagingResultsCallable(futures).call());
- stageManifest(stagingResult);
+ return stageManifest(stagingResult);
}
- private void stageManifest(CompletionStage<StagingResult> stagingFuture)
+ private String stageManifest(CompletionStage<StagingResult> stagingFuture)
throws InterruptedException {
try {
StagingResult stagingResult = MoreFutures.get(stagingFuture);
if (stagingResult.isSuccess()) {
Manifest manifest =
Manifest.newBuilder().addAllArtifact(stagingResult.getMetadata()).build();
- blockingStub.commitManifest(
- CommitManifestRequest.newBuilder().setManifest(manifest).build());
+ CommitManifestResponse response =
+ blockingStub.commitManifest(
+
CommitManifestRequest.newBuilder().setManifest(manifest).build());
+ return response.getStagingToken();
} else {
RuntimeException failure =
new RuntimeException(
@@ -124,9 +132,9 @@ private void stageManifest(CompletionStage<StagingResult>
stagingFuture)
}
private class StagingCallable implements ThrowingSupplier<ArtifactMetadata> {
- private final File file;
+ private final StagedFile file;
- private StagingCallable(File file) {
+ private StagingCallable(StagedFile file) {
this.file = file;
}
@@ -135,11 +143,12 @@ public ArtifactMetadata get() throws Exception {
// TODO: Add Retries
PutArtifactResponseObserver responseObserver = new
PutArtifactResponseObserver();
StreamObserver<PutArtifactRequest> requestObserver =
stub.putArtifact(responseObserver);
- ArtifactMetadata metadata =
ArtifactMetadata.newBuilder().setName(file.getName()).build();
+ ArtifactMetadata metadata =
+ ArtifactMetadata.newBuilder().setName(file.getStagingName()).build();
requestObserver.onNext(PutArtifactRequest.newBuilder().setMetadata(metadata).build());
MessageDigest md5Digest = MessageDigest.getInstance("MD5");
- FileChannel channel = new FileInputStream(file).getChannel();
+ FileChannel channel = new FileInputStream(file.getFile()).getChannel();
ByteBuffer readBuffer = ByteBuffer.allocate(bufferSize);
while (!responseObserver.isTerminal() && channel.position() <
channel.size()) {
readBuffer.clear();
@@ -193,18 +202,19 @@ public void awaitTermination() throws
InterruptedException {
}
private static class ExtractStagingResultsCallable implements
Callable<StagingResult> {
- private final Map<File, CompletionStage<ArtifactMetadata>> futures;
+ private final Map<StagedFile, CompletionStage<ArtifactMetadata>> futures;
private ExtractStagingResultsCallable(
- Map<File, CompletionStage<ArtifactMetadata>> futures) {
+ Map<StagedFile, CompletionStage<ArtifactMetadata>> futures) {
this.futures = futures;
}
@Override
public StagingResult call() {
Set<ArtifactMetadata> metadata = new HashSet<>();
- Map<File, Throwable> failures = new HashMap<>();
- for (Entry<File, CompletionStage<ArtifactMetadata>> stagedFileResult :
futures.entrySet()) {
+ Map<StagedFile, Throwable> failures = new HashMap<>();
+ for (Entry<StagedFile, CompletionStage<ArtifactMetadata>>
stagedFileResult :
+ futures.entrySet()) {
try {
metadata.add(MoreFutures.get(stagedFileResult.getValue()));
} catch (ExecutionException ee) {
@@ -222,13 +232,26 @@ public StagingResult call() {
}
}
+ /** A file along with a staging name. */
+ @AutoValue
+ public abstract static class StagedFile {
+ public static StagedFile of(File file, String stagingName) {
+ return new AutoValue_ArtifactServiceStager_StagedFile(file, stagingName);
+ }
+
+ /** The file to stage. */
+ public abstract File getFile();
+ /** Staging handle to this file. */
+ public abstract String getStagingName();
+ }
+
@AutoValue
abstract static class StagingResult {
static StagingResult success(Set<ArtifactMetadata> metadata) {
return new AutoValue_ArtifactServiceStager_StagingResult(metadata,
Collections.emptyMap());
}
- static StagingResult failure(Map<File, Throwable> failures) {
+ static StagingResult failure(Map<StagedFile, Throwable> failures) {
return new AutoValue_ArtifactServiceStager_StagingResult(
null, failures);
}
@@ -240,6 +263,7 @@ boolean isSuccess() {
@Nullable
abstract Set<ArtifactMetadata> getMetadata();
- abstract Map<File, Throwable> getFailures();
+ abstract Map<StagedFile, Throwable> getFailures();
}
+
}
diff --git
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ArtifactServiceStagerTest.java
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ArtifactServiceStagerTest.java
index 5c76ccf7f2f..a61ab9fcb83 100644
---
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ArtifactServiceStagerTest.java
+++
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/ArtifactServiceStagerTest.java
@@ -40,6 +40,7 @@
import java.util.HashSet;
import java.util.Set;
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.ArtifactMetadata;
+import
org.apache.beam.runners.core.construction.ArtifactServiceStager.StagedFile;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -87,7 +88,7 @@ public void testStage() throws Exception {
contentChannel.write(ByteBuffer.wrap(content));
}
- stager.stage(Collections.singleton(file));
+ stager.stage(Collections.singleton(StagedFile.of(file, file.getName())));
assertThat(service.getStagedArtifacts().entrySet(), hasSize(1));
byte[] stagedContent =
Iterables.getOnlyElement(service.getStagedArtifacts().values());
@@ -122,7 +123,11 @@ public void testStagingMultipleFiles() throws Exception {
contentChannel.write(ByteBuffer.wrap(thirdContent));
}
- stager.stage(ImmutableList.of(file, otherFile, thirdFile));
+ stager.stage(
+ ImmutableList.of(
+ StagedFile.of(file, file.getName()),
+ StagedFile.of(otherFile, otherFile.getName()),
+ StagedFile.of(thirdFile, thirdFile.getName())));
assertThat(service.getManifest().getArtifactCount(), equalTo(3));
assertThat(service.getStagedArtifacts().entrySet(), hasSize(3));
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/InProcessSdkHarness.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/InProcessSdkHarness.java
index 58f2614142e..fb177d059c6 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/InProcessSdkHarness.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/InProcessSdkHarness.java
@@ -19,8 +19,6 @@
package org.apache.beam.runners.fnexecution;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import io.grpc.ManagedChannel;
-import io.grpc.inprocess.InProcessChannelBuilder;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -34,8 +32,8 @@
import org.apache.beam.runners.fnexecution.data.GrpcDataService;
import org.apache.beam.runners.fnexecution.logging.GrpcLoggingService;
import org.apache.beam.runners.fnexecution.logging.Slf4jLogWriter;
-import org.apache.beam.sdk.fn.channel.ManagedChannelFactory;
import org.apache.beam.sdk.fn.stream.StreamObserverFactory;
+import org.apache.beam.sdk.fn.test.InProcessManagedChannelFactory;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.junit.rules.ExternalResource;
import org.junit.rules.TestRule;
@@ -98,12 +96,7 @@ protected void before() throws Exception {
PipelineOptionsFactory.create(),
loggingServer.getApiServiceDescriptor(),
controlServer.getApiServiceDescriptor(),
- new ManagedChannelFactory() {
- @Override
- public ManagedChannel forDescriptor(ApiServiceDescriptor
apiServiceDescriptor) {
- return
InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
- }
- },
+ new InProcessManagedChannelFactory(),
StreamObserverFactory.direct());
return null;
});
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
index 7e32a6b7a1f..d6d8fee568b 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
@@ -23,8 +23,6 @@
import static org.junit.Assert.assertThat;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import io.grpc.ManagedChannel;
-import io.grpc.inprocess.InProcessChannelBuilder;
import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
@@ -39,7 +37,6 @@
import java.util.concurrent.ThreadFactory;
import org.apache.beam.fn.harness.FnHarness;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.Target;
-import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
import org.apache.beam.model.pipeline.v1.RunnerApi;
import org.apache.beam.model.pipeline.v1.RunnerApi.Components;
import org.apache.beam.runners.core.construction.PipelineTranslation;
@@ -63,8 +60,8 @@
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.fn.channel.ManagedChannelFactory;
import org.apache.beam.sdk.fn.stream.StreamObserverFactory;
+import org.apache.beam.sdk.fn.test.InProcessManagedChannelFactory;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
@@ -122,12 +119,7 @@ public void setup() throws Exception {
PipelineOptionsFactory.create(),
loggingServer.getApiServiceDescriptor(),
controlServer.getApiServiceDescriptor(),
- new ManagedChannelFactory() {
- @Override
- public ManagedChannel forDescriptor(ApiServiceDescriptor
apiServiceDescriptor) {
- return
InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
- }
- },
+ new InProcessManagedChannelFactory(),
StreamObserverFactory.direct()));
// TODO: https://issues.apache.org/jira/browse/BEAM-4149 Use proper worker
id.
InstructionRequestHandler controlClient =
diff --git
a/runners/local-artifact-service-java/src/test/java/org/apache/beam/artifact/local/LocalFileSystemArtifactRetrievalServiceTest.java
b/runners/local-artifact-service-java/src/test/java/org/apache/beam/artifact/local/LocalFileSystemArtifactRetrievalServiceTest.java
index 82c6f54685e..1f7a5237e9f 100644
---
a/runners/local-artifact-service-java/src/test/java/org/apache/beam/artifact/local/LocalFileSystemArtifactRetrievalServiceTest.java
+++
b/runners/local-artifact-service-java/src/test/java/org/apache/beam/artifact/local/LocalFileSystemArtifactRetrievalServiceTest.java
@@ -30,9 +30,8 @@
import io.grpc.stub.StreamObserver;
import java.io.ByteArrayOutputStream;
import java.io.File;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.nio.ByteBuffer;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -48,6 +47,7 @@
import org.apache.beam.model.jobmanagement.v1.ArtifactApi.Manifest;
import org.apache.beam.model.jobmanagement.v1.ArtifactRetrievalServiceGrpc;
import org.apache.beam.runners.core.construction.ArtifactServiceStager;
+import
org.apache.beam.runners.core.construction.ArtifactServiceStager.StagedFile;
import org.apache.beam.runners.fnexecution.GrpcFnServer;
import org.apache.beam.runners.fnexecution.InProcessServerFactory;
import org.apache.beam.runners.fnexecution.ServerFactory;
@@ -186,11 +186,11 @@ public void onCompleted() {
}
private void stageAndCreateRetrievalService(Map<String, byte[]> artifacts)
throws Exception {
- List<File> artifactFiles = new ArrayList<>();
+ List<StagedFile> artifactFiles = new ArrayList<>();
for (Map.Entry<String, byte[]> artifact : artifacts.entrySet()) {
File artifactFile = tmp.newFile(artifact.getKey());
- new
FileOutputStream(artifactFile).getChannel().write(ByteBuffer.wrap(artifact.getValue()));
- artifactFiles.add(artifactFile);
+ Files.write(artifactFile.toPath(), artifact.getValue());
+ artifactFiles.add(StagedFile.of(artifactFile, artifactFile.getName()));
}
ArtifactServiceStager stager =
diff --git a/runners/reference/java/build.gradle
b/runners/reference/java/build.gradle
index 94109baf5de..b5f1413ecae 100644
--- a/runners/reference/java/build.gradle
+++ b/runners/reference/java/build.gradle
@@ -27,8 +27,11 @@ framework to execute user-definied functions."""
dependencies {
shadow project(path: ":beam-model-pipeline", configuration: "shadow")
shadow project(path: ":beam-runners-core-construction-java", configuration:
"shadow")
+ shadow project(path: ":beam-sdks-java-fn-execution", configuration: "shadow")
shadow library.java.slf4j_api
- testCompile library.java.junit
+ shadowTest project(path: ":beam-runners-core-construction-java",
configuration: "shadowTest")
testCompile library.java.hamcrest_core
+ testCompile library.java.junit
+ testCompile library.java.mockito_core
testCompile library.java.slf4j_jdk14
}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/CloseableResource.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/CloseableResource.java
new file mode 100644
index 00000000000..e10960fcb0e
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/CloseableResource.java
@@ -0,0 +1,120 @@
+/*
+ * 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.beam.runners.reference;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+import javax.annotation.Nullable;
+
+/**
+ * An {@link AutoCloseable} that wraps a resource that needs to be cleaned up
but does not implement
+ * {@link AutoCloseable} itself.
+ *
+ * <p>Recipients of a {@link CloseableResource} are in general responsible for
cleanup. Ownership
+ * can be transferred from one context to another via {@link #transfer()}.
Transferring relinquishes
+ * ownership from the original resource. This allows resources to be safely
constructed and
+ * transferred within a try-with-resources block. For example:
+ *
+ * {@code try (CloseableResource<Foo> resource = CloseableResource.of(...)) {
+ * // Do something with resource.
+ * ...
+ * // Then transfer ownership to some consumer.
+ * resourceConsumer(resource.transfer());
+ * }
+ * }
+ *
+ * <p>Not thread-safe.
+ */
+public class CloseableResource<T> implements AutoCloseable {
+
+ private final T resource;
+
+ /**
+ * {@link Closer } for the underlying resource. Closers are nullable to
allow transfer of
+ * ownership. However, newly-constructed {@link CloseableResource
CloseableResources} must always
+ * have non-null closers.
+ */
+ @Nullable private Closer<T> closer;
+
+ private boolean isClosed = false;
+
+ private CloseableResource(T resource, Closer<T> closer) {
+ this.resource = resource;
+ this.closer = closer;
+ }
+
+ /** Creates a {@link CloseableResource} with the given resource and closer.
*/
+ public static <T> CloseableResource<T> of(T resource, Closer<T> closer) {
+ checkArgument(resource != null, "Resource must be non-null");
+ checkArgument(closer != null, "%s must be non-null",
Closer.class.getName());
+ return new CloseableResource<>(resource, closer);
+ }
+
+ /** Gets the underlying resource. */
+ public T get() {
+ checkState(closer != null, "%s has transferred ownership",
CloseableResource.class.getName());
+ checkState(!isClosed, "% is closed", CloseableResource.class.getName());
+ return resource;
+ }
+
+ /**
+ * Returns a new {@link CloseableResource} that owns the underlying resource
and relinquishes
+ * ownership from this {@link CloseableResource}. {@link #close()} on the
original instance
+ * becomes a no-op.
+ */
+ public CloseableResource<T> transfer() {
+ checkState(closer != null, "%s has transferred ownership",
CloseableResource.class.getName());
+ checkState(!isClosed, "% is closed", CloseableResource.class.getName());
+ CloseableResource<T> other = CloseableResource.of(resource, closer);
+ this.closer = null;
+ return other;
+ }
+
+ /**
+ * Closes the underlying resource. The closer will only be executed on the
first call.
+ *
+ * @throws CloseException wrapping any exceptions thrown while closing
+ */
+ @Override
+ public void close() throws CloseException {
+ if (closer != null && !isClosed) {
+ try {
+ closer.close(resource);
+ } catch (Exception e) {
+ throw new CloseException(e);
+ } finally {
+ // Mark resource as closed even if we catch an exception.
+ isClosed = true;
+ }
+ }
+ }
+
+ /** A function that knows how to clean up after a resource. */
+ @FunctionalInterface
+ public interface Closer<T> {
+ void close(T resource) throws Exception;
+ }
+
+ /** An exception that wraps errors thrown while a resource is being closed.
*/
+ public static class CloseException extends Exception {
+ private CloseException(Exception e) {
+ super("Error closing resource", e);
+ }
+ }
+}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/JobServicePipelineResult.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/JobServicePipelineResult.java
new file mode 100644
index 00000000000..dbc33c7e22c
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/JobServicePipelineResult.java
@@ -0,0 +1,147 @@
+/*
+ * 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.beam.runners.reference;
+
+import com.google.protobuf.ByteString;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.model.jobmanagement.v1.JobApi.CancelJobRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.CancelJobResponse;
+import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateResponse;
+import
org.apache.beam.model.jobmanagement.v1.JobServiceGrpc.JobServiceBlockingStub;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class JobServicePipelineResult implements PipelineResult {
+
+ private static final long POLL_INTERVAL_MS = 10 * 1000;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(JobServicePipelineResult.class);
+
+ private final ByteString jobId;
+ private final CloseableResource<JobServiceBlockingStub> jobService;
+
+ JobServicePipelineResult(ByteString jobId,
CloseableResource<JobServiceBlockingStub> jobService) {
+ this.jobId = jobId;
+ this.jobService = jobService;
+ }
+
+ @Override
+ public State getState() {
+ JobServiceBlockingStub stub = jobService.get();
+ GetJobStateResponse response =
+
stub.getState(GetJobStateRequest.newBuilder().setJobIdBytes(jobId).build());
+ return getJavaState(response.getState());
+ }
+
+ @Override
+ public State cancel() {
+ JobServiceBlockingStub stub = jobService.get();
+ CancelJobResponse response =
+
stub.cancel(CancelJobRequest.newBuilder().setJobIdBytes(jobId).build());
+ return getJavaState(response.getState());
+ }
+
+ @Override
+ public State waitUntilFinish(Duration duration) {
+ if (duration.compareTo(Duration.millis(1)) < 1) {
+ // Equivalent to infinite timeout.
+ return waitUntilFinish();
+ } else {
+ CompletableFuture<State> result =
CompletableFuture.supplyAsync(this::waitUntilFinish);
+ try {
+ return result.get(duration.getMillis(), TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // Null result indicates a timeout.
+ return null;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public State waitUntilFinish() {
+ JobServiceBlockingStub stub = jobService.get();
+ GetJobStateRequest request =
GetJobStateRequest.newBuilder().setJobIdBytes(jobId).build();
+ GetJobStateResponse response = stub.getState(request);
+ State lastState = getJavaState(response.getState());
+ while (!lastState.isTerminal()) {
+ try {
+ Thread.sleep(POLL_INTERVAL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ response = stub.getState(request);
+ lastState = getJavaState(response.getState());
+ }
+ try {
+ jobService.close();
+ } catch (Exception e) {
+ LOG.warn("Error cleaning up job service", e);
+ }
+ return lastState;
+ }
+
+ @Override
+ public MetricResults metrics() {
+ throw new UnsupportedOperationException("Not yet implemented.");
+ }
+
+ private static State getJavaState(JobApi.JobState.Enum protoState) {
+ switch (protoState) {
+ case UNSPECIFIED:
+ return State.UNKNOWN;
+ case STOPPED:
+ return State.STOPPED;
+ case RUNNING:
+ return State.RUNNING;
+ case DONE:
+ return State.DONE;
+ case FAILED:
+ return State.FAILED;
+ case CANCELLED:
+ return State.CANCELLED;
+ case UPDATED:
+ return State.UPDATED;
+ case DRAINING:
+ // TODO: Determine the correct mappings for the states below.
+ return State.UNKNOWN;
+ case DRAINED:
+ return State.UNKNOWN;
+ case STARTING:
+ return State.RUNNING;
+ case CANCELLING:
+ return State.CANCELLED;
+ default:
+ LOG.warn("Unrecognized state from server: {}", protoState);
+ return State.UNKNOWN;
+ }
+ }
+}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/PortableRunner.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/PortableRunner.java
new file mode 100644
index 00000000000..13bb9f5780d
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/PortableRunner.java
@@ -0,0 +1,245 @@
+/*
+ * 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.beam.runners.reference;
+
+import static com.google.common.base.Preconditions.checkState;
+import static
org.apache.beam.runners.core.construction.PipelineResources.detectClassPathResourcesToStage;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+import com.google.protobuf.ByteString;
+import io.grpc.ManagedChannel;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Set;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PrepareJobRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PrepareJobResponse;
+import org.apache.beam.model.jobmanagement.v1.JobApi.RunJobRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.RunJobResponse;
+import org.apache.beam.model.jobmanagement.v1.JobServiceGrpc;
+import
org.apache.beam.model.jobmanagement.v1.JobServiceGrpc.JobServiceBlockingStub;
+import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
+import org.apache.beam.runners.core.construction.ArtifactServiceStager;
+import
org.apache.beam.runners.core.construction.ArtifactServiceStager.StagedFile;
+import org.apache.beam.runners.core.construction.JavaReadViaImpulse;
+import org.apache.beam.runners.core.construction.PipelineOptionsTranslation;
+import org.apache.beam.runners.core.construction.PipelineTranslation;
+import org.apache.beam.runners.reference.CloseableResource.CloseException;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.fn.channel.ManagedChannelFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsValidator;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.util.ZipFiles;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** A {@link PipelineRunner} a {@link Pipeline} against a {@code JobService}.
*/
+public class PortableRunner extends PipelineRunner<PipelineResult> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PortableRunner.class);
+
+ /** Provided pipeline options. */
+ private final PipelineOptions options;
+ /** Job API endpoint. */
+ private final String endpoint;
+ /** Files to stage to artifact staging service. They will ultimately be
added to the classpath. */
+ private final Collection<StagedFile> filesToStage;
+ /** Channel factory used to create communication channel with job and
staging services. */
+ private final ManagedChannelFactory channelFactory;
+
+ /**
+ * Constructs a runner from the provided options.
+ *
+ * @param options Properties which configure the runner.
+ * @return The newly created runner.
+ */
+ public static PortableRunner fromOptions(PipelineOptions options) {
+ return create(options, ManagedChannelFactory.createDefault());
+ }
+
+ @VisibleForTesting
+ static PortableRunner create(PipelineOptions options, ManagedChannelFactory
channelFactory) {
+ PortablePipelineOptions portableOptions =
+ PipelineOptionsValidator.validate(PortablePipelineOptions.class,
options);
+
+ String endpoint = portableOptions.getJobEndpoint();
+
+ // Deduplicate artifacts.
+ Set<String> pathsToStage = Sets.newHashSet();
+ if (portableOptions.getFilesToStage() == null) {
+
pathsToStage.addAll(detectClassPathResourcesToStage(PortableRunner.class.getClassLoader()));
+ if (pathsToStage.isEmpty()) {
+ throw new IllegalArgumentException("No classpath elements found.");
+ }
+ LOG.debug(
+ "PortablePipelineOptions.filesToStage was not specified. "
+ + "Defaulting to files from the classpath: {}",
+ pathsToStage.size());
+ } else {
+ pathsToStage.addAll(portableOptions.getFilesToStage());
+ }
+
+ ImmutableList.Builder<StagedFile> filesToStage = ImmutableList.builder();
+ for (String path : pathsToStage) {
+ File file = new File(path);
+ if (new File(path).exists()) {
+ // Spurious items get added to the classpath. Filter by just those
that exist.
+ if (file.isDirectory()) {
+ // Zip up directories so we can upload them to the artifact service.
+ try {
+ filesToStage.add(createStagingFile(zipDirectory(file)));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ filesToStage.add(createStagingFile(file));
+ }
+ }
+ }
+
+ return new PortableRunner(options, endpoint, filesToStage.build(),
channelFactory);
+ }
+
+ private PortableRunner(
+ PipelineOptions options,
+ String endpoint,
+ Collection<StagedFile> filesToStage,
+ ManagedChannelFactory channelFactory) {
+ this.options = options;
+ this.endpoint = endpoint;
+ this.filesToStage = filesToStage;
+ this.channelFactory = channelFactory;
+ }
+
+ @Override
+ public PipelineResult run(Pipeline pipeline) {
+
pipeline.replaceAll(ImmutableList.of(JavaReadViaImpulse.boundedOverride()));
+
+ LOG.debug("Initial files to stage: " + filesToStage);
+
+ PrepareJobRequest prepareJobRequest =
+ PrepareJobRequest.newBuilder()
+ .setJobName(options.getJobName())
+ .setPipeline(PipelineTranslation.toProto(pipeline))
+ .setPipelineOptions(PipelineOptionsTranslation.toProto(options))
+ .build();
+
+ ManagedChannel jobServiceChannel =
+ channelFactory.forDescriptor(
+ ApiServiceDescriptor.newBuilder()
+ .setUrl(endpoint).build());
+
+ JobServiceBlockingStub jobService =
JobServiceGrpc.newBlockingStub(jobServiceChannel);
+ try (CloseableResource<JobServiceBlockingStub> wrappedJobService =
+ CloseableResource.of(jobService, (unused) ->
jobServiceChannel.shutdown())) {
+
+ PrepareJobResponse prepareJobResponse =
jobService.prepare(prepareJobRequest);
+ LOG.info("PrepareJobResponse: {}", prepareJobResponse);
+
+ ApiServiceDescriptor artifactStagingEndpoint =
+ prepareJobResponse.getArtifactStagingEndpoint();
+
+ String stagingToken = null;
+ try (CloseableResource<ManagedChannel> artifactChannel =
+ CloseableResource.of(
+ channelFactory.forDescriptor(artifactStagingEndpoint),
ManagedChannel::shutdown)) {
+ ArtifactServiceStager stager =
ArtifactServiceStager.overChannel(artifactChannel.get());
+ LOG.debug("Actual files staged: {}", filesToStage);
+ stagingToken = stager.stage(filesToStage);
+ } catch (CloseableResource.CloseException e) {
+ LOG.warn("Error closing artifact staging channel", e);
+ // CloseExceptions should only be thrown while closing the channel.
+ checkState(stagingToken != null);
+ } catch (Exception e) {
+ throw new RuntimeException("Error staging files.", e);
+ }
+
+ RunJobRequest runJobRequest =
+ RunJobRequest.newBuilder()
+ .setPreparationId(prepareJobResponse.getPreparationId())
+ .setStagingToken(stagingToken)
+ .build();
+
+ RunJobResponse runJobResponse = jobService.run(runJobRequest);
+
+ LOG.info("RunJobResponse: {}", runJobResponse);
+ ByteString jobId = runJobResponse.getJobIdBytes();
+
+ return new JobServicePipelineResult(jobId, wrappedJobService.transfer());
+ } catch (CloseException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "PortableRunner#" + hashCode();
+ }
+
+ private static File zipDirectory(File directory) throws IOException {
+ File zipFile = File.createTempFile(directory.getName(), ".zip");
+ try (FileOutputStream fos = new FileOutputStream(zipFile)) {
+ ZipFiles.zipDirectory(directory, fos);
+ }
+ return zipFile;
+ }
+
+ private static StagedFile createStagingFile(File file) {
+ // TODO: https://issues.apache.org/jira/browse/BEAM-4109 Support arbitrary
names in the staging
+ // service itself.
+ // HACK: Encode the path name ourselves because the local artifact staging
service currently
+ // assumes artifact names correspond to a flat directory. Artifact staging
services should
+ // generally accept arbitrary artifact names.
+ // NOTE: Base64 url encoding does not work here because the stage artifact
names tend to be long
+ // and exceed file length limits on the artifact stager.
+ String encodedPath = escapePath(file.getPath());
+ return StagedFile.of(file, encodedPath);
+ }
+
+ /** Create a filename-friendly artifact name for the given path. */
+ // TODO: Are we missing any commonly allowed path characters that are
disallowed in file names?
+ private static String escapePath(String path) {
+ StringBuilder result = new StringBuilder(2 * path.length());
+ for (int i = 0; i < path.length(); i++) {
+ char c = path.charAt(i);
+ switch (c) {
+ case '_':
+ result.append("__");
+ break;
+ case '/':
+ result.append("_.");
+ break;
+ case '\\':
+ result.append("._");
+ break;
+ case '.':
+ result.append("..");
+ break;
+ default:
+ result.append(c);
+ }
+ }
+ return result.toString();
+ }
+}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/InProcessManagedChannelFactory.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/InProcessManagedChannelFactory.java
new file mode 100644
index 00000000000..e134aecc5be
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/InProcessManagedChannelFactory.java
@@ -0,0 +1,36 @@
+/*
+ * 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.beam.runners.reference.testing;
+
+import io.grpc.ManagedChannel;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
+import org.apache.beam.sdk.fn.channel.ManagedChannelFactory;
+
+/**
+ * A {@link org.apache.beam.sdk.fn.channel.ManagedChannelFactory} that uses
in-process channels.
+ *
+ * <p>The channel builder uses {@link ApiServiceDescriptor#getUrl()} as the
unique in-process name.
+ */
+public class InProcessManagedChannelFactory extends ManagedChannelFactory {
+
+ @Override
+ public ManagedChannel forDescriptor(ApiServiceDescriptor
apiServiceDescriptor) {
+ return
InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
+ }
+}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/TestJobService.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/TestJobService.java
new file mode 100644
index 00000000000..c0317143ddc
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/TestJobService.java
@@ -0,0 +1,78 @@
+/*
+ * 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.beam.runners.reference.testing;
+
+import io.grpc.stub.StreamObserver;
+import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateResponse;
+import org.apache.beam.model.jobmanagement.v1.JobApi.JobState;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PrepareJobRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PrepareJobResponse;
+import org.apache.beam.model.jobmanagement.v1.JobApi.RunJobRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.RunJobResponse;
+import
org.apache.beam.model.jobmanagement.v1.JobServiceGrpc.JobServiceImplBase;
+import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
+
+/**
+ * A JobService for tests.
+ *
+ * <p>A {@link TestJobService} always returns a fixed staging endpoint, job
preparation id, job id,
+ * and job state. As soon as a job is run, it is put into the given job state.
+ */
+public class TestJobService extends JobServiceImplBase {
+
+ private final ApiServiceDescriptor stagingEndpoint;
+ private final String preparationId;
+ private final String jobId;
+ private final JobState.Enum jobState;
+
+ public TestJobService(
+ ApiServiceDescriptor stagingEndpoint,
+ String preparationId,
+ String jobId,
+ JobState.Enum jobState) {
+ this.stagingEndpoint = stagingEndpoint;
+ this.preparationId = preparationId;
+ this.jobId = jobId;
+ this.jobState = jobState;
+ }
+
+ @Override
+ public void prepare(
+ PrepareJobRequest request, StreamObserver<PrepareJobResponse>
responseObserver) {
+ responseObserver.onNext(
+ PrepareJobResponse.newBuilder()
+ .setPreparationId(preparationId)
+ .setArtifactStagingEndpoint(stagingEndpoint)
+ .build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void run(RunJobRequest request, StreamObserver<RunJobResponse>
responseObserver) {
+
responseObserver.onNext(RunJobResponse.newBuilder().setJobId(jobId).build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void getState(
+ GetJobStateRequest request, StreamObserver<GetJobStateResponse>
responseObserver) {
+
responseObserver.onNext(GetJobStateResponse.newBuilder().setState(jobState).build());
+ responseObserver.onCompleted();
+ }
+}
diff --git
a/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/package-info.java
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/package-info.java
new file mode 100644
index 00000000000..a0969c391a6
--- /dev/null
+++
b/runners/reference/java/src/main/java/org/apache/beam/runners/reference/testing/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+/** Testing utilities for the reference runner. */
+package org.apache.beam.runners.reference.testing;
diff --git
a/runners/reference/java/src/test/java/org/apache/beam/runners/reference/CloseableResourceTest.java
b/runners/reference/java/src/test/java/org/apache/beam/runners/reference/CloseableResourceTest.java
new file mode 100644
index 00000000000..82c1fb5f306
--- /dev/null
+++
b/runners/reference/java/src/test/java/org/apache/beam/runners/reference/CloseableResourceTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.beam.runners.reference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.beam.runners.reference.CloseableResource.CloseException;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link CloseableResource}. */
+@RunWith(JUnit4.class)
+public class CloseableResourceTest {
+ @Rule public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void alwaysReturnsSameResource() {
+ Foo foo = new Foo();
+ CloseableResource<Foo> resource = CloseableResource.of(foo, (ignored) ->
{});
+ assertThat(resource.get(), is(foo));
+ assertThat(resource.get(), is(foo));
+ }
+
+ @Test
+ public void callsCloser() throws Exception {
+ AtomicBoolean closed = new AtomicBoolean(false);
+ try (CloseableResource<Foo> ignored =
+ CloseableResource.of(
+ new Foo(),
+ (foo) -> {
+ closed.set(true);
+ })) {
+ // Do nothing.
+ }
+ assertThat(closed.get(), is(true));
+ }
+
+ @Test
+ public void wrapsExceptionsInCloseException() throws Exception {
+ Exception wrapped = new Exception();
+ thrown.expect(CloseException.class);
+ thrown.expectCause(is(wrapped));
+ try (CloseableResource<Foo> ignored =
+ CloseableResource.of(
+ new Foo(),
+ (foo) -> {
+ throw wrapped;
+ })) {
+ // Do nothing.
+ }
+ }
+
+ @Test
+ public void transferReleasesCloser() throws Exception {
+ try (CloseableResource<Foo> foo =
+ CloseableResource.of(
+ new Foo(), (unused) -> fail("Transferred resource should not be
closed"))) {
+ foo.transfer();
+ }
+ }
+
+ @Test
+ public void transferMovesOwnership() throws Exception {
+ AtomicBoolean closed = new AtomicBoolean(false);
+ CloseableResource<Foo> original = CloseableResource.of(new Foo(), (unused)
-> closed.set(true));
+ CloseableResource<Foo> transferred = original.transfer();
+ transferred.close();
+ assertThat(closed.get(), is(true));
+ }
+
+ @Test
+ public void cannotTransferClosed() throws Exception {
+ CloseableResource<Foo> foo = CloseableResource.of(new Foo(), (unused) ->
{});
+ foo.close();
+ thrown.expect(IllegalStateException.class);
+ foo.transfer();
+ }
+
+ @Test
+ public void cannotTransferTwice() {
+ CloseableResource<Foo> foo = CloseableResource.of(new Foo(), (unused) ->
{});
+ foo.transfer();
+ thrown.expect(IllegalStateException.class);
+ foo.transfer();
+ }
+
+ private static class Foo {}
+}
diff --git
a/runners/reference/java/src/test/java/org/apache/beam/runners/reference/PortableRunnerTest.java
b/runners/reference/java/src/test/java/org/apache/beam/runners/reference/PortableRunnerTest.java
new file mode 100644
index 00000000000..8000d4f7ec1
--- /dev/null
+++
b/runners/reference/java/src/test/java/org/apache/beam/runners/reference/PortableRunnerTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.beam.runners.reference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessServerBuilder;
+import java.io.IOException;
+import java.io.Serializable;
+import org.apache.beam.model.jobmanagement.v1.JobApi.JobState;
+import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
+import org.apache.beam.runners.core.construction.InMemoryArtifactStagerService;
+import org.apache.beam.runners.reference.testing.TestJobService;
+import org.apache.beam.sdk.PipelineResult.State;
+import org.apache.beam.sdk.fn.test.InProcessManagedChannelFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link PortableRunner}. */
+@RunWith(JUnit4.class)
+public class PortableRunnerTest implements Serializable {
+
+ private static final String ENDPOINT_URL = "foo:3000";
+ private static final ApiServiceDescriptor ENDPOINT_DESCRIPTOR =
+ ApiServiceDescriptor.newBuilder().setUrl(ENDPOINT_URL).build();
+
+ private PipelineOptions options = createPipelineOptions();
+
+ @Rule public transient TestPipeline p = TestPipeline.fromOptions(options);
+
+ @Test
+ public void stagesAndRunsJob() throws Exception {
+ try (CloseableResource<Server> server =
createJobServer(JobState.Enum.DONE)) {
+ PortableRunner runner = PortableRunner.create(options, new
InProcessManagedChannelFactory());
+ State state = runner.run(p).waitUntilFinish();
+ assertThat(state, is(State.DONE));
+ }
+ }
+
+ private static CloseableResource<Server> createJobServer(JobState.Enum
jobState)
+ throws IOException {
+ CloseableResource<Server> server =
+ CloseableResource.of(
+ InProcessServerBuilder.forName(ENDPOINT_URL)
+ .addService(new TestJobService(ENDPOINT_DESCRIPTOR, "prepId",
"jobId", jobState))
+ .addService(new InMemoryArtifactStagerService())
+ .build(),
+ Server::shutdown);
+ server.get().start();
+ return server;
+ }
+
+ private static PipelineOptions createPipelineOptions() {
+ PortablePipelineOptions options =
+ PipelineOptionsFactory.create().as(PortablePipelineOptions.class);
+ options.setJobEndpoint(ENDPOINT_URL);
+ options.setRunner(PortableRunner.class);
+ return options;
+ }
+
+}
diff --git
a/runners/reference/job-server/src/test/java/org/apache/beam/runners/reference/job/ReferenceRunnerJobServiceTest.java
b/runners/reference/job-server/src/test/java/org/apache/beam/runners/reference/job/ReferenceRunnerJobServiceTest.java
index cfccc05c4dd..6d120c532b1 100644
---
a/runners/reference/job-server/src/test/java/org/apache/beam/runners/reference/job/ReferenceRunnerJobServiceTest.java
+++
b/runners/reference/job-server/src/test/java/org/apache/beam/runners/reference/job/ReferenceRunnerJobServiceTest.java
@@ -40,6 +40,7 @@
import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
import org.apache.beam.model.pipeline.v1.RunnerApi.Pipeline;
import org.apache.beam.runners.core.construction.ArtifactServiceStager;
+import
org.apache.beam.runners.core.construction.ArtifactServiceStager.StagedFile;
import org.apache.beam.runners.fnexecution.GrpcFnServer;
import org.apache.beam.runners.fnexecution.InProcessServerFactory;
import org.hamcrest.Description;
@@ -98,8 +99,9 @@ public void testPrepareJob() throws Exception {
InProcessChannelBuilder.forName(stagingEndpoint.getUrl()).build());
File foo = writeTempFile("foo", "foo, bar, baz".getBytes());
File bar = writeTempFile("spam", "spam, ham, eggs".getBytes());
- stager.stage(ImmutableList.of(foo, bar));
- List<byte[]> tempDirFiles = readFlattendFiles(runnerTemp.getRoot());
+ stager.stage(
+ ImmutableList.of(StagedFile.of(foo, foo.getName()), StagedFile.of(bar,
bar.getName())));
+ List<byte[]> tempDirFiles = readFlattenedFiles(runnerTemp.getRoot());
assertThat(
tempDirFiles,
hasItems(
@@ -122,11 +124,11 @@ public void describeTo(Description description) {
};
}
- private List<byte[]> readFlattendFiles(File root) throws Exception {
+ private List<byte[]> readFlattenedFiles(File root) throws Exception {
if (root.isDirectory()) {
List<byte[]> children = new ArrayList<>();
for (File child : root.listFiles()) {
- children.addAll(readFlattendFiles(child));
+ children.addAll(readFlattenedFiles(child));
}
return children;
} else {
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PortablePipelineOptions.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PortablePipelineOptions.java
new file mode 100644
index 00000000000..70efb7f16bd
--- /dev/null
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PortablePipelineOptions.java
@@ -0,0 +1,46 @@
+/*
+ * 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.beam.sdk.options;
+
+import java.util.List;
+import org.apache.beam.sdk.options.Validation.Required;
+
+/** Pipeline options common to all portable runners. */
+public interface PortablePipelineOptions extends PipelineOptions {
+
+ // TODO: https://issues.apache.org/jira/browse/BEAM-4106: Consider pulling
this out into a new
+ // options interface, e.g., FileStagingOptions.
+ /**
+ * List of local files to make available to workers.
+ *
+ * <p>Files are placed on the worker's classpath.
+ *
+ * <p>The default value is the list of jars from the main program's
classpath.
+ */
+ @Description(
+ "Files to stage to the artifact service and make available to workers.
Files are placed on "
+ + "the worker's classpath. The default value is all files from the
classpath.")
+ List<String> getFilesToStage();
+ void setFilesToStage(List<String> value);
+
+ @Description(
+ "Job service endpoint to use. Should be in the form of address and port,
e.g. localhost:3000")
+ @Required
+ String getJobEndpoint();
+ void setJobEndpoint(String endpoint);
+}
diff --git
a/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/test/InProcessManagedChannelFactory.java
b/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/test/InProcessManagedChannelFactory.java
new file mode 100644
index 00000000000..c25203e5d62
--- /dev/null
+++
b/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/test/InProcessManagedChannelFactory.java
@@ -0,0 +1,36 @@
+/*
+ * 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.beam.sdk.fn.test;
+
+import io.grpc.ManagedChannel;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
+import org.apache.beam.sdk.fn.channel.ManagedChannelFactory;
+
+/**
+ * A {@link ManagedChannelFactory} that uses in-process channels.
+ *
+ * <p>The channel builder uses {@link ApiServiceDescriptor#getUrl()} as the
unique in-process name.
+ */
+public class InProcessManagedChannelFactory extends ManagedChannelFactory {
+
+ @Override
+ public ManagedChannel forDescriptor(ApiServiceDescriptor
apiServiceDescriptor) {
+ return
InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 95277)
Time Spent: 18h (was: 17h 50m)
> Portable Runner Job API shim
> ----------------------------
>
> Key: BEAM-4071
> URL: https://issues.apache.org/jira/browse/BEAM-4071
> Project: Beam
> Issue Type: New Feature
> Components: runner-core
> Reporter: Ben Sidhom
> Assignee: Ben Sidhom
> Priority: Minor
> Time Spent: 18h
> Remaining Estimate: 0h
>
> There needs to be a way to execute Java-SDK pipelines against a portable job
> server. The job server itself is expected to be started up out-of-band. The
> "PortableRunner" should take an option indicating the Job API endpoint and
> defer other runner configurations to the backend itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
