Add SDK capability to detect if the SDK Fn Harness data channel is busy.

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

@@ -503,6 +503,11 @@ 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 that the SDK is still busy consuming the data that as already
+  // been received on the data channel. If this is set, a runner may abstain
+  // from sending further data on the data channel until this field becomes
+  // unset. This field is currently used during shuffle reads on large elements.
+  optional bool consuming_received_data = 6;
 
   reserved 1, 2, 4;
 }

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto

@@ -1672,6 +1672,11 @@ message StandardProtocols {
     // during bundle processing. This is disabled by default and enabled with the
     // `enable_data_sampling` experiment.
     DATA_SAMPLING = 8 [(beam_urn) = "beam:protocol:data_sampling:v1"];
+
+    // Indicates whether the SDK sets the consuming_received_data field on
+    // the ProcessBundleProgressResponse messages.
+    SDK_CONSUMING_RECEIVED_DATA = 9
+    [(beam_urn) = "beam:protocol:sdk_consuming_received_data:v1"];
   }
 }