You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sc...@apache.org on 2026/04/07 12:15:44 UTC
(beam) branch master updated: add waitForReady to logging stream and control stream to handle sdk starting before the runner. (#37982)
This is an automated email from the ASF dual-hosted git repository.
scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new f7c280c77d2 add waitForReady to logging stream and control stream to handle sdk starting before the runner. (#37982)
f7c280c77d2 is described below
commit f7c280c77d231f04cc8cf60bfd5cb17b32c8b2af
Author: Sam Whittle <sc...@users.noreply.github.com>
AuthorDate: Tue Apr 7 12:15:38 2026 +0000
add waitForReady to logging stream and control stream to handle sdk starting before the runner. (#37982)
---
.../java/org/apache/beam/fn/harness/FnHarness.java | 35 ++++++++++------------
.../fn/harness/logging/BeamFnLoggingClient.java | 3 +-
2 files changed, 18 insertions(+), 20 deletions(-)
diff --git 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
index 034695237d8..d0724432f3c 100644
--- 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
@@ -144,16 +144,23 @@ public class FnHarness {
@VisibleForTesting
public static void main(Function<String, String> environmentVarGetter) throws Exception {
JvmInitializers.runOnStartup();
- System.out.format("SDK Fn Harness started%n");
- System.out.format("Harness ID %s%n", environmentVarGetter.apply(HARNESS_ID));
- System.out.format(
- "Logging location %s%n", environmentVarGetter.apply(LOGGING_API_SERVICE_DESCRIPTOR));
- System.out.format(
- "Control location %s%n", environmentVarGetter.apply(CONTROL_API_SERVICE_DESCRIPTOR));
- System.out.format(
- "Status location %s%n", environmentVarGetter.apply(STATUS_API_SERVICE_DESCRIPTOR));
+
+ Endpoints.ApiServiceDescriptor loggingApiServiceDescriptor =
+ getApiServiceDescriptor(environmentVarGetter.apply(LOGGING_API_SERVICE_DESCRIPTOR));
+ Endpoints.ApiServiceDescriptor controlApiServiceDescriptor =
+ getApiServiceDescriptor(environmentVarGetter.apply(CONTROL_API_SERVICE_DESCRIPTOR));
+ Endpoints.ApiServiceDescriptor statusApiServiceDescriptor =
+ environmentVarGetter.apply(STATUS_API_SERVICE_DESCRIPTOR) == null
+ ? null
+ : getApiServiceDescriptor(environmentVarGetter.apply(STATUS_API_SERVICE_DESCRIPTOR));
String id = environmentVarGetter.apply(HARNESS_ID);
+ System.out.format("SDK Fn Harness started%n");
+ System.out.format("Harness ID %s%n", id);
+ System.out.format("Logging location %s%n", loggingApiServiceDescriptor);
+ System.out.format("Control location %s%n", controlApiServiceDescriptor);
+ System.out.format("Status location %s%n", statusApiServiceDescriptor);
+
String pipelineOptionsJson = environmentVarGetter.apply(PIPELINE_OPTIONS);
// Try looking for a file first. If that exists it should override PIPELINE_OPTIONS to avoid
// maxing out the kernel's environment space
@@ -179,16 +186,6 @@ public class FnHarness {
PipelineOptions options = PipelineOptionsTranslation.fromJson(pipelineOptionsJson);
- Endpoints.ApiServiceDescriptor loggingApiServiceDescriptor =
- getApiServiceDescriptor(environmentVarGetter.apply(LOGGING_API_SERVICE_DESCRIPTOR));
-
- Endpoints.ApiServiceDescriptor controlApiServiceDescriptor =
- getApiServiceDescriptor(environmentVarGetter.apply(CONTROL_API_SERVICE_DESCRIPTOR));
-
- Endpoints.ApiServiceDescriptor statusApiServiceDescriptor =
- environmentVarGetter.apply(STATUS_API_SERVICE_DESCRIPTOR) == null
- ? null
- : getApiServiceDescriptor(environmentVarGetter.apply(STATUS_API_SERVICE_DESCRIPTOR));
String runnerCapabilitesOrNull = environmentVarGetter.apply(RUNNER_CAPABILITIES);
Set<String> runnerCapabilites =
runnerCapabilitesOrNull == null
@@ -415,7 +412,7 @@ public class FnHarness {
// directExecutor() when building the channel.
BeamFnControlClient control =
new BeamFnControlClient(
- controlStub.withExecutor(MoreExecutors.directExecutor()),
+ controlStub.withExecutor(MoreExecutors.directExecutor()).withWaitForReady(),
outboundObserverFactory,
executorService,
handlers);
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
index 99a7ee64832..97179115c04 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
@@ -214,7 +214,8 @@ public class BeamFnLoggingClient implements LoggingClient {
this.streamPhaser = new AdvancingPhaser(1);
this.channel = channel;
- BeamFnLoggingGrpc.BeamFnLoggingStub stub = BeamFnLoggingGrpc.newStub(channel);
+ BeamFnLoggingGrpc.BeamFnLoggingStub stub =
+ BeamFnLoggingGrpc.newStub(channel).withWaitForReady();
this.inboundObserver = new LogControlObserver();
this.outboundObserver =
new DirectStreamObserver<BeamFnApi.LogEntry.List>(