You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2018/09/19 21:07:13 UTC
[beam] branch revert-6151-beam4780-2 created (now 7714ae8)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a change to branch revert-6151-beam4780-2
in repository https://gitbox.apache.org/repos/asf/beam.git.
at 7714ae8 Revert "[BEAM-4780] Updating to DockerJobBundleFactory in ReferenceRunner."
This branch includes the following new commits:
new 7714ae8 Revert "[BEAM-4780] Updating to DockerJobBundleFactory in ReferenceRunner."
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.
[beam] 01/01: Revert "[BEAM-4780] Updating to
DockerJobBundleFactory in ReferenceRunner."
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch revert-6151-beam4780-2
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 7714ae822faa981eb4b23ca271e3598f99238852
Author: Lukasz Cwik <lc...@google.com>
AuthorDate: Wed Sep 19 14:07:04 2018 -0700
Revert "[BEAM-4780] Updating to DockerJobBundleFactory in ReferenceRunner."
---
.../runners/direct/portable/ReferenceRunner.java | 23 +---------------------
1 file changed, 1 insertion(+), 22 deletions(-)
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/ReferenceRunner.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/ReferenceRunner.java
index ee3ae0e..ae4eb57 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/ReferenceRunner.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/ReferenceRunner.java
@@ -69,7 +69,6 @@ import org.apache.beam.runners.fnexecution.InProcessServerFactory;
import org.apache.beam.runners.fnexecution.ServerFactory;
import org.apache.beam.runners.fnexecution.artifact.ArtifactRetrievalService;
import org.apache.beam.runners.fnexecution.control.ControlClientPool;
-import org.apache.beam.runners.fnexecution.control.DockerJobBundleFactory;
import org.apache.beam.runners.fnexecution.control.FnApiControlClientPoolService;
import org.apache.beam.runners.fnexecution.control.JobBundleFactory;
import org.apache.beam.runners.fnexecution.control.MapControlClientPool;
@@ -80,7 +79,6 @@ import org.apache.beam.runners.fnexecution.environment.EnvironmentFactory;
import org.apache.beam.runners.fnexecution.environment.InProcessEnvironmentFactory;
import org.apache.beam.runners.fnexecution.logging.GrpcLoggingService;
import org.apache.beam.runners.fnexecution.logging.Slf4jLogWriter;
-import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
import org.apache.beam.runners.fnexecution.provisioning.StaticGrpcProvisionService;
import org.apache.beam.runners.fnexecution.state.GrpcStateService;
import org.apache.beam.runners.fnexecution.wire.LengthPrefixUnknownCoders;
@@ -162,7 +160,6 @@ public class ReferenceRunner {
ServerFactory serverFactory = createServerFactory();
ControlClientPool controlClientPool = MapControlClientPool.create();
ExecutorService dataExecutor = Executors.newCachedThreadPool();
- // TODO(BEAM-5095): Fill out ProvisionInfo/JobInfo properly, without placeholder values.
ProvisionInfo provisionInfo =
ProvisionInfo.newBuilder()
.setJobId("id")
@@ -171,7 +168,6 @@ public class ReferenceRunner {
.setWorkerId("foo")
.setResourceLimits(Resources.getDefaultInstance())
.build();
- JobInfo jobInfo = JobInfo.create("id", "reference", "retrieval-token", options);
try (GrpcFnServer<GrpcLoggingService> logging =
GrpcFnServer.allocatePortAndCreateFor(
GrpcLoggingService.forWriter(Slf4jLogWriter.getDefault()), serverFactory);
@@ -202,7 +198,7 @@ public class ReferenceRunner {
createEnvironmentFactory(
control, logging, artifact, provisioning, controlClientPool.getSource());
JobBundleFactory jobBundleFactory =
- createJobBundleFactory(jobInfo, environmentFactory, data, state);
+ SingleEnvironmentInstanceJobBundleFactory.create(environmentFactory, data, state);
TransformEvaluatorRegistry transformRegistry =
TransformEvaluatorRegistry.portableRegistry(
@@ -257,23 +253,6 @@ public class ReferenceRunner {
}
}
- private JobBundleFactory createJobBundleFactory(
- JobInfo jobInfo,
- EnvironmentFactory environmentFactory,
- GrpcFnServer<GrpcDataService> data,
- GrpcFnServer<GrpcStateService> state)
- throws Exception {
- switch (environmentType) {
- case DOCKER:
- return DockerJobBundleFactory.create(jobInfo);
- case IN_PROCESS:
- return SingleEnvironmentInstanceJobBundleFactory.create(environmentFactory, data, state);
- default:
- throw new IllegalArgumentException(
- String.format("Unknown %s %s", EnvironmentType.class.getSimpleName(), environmentType));
- }
- }
-
@VisibleForTesting
static class PortableGroupByKeyReplacer implements TransformReplacement {
@Override