[BEAM-4145] Populate the worker_id metadata in the Java SDK Harness by tgroh · Pull Request #5456 · apache/beam

tgroh · 2018-05-23T18:45:09Z

The ID is required for any control service with multiple clients, which is the general case.

Enable ManagedChannelFactory implementations to register interceptors on all of the
created channels.

Add an interceptor which attaches the ID to the metadata. Use this interceptor in the
Sdk Harness Control Client.

Populate the ID passed via the container contract in the boot go code.

Follow this checklist to help us incorporate your contribution quickly and easily:

Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

tgroh · 2018-05-23T18:58:52Z

R: @angoenka @youngoli

tgroh · 2018-05-24T20:29:09Z

run java precommit

angoenka · 2018-05-25T21:12:39Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java


-  public abstract ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor);
+  public final ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor) {
+    return forDescriptorOnly(apiServiceDescriptor);


Shall we get away with this indirection?
Implementer can always overwrite this public method (if we drop the final).

angoenka · 2018-05-25T21:22:19Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

+    @Override
+    protected ManagedChannel forDescriptor(
+        ApiServiceDescriptor apiServiceDescriptor, List<ClientInterceptor> interceptors) {
+      return channelFactory.forDescriptor(apiServiceDescriptor, interceptors);


The behavior in not well defined.
Though this is an intercepted factory, the original interceptors (interceptors from constructor) are not applied.

angoenka · 2018-05-25T21:25:54Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

+
+    @Override
+    public ManagedChannelFactory withInterceptors(List<ClientInterceptor> interceptors) {
+      return new InterceptedManagedChannelFactory(channelFactory, interceptors);


I suppose this method is for Factory chaining.
Should we pass "this" in that case?

I would suggest to remove chaining in this manner and provide a builder to do the chaining.
Something like this:
ManagedChannelFactoryBuilder.builder(channelFactory).withInterceptors(interceptors)

We are loosing the original list of interceptors here.

Yes, I'm also curious about this. If it's not for factory chaining and isn't necessary for another reason, could it be removed?

angoenka · 2018-05-25T21:30:05Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

+    return forDescriptorOnly(apiServiceDescriptor);
+  }
+
+  protected ManagedChannel forDescriptorOnly(ApiServiceDescriptor descriptor) {


Based on the previous comment, we can remove this method.

angoenka · 2018-05-25T21:34:30Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

    }
  }
+
+  private static class InterceptedManagedChannelFactory extends ManagedChannelFactory {


What is the purpose of this class?
Based on method ManagedChannelFactory.forDescriptor(ApiServiceDescriptor apiServiceDescriptor, List<ClientInterceptor> interceptors)
All implementations of ManagedChannelFactory should support interception and if not, they should throw exception.

angoenka · 2018-05-25T21:41:55Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java


      LOG.info("Fn Harness started");
-      EnumMap<BeamFnApi.InstructionRequest.RequestCase,
+      EnumMap<


Please reformat the file.

angoenka · 2018-05-25T21:43:44Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java

-        options,
-        loggingApiServiceDescriptor,
-        channelFactory::forDescriptor)) {
+    try (BeamFnLoggingClient logging =


logging is not used. Shall we remove it?

No. This is done to close the client automatically at the completion of the try block, and the client must exist to send intercepted LOG messages to the logging server.

Can we add a comment explaining this as logging is unused which makes this code block non intuitive.

angoenka · 2018-05-25T21:54:26Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/AddHarnessIdInterceptor.java

-  @Override
-  public ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor) {
-    return InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
+  public static ClientInterceptor create(String harnessId) {


Shall we have a null check for harnessId?

angoenka · 2018-05-25T21:55:12Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/AddHarnessIdInterceptor.java

+import io.grpc.stub.MetadataUtils;
+
+/** A {@link ClientInterceptor} that attaches a provided SDK Harness ID to outgoing messages. */
+public class AddHarnessIdInterceptor {


Not sure if we need a repository of interceptors. Just putting it as a thought.

I find this much more readable as a factory method than as an inline key and MetadataUtils call - it tells us exactly the thing it does.

angoenka · 2018-05-25T21:56:01Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/BeamFnControlClient.java

-              StreamObserverClientFactory<InstructionRequest, BeamFnApi.InstructionResponse>,
-              StreamObserver<BeamFnApi.InstructionRequest>,
-              StreamObserver<BeamFnApi.InstructionResponse>>
+              StreamObserverClientFactory<InstructionRequest, InstructionResponse>,


This is autoformatted.

tgroh

I've changed the implementation of ManagedChannelFactory a little. Instead of having multiple methods, the public API is exactly forDescriptor(ApiServiceDescriptor), and the abstract API is builderFor(ApiServiceDescriptor), which returns a ManagedChannelBuilder - which can have interceptors attached, or can be otherwise modified as appropriate for the factory.

tgroh · 2018-05-26T00:52:37Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java

-        ObjectMapper.findModules(ReflectHelpers.findClassLoader()));
-    PipelineOptions options = objectMapper.readValue(
-        System.getenv(PIPELINE_OPTIONS), PipelineOptions.class);
+    String id = System.getenv(HARNESS_ID);


tgroh · 2018-05-26T00:53:25Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java


      LOG.info("Fn Harness started");
-      EnumMap<BeamFnApi.InstructionRequest.RequestCase,
+      EnumMap<


tgroh · 2018-05-26T00:53:40Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java

-        options,
-        loggingApiServiceDescriptor,
-        channelFactory::forDescriptor)) {
+    try (BeamFnLoggingClient logging =


No. This is done to close the client automatically at the completion of the try block, and the client must exist to send intercepted LOG messages to the logging server.

tgroh · 2018-05-26T00:54:28Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/AddHarnessIdInterceptor.java

-  @Override
-  public ManagedChannel forDescriptor(ApiServiceDescriptor apiServiceDescriptor) {
-    return InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();
+  public static ClientInterceptor create(String harnessId) {


tgroh · 2018-05-26T00:54:57Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/AddHarnessIdInterceptor.java

+import io.grpc.stub.MetadataUtils;
+
+/** A {@link ClientInterceptor} that attaches a provided SDK Harness ID to outgoing messages. */
+public class AddHarnessIdInterceptor {


I find this much more readable as a factory method than as an inline key and MetadataUtils call - it tells us exactly the thing it does.

tgroh · 2018-05-26T00:55:14Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/BeamFnControlClient.java

-              StreamObserverClientFactory<InstructionRequest, BeamFnApi.InstructionResponse>,
-              StreamObserver<BeamFnApi.InstructionRequest>,
-              StreamObserver<BeamFnApi.InstructionResponse>>
+              StreamObserverClientFactory<InstructionRequest, InstructionResponse>,


This is autoformatted.

youngoli

Approving, with one comment.

youngoli · 2018-06-01T19:26:06Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

+
+    @Override
+    public ManagedChannelFactory withInterceptors(List<ClientInterceptor> interceptors) {
+      return new InterceptedManagedChannelFactory(channelFactory, interceptors);


Yes, I'm also curious about this. If it's not for factory chaining and isn't necessary for another reason, could it be removed?

This populates a metadata field with the Harness ID.

This duplicates an available implementation in java-fn-execution, which is the one we use everywhere.

Expand the public API to add a `withInterceptors` method, which returns a ManagedChannelFactory which will create calls via the same underlying implementation, but with the provided interceptors attached. Expand the implementation requirements to take a list of ClientInterceptors, which must intercept any returned channel. The public API to get a channel from a ManagedChannelFactory remains the same.

Update the java container boot code to populate an environment variable with the harness ID. Add an ID parameter to FnHarness main methods. Pass the ID to the BeamFnControlClient in its constructor, and attach it to all outgoing `control` calls.

angoenka

PR looks good overall.
There is 1 major underlying comment at https://github.com/apache/beam/pull/5456/files#diff-379a0983f269015757adc77995b9eaa4R118
and a minor comment on commenting of 'logging'

angoenka · 2018-05-31T23:57:02Z

sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java

+
+    @Override
+    public ManagedChannelFactory withInterceptors(List<ClientInterceptor> interceptors) {
+      return new InterceptedManagedChannelFactory(channelFactory, interceptors);


We are loosing the original list of interceptors here.

angoenka · 2018-06-04T18:44:04Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java

-        options,
-        loggingApiServiceDescriptor,
-        channelFactory::forDescriptor)) {
+    try (BeamFnLoggingClient logging =


Can we add a comment explaining this as logging is unused which makes this code block non intuitive.

angoenka · 2018-06-04T18:44:46Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/AddHarnessIdInterceptor.java

+import io.grpc.stub.MetadataUtils;
+
+/** A {@link ClientInterceptor} that attaches a provided SDK Harness ID to outgoing messages. */
+public class AddHarnessIdInterceptor {


jkff · 2018-06-19T00:36:13Z

This PR is replaced by #5680.

tgroh force-pushed the fn_harness_id branch from 30102e7 to 27c1c86 Compare May 24, 2018 17:51

angoenka reviewed May 25, 2018

View reviewed changes

tgroh commented May 26, 2018

View reviewed changes

youngoli approved these changes Jun 1, 2018

View reviewed changes

tgroh added 4 commits June 1, 2018 18:56

Add an AddHarnessIdInterceptor

adcf704

This populates a metadata field with the Harness ID.

Remove the runners/reference InProcessManagedChannelFactory

9f41fcd

This duplicates an available implementation in java-fn-execution, which is the one we use everywhere.

tgroh force-pushed the fn_harness_id branch from e4f8a66 to 6641523 Compare June 2, 2018 01:56

angoenka reviewed Jun 4, 2018

View reviewed changes

jkff mentioned this pull request Jun 19, 2018

[BEAM-4145] Populate the worker_id metadata in the Java SDK Harness #5680

Merged

jkff closed this Jun 19, 2018

Conversation

tgroh commented May 23, 2018

Uh oh!

tgroh commented May 23, 2018

Uh oh!

tgroh commented May 24, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tgroh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

youngoli left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

angoenka left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jkff commented Jun 19, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects