Skip to content

docs: Document Fn Execution API protos #39293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
damccorm wants to merge 1 commit into
base: master
Choose a base branch
from
+62 −5
Open

docs: Document Fn Execution API protos #39293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 61 additions & 4 deletions model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,7 @@ service BeamFnControl {

// Requests the ProcessBundleDescriptor with the given id.
message GetProcessBundleDescriptorRequest {
// (Required) The id of the ProcessBundleDescriptor to retrieve.
string process_bundle_descriptor_id = 1;
}

Expand Down Expand Up @@ -426,6 +427,8 @@ message ProcessBundleRequest {
bool only_bundle_for_keys = 5;
}

// The response to a ProcessBundleRequest, returned by the SDK harness after
// bundle processing completes (or is partially completed with residual roots).
message ProcessBundleResponse {
// (Optional) Specifies that the bundle has not been completed and the
// following applications need to be scheduled and executed in the future.
Expand Down Expand Up @@ -495,6 +498,8 @@ message MonitoringInfosMetadataRequest {
repeated string monitoring_info_id = 1;
}

// A response containing progress information for a currently active bundle.
// This is returned in response to a ProcessBundleProgressRequest.
message ProcessBundleProgressResponse {
// DEPRECATED (Required) The list of metrics or other MonitoredState
// collected while processing this bundle.
Expand Down Expand Up @@ -705,12 +710,17 @@ message ProcessBundleSplitResponse {
}


// A request sent by the runner to finalize a bundle that previously indicated
// it requires finalization (via requires_finalization in ProcessBundleResponse).
// This is sent after the runner has committed the output of the bundle, allowing
// the SDK to perform any post-commit cleanup.
message FinalizeBundleRequest {
// (Required) A reference to a completed process bundle request with the given
// instruction id.
string instruction_id = 1;
}

// A response to a FinalizeBundleRequest, indicating that finalization is complete.
message FinalizeBundleResponse {
// Empty
}
Expand Down Expand Up @@ -749,10 +759,16 @@ message Elements {
bool is_last = 4;
}

// Indicates whether the bundle is being drained. When draining, sources
// should stop producing new data and the SDK should finish processing
// in-flight elements.
message DrainMode {
enum Enum {
// Drain mode has not been specified.
UNSPECIFIED = 0;
// The bundle is not being drained; normal processing continues.
NOT_DRAINING = 1;
// The bundle is being drained; sources should stop producing new data.
DRAINING = 2;
}
}
Expand Down Expand Up @@ -783,6 +799,8 @@ message Elements {
// Element metadata passed as part of WindowedValue to make WindowedValue
// extensible and backward compatible
message ElementMetadata {
// (Optional) Indicates whether the bundle is being drained, allowing the
// SDK to handle drain semantics for this element.
optional DrainMode.Enum drain = 1;
// (Optional) As part of https://www.w3.org/TR/trace-context/ we are forwarding a trace and participating in it.
// Traceparent header represents the incoming request in a tracing system in a common format.
Expand Down Expand Up @@ -847,6 +865,9 @@ service BeamFnData {
* State API
*/

// A request sent by the SDK harness to the runner to get, append, or clear

@tvalentyn tvalentyn Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we should not repeat things that are self-evident and can become stale overtime. We could say something general to manipulate state perhaps.

// state associated with a particular StateKey. State requests are scoped to
// the instruction (bundle) that is currently being processed.
message StateRequest {
// (Required) A unique identifier provided by the SDK which represents this
// requests execution. The StateResponse corresponding with this request
Expand Down Expand Up @@ -874,6 +895,8 @@ message StateRequest {
}
}

// A response from the runner to a StateRequest, containing the result of
// the requested state operation.
message StateResponse {
// (Required) A reference provided by the SDK which represents a requests
// execution. The StateResponse must have the matching id when responding
Expand All @@ -898,6 +921,10 @@ message StateResponse {
}
}

// An API for the SDK harness to access state stored by the runner on behalf
// of the SDK. State is scoped to the currently processing bundle and can be
// used for user state, side inputs, and runner-managed references.
// Stable
service BeamFnState {
// Used to get/append/clear state stored by the runner on behalf of the SDK.
rpc State(
Expand All @@ -909,6 +936,9 @@ service BeamFnState {
stream StateResponse) {}
}

// A key identifying which state to access. Exactly one of the oneof type
// fields must be set, determining the kind of state being requested
// (e.g. side input data, user state, or runner-managed references).
message StateKey {
message Runner {
// (Required) Opaque information supplied by the runner. Used to support
Expand Down Expand Up @@ -1184,18 +1214,24 @@ message StateAppendRequest {
bytes data = 1;
}

// A response to append state.
// A response to a StateAppendRequest. Currently empty as append operations
// do not return data.
message StateAppendResponse {}

// A request to clear state.
// A request to clear all state associated with the given StateKey.
message StateClearRequest {}

// A response to clear state.
// A response to a StateClearRequest. Currently empty as clear operations
// do not return data.
message StateClearResponse {}

// A message describes a sort key range [start, end).
// A message describing a sort key range [start, end) for ordered list state.
// Used in OrderedListUserState to specify which portion of the ordered list
// to retrieve.
message OrderedListRange {
// (Required) The inclusive start of the sort key range.
int64 start = 1;
// (Required) The exclusive end of the sort key range.
int64 end = 2;
}

Expand Down Expand Up @@ -1289,6 +1325,8 @@ message LogEntry {
google.protobuf.Struct custom_data = 9;
}

// A control message sent from the runner to the SDK harness to configure
// logging behavior. Currently empty, reserved for future log-level control.
message LogControl {}

// Stable
Expand All @@ -1303,27 +1341,46 @@ service BeamFnLogging {
stream LogControl) {}
}

// A request from the runner to start a new SDK worker process with the given
// configuration, including endpoint addresses and parameters.
message StartWorkerRequest {
// (Required) A unique identifier for the worker to start.
string worker_id = 1;
// (Required) The control endpoint the worker should connect to.
org.apache.beam.model.pipeline.v1.ApiServiceDescriptor control_endpoint = 2;
// (Optional) The logging endpoint the worker should use.
org.apache.beam.model.pipeline.v1.ApiServiceDescriptor logging_endpoint = 3;
// (Optional) The artifact retrieval endpoint the worker should use.
org.apache.beam.model.pipeline.v1.ApiServiceDescriptor artifact_endpoint = 4;
// (Optional) The provisioning endpoint the worker should use.
org.apache.beam.model.pipeline.v1.ApiServiceDescriptor provision_endpoint = 5;
// (Optional) Additional runner-specific parameters to pass to the worker.
map<string, string> params = 10;
}

// A response from a StartWorkerRequest. If the worker failed to start,
// the error field will contain a human-readable error message.
message StartWorkerResponse {
// (Optional) Error message if the worker failed to start.
string error = 1;
}

// A request from the runner to stop a running SDK worker process.
message StopWorkerRequest {
// (Required) The unique identifier of the worker to stop.
string worker_id = 1;
}

// A response from a StopWorkerRequest. If the worker failed to stop

@tvalentyn tvalentyn Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: i'd remove 1374-1375 as self-evident or shorten to first sentence + append content to the error as suggested

// cleanly, the error field will contain a human-readable error message.
message StopWorkerResponse {
// (Optional) Error message if the worker failed to stop.

@tvalentyn tvalentyn Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
// (Optional) Error message if the worker failed to stop.
// (Optional) Human-readable error message if the worker failed to stop.

string error = 1;
}

// An API for the runner to manage external SDK worker processes.
// Allows the runner to start and stop SDK worker instances, typically used
// for containerized or process-based execution environments.
service BeamFnExternalWorkerPool {
// Start the SDK worker with the given ID.
rpc StartWorker (StartWorkerRequest) returns (StartWorkerResponse) {}
Expand Down
2 changes: 1 addition & 1 deletion ...n-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_provision_api.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ message ProvisionInfo {
// (optional) The control endpoint this SDK should use.
org.apache.beam.model.pipeline.v1.ApiServiceDescriptor control_endpoint = 10;

// The set of dependencies that should be staged into this environment.
// (optional) The set of dependencies that should be staged into this environment.
repeated org.apache.beam.model.pipeline.v1.ArtifactInformation dependencies = 11;

// (optional) A set of capabilities that this SDK is allowed to use in its
Expand Down
Loading