You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jk...@apache.org on 2018/06/21 19:39:32 UTC
[beam] 01/01: Merge pull request #5680: [BEAM-4145] Populate the
worker_id metadata in the Java SDK Harness
This is an automated email from the ASF dual-hosted git repository.
jkff pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 77e93d925b6aaf6c0f764b521c60a1d35976a821
Merge: 809fcf9 56f1ecc
Author: Eugene Kirpichov <ek...@gmail.com>
AuthorDate: Thu Jun 21 12:39:27 2018 -0700
Merge pull request #5680: [BEAM-4145] Populate the worker_id metadata in the Java SDK Harness
[BEAM-4145] Populate the worker_id metadata in the Java SDK Harness
.../environment/InProcessEnvironmentFactory.java | 1 +
.../fnexecution/control/RemoteExecutionTest.java | 1 +
.../testing/InProcessManagedChannelFactory.java | 36 -----------
sdks/go/pkg/beam/util/grpcx/metadata.go | 2 +-
sdks/java/container/boot.go | 1 +
.../beam/sdk/fn/channel/ManagedChannelFactory.java | 63 ++++++++++++++----
.../fn/test/InProcessManagedChannelFactory.java | 6 +-
.../java/org/apache/beam/fn/harness/FnHarness.java | 75 ++++++++++++----------
.../harness/control/AddHarnessIdInterceptor.java | 41 ++++++++++++
.../fn/harness/control/BeamFnControlClient.java | 17 +++--
.../org/apache/beam/fn/harness/FnHarnessTest.java | 2 +-
.../harness/control/BeamFnControlClientTest.java | 15 ++---
12 files changed, 158 insertions(+), 102 deletions(-)
diff --cc runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
index e969e5f,cd9f8b0..e2e9bcb
--- 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
@@@ -132,10 -129,10 +132,11 @@@ public class RemoteExecutionTest implem
// Create the SDK harness, and wait until it connects
sdkHarnessExecutor = Executors.newSingleThreadExecutor(threadFactory);
- sdkHarnessExecutor.submit(
- () ->
+ sdkHarnessExecutorFuture = sdkHarnessExecutor.submit(
+ () -> {
+ try {
FnHarness.main(
+ "id",
PipelineOptionsFactory.create(),
loggingServer.getApiServiceDescriptor(),
controlServer.getApiServiceDescriptor(),
diff --cc sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
index 5bac1ac,aa6a6fb..d13d42a
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
@@@ -120,12 -131,13 +131,13 @@@ public class FnHarness
Endpoints.ApiServiceDescriptor loggingApiServiceDescriptor,
Endpoints.ApiServiceDescriptor controlApiServiceDescriptor,
ManagedChannelFactory channelFactory,
- OutboundObserverFactory outboundObserverFactory) {
+ OutboundObserverFactory outboundObserverFactory) throws Exception {
IdGenerator idGenerator = IdGenerators.decrementingLongs();
- try (BeamFnLoggingClient logging = new BeamFnLoggingClient(
- options,
- loggingApiServiceDescriptor,
- channelFactory::forDescriptor)) {
+ // The logging client variable is not used per se, but during its lifetime (until close()) it
+ // intercepts logging and sends it to the logging service.
+ try (BeamFnLoggingClient logging =
+ new BeamFnLoggingClient(
+ options, loggingApiServiceDescriptor, channelFactory::forDescriptor)) {
LOG.info("Fn Harness started");
// Register standard file systems.