[GitHub] [beam] tweise commented on a change in pull request #14942: [BEAM-8137] Add Main method to ExternalWorkerService

[GitBox <gi...@apache.org>]

tweise commented on a change in pull request #14942:
URL: https://github.com/apache/beam/pull/14942#discussion_r646747491



##########
File path: runners/portability/java/src/main/java/org/apache/beam/runners/portability/ExternalWorkerService.java
##########
@@ -90,10 +97,44 @@ public void stopWorker(
   public void close() {}
 
   public GrpcFnServer<ExternalWorkerService> start() throws Exception {
-    GrpcFnServer<ExternalWorkerService> server =
-        GrpcFnServer.allocatePortAndCreateFor(this, serverFactory);
+    final String externalServiceAddress =
+        Environments.getExternalServiceAddress(options.as(PortablePipelineOptions.class));
+    GrpcFnServer<ExternalWorkerService> server;
+    if (externalServiceAddress.isEmpty()) {
+      server = GrpcFnServer.allocatePortAndCreateFor(this, serverFactory);
+    } else {
+      server =
+          GrpcFnServer.create(
+              this,
+              Endpoints.ApiServiceDescriptor.newBuilder().setUrl(externalServiceAddress).build(),
+              serverFactory);
+    }
     LOG.debug(
         "Listening for worker start requests at {}.", server.getApiServiceDescriptor().getUrl());
     return server;
   }
+
+  public static void main(String[] args) throws Exception {
+    main(System::getenv);
+  }
+
+  public static void main(Function<String, String> environmentVarGetter) throws Exception {
+    System.out.format("Starting external worker service%n");
+    System.out.format("Pipeline options %s%n", environmentVarGetter.apply(PIPELINE_OPTIONS));
+    PipelineOptions options =
+        PipelineOptionsTranslation.fromJson(environmentVarGetter.apply(PIPELINE_OPTIONS));
+
+    try (GrpcFnServer<ExternalWorkerService> server = new ExternalWorkerService(options).start()) {
+      System.out.format(
+          "External worker service started at address: %s",
+          server.getApiServiceDescriptor().getUrl());
+      while (true) {
+        Sleeper.DEFAULT.sleep(60 * 60 * 24 * 1000);

Review comment:
       Document why we wait/sleep here.

##########
File path: runners/portability/java/src/main/java/org/apache/beam/runners/portability/ExternalWorkerService.java
##########
@@ -90,10 +97,44 @@ public void stopWorker(
   public void close() {}
 
   public GrpcFnServer<ExternalWorkerService> start() throws Exception {
-    GrpcFnServer<ExternalWorkerService> server =
-        GrpcFnServer.allocatePortAndCreateFor(this, serverFactory);
+    final String externalServiceAddress =
+        Environments.getExternalServiceAddress(options.as(PortablePipelineOptions.class));
+    GrpcFnServer<ExternalWorkerService> server;
+    if (externalServiceAddress.isEmpty()) {
+      server = GrpcFnServer.allocatePortAndCreateFor(this, serverFactory);
+    } else {
+      server =
+          GrpcFnServer.create(
+              this,
+              Endpoints.ApiServiceDescriptor.newBuilder().setUrl(externalServiceAddress).build(),
+              serverFactory);
+    }
     LOG.debug(
         "Listening for worker start requests at {}.", server.getApiServiceDescriptor().getUrl());
     return server;
   }
+
+  public static void main(String[] args) throws Exception {

Review comment:
       Perhaps documentation with an example how to start the worker pool would be useful?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org