priyansndesai commented on code in PR #31420:
URL: https://github.com/apache/beam/pull/31420#discussion_r1619069578
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java:
##########
@@ -119,7 +127,12 @@ public void close() throws Exception {
public void awaitCompletion() throws Exception {
try {
while (true) {
+ // The SDK is available to process data right before it is ready to
take elements off the
+ // queue.
+ consumingReceivedData.set(false);
BeamFnApi.Elements elements = queue.take();
+ // The SDK is now no longer available to receive more data, so we set
it to false.
Review Comment:
Addressed.
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -151,6 +157,8 @@ func (n *DataSource) process(ctx context.Context, data
func(bcr *byteCountReader
// io.EOF means the reader successfully drained.
// We're ready for a new buffer.
case <-ctx.Done():
+ // now that it is done processing received data, we set
it to false.
Review Comment:
It is to ensure that when we exit, we have correctly reset the value.
##########
model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto:
##########
@@ -503,6 +503,8 @@ message ProcessBundleProgressResponse {
// as the MonitoringInfo could be reconstructed fully by overwriting its
// payload field with the bytes specified here.
map<string, bytes> monitoring_data = 5;
+ // Indicates if the SDK is consuming received data or not.
Review Comment:
Clarified. Thanks!
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
