github-actions[bot] commented on code in PR #40788:
URL: https://github.com/apache/beam/pull/40788#discussion_r4371934848
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java:
##########
@@ -86,6 +89,7 @@ private CloseException() {
private int numEndpointsThatAreIncomplete;
private final AtomicBoolean consumingReceivedData;
+ private @Nullable IntConsumer onElementsConsumed;
Review Comment:
**🟡 Medium**: `onElementsConsumed` is written by `setOnElementsConsumed`
during `registerConsumer` and read without synchronization in
`awaitCompletion()`, `close()`, and `reset()`. Those methods can run on
different threads without a guaranteed happens-before edge (for example, when
`awaitCompletion()` is started asynchronously before `registerReceiver` is
called, as in `BeamFnDataGrpcClientTest.testForInboundConsumerThatIsPoisoned`,
or when `close()` is called from a control-plane thread). Marking this field
`volatile` ensures visibility across threads.
```suggestion
private volatile @Nullable IntConsumer onElementsConsumed;
```
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java:
##########
@@ -114,6 +140,9 @@ public StreamObserver<BeamFnApi.Elements>
getOutboundObserver() {
*/
public void registerConsumer(
String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements>
receiver) {
Review Comment:
**🟡 Medium**: Two subtle issues with setting `onElementsConsumed` here:
1. It mutates `receiver` before `receivers.compute(...)` verifies that
`instructionId` is not already registered or poisoned.
2. Conditioning `setOnElementsConsumed` on `inboundObserver.requestObserver
!= null` at registration time while `forwardToConsumerForInstructionId` also
checks `this.requestObserver != null` at message delivery time creates an
ordering hazard: if `registerConsumer` is called before `beforeStart(stream)`
(e.g., in tests or lazy stream initialization), `onElementsConsumed` remains
`null`, but subsequent `onNext` calls see `this.requestObserver != null`,
increment `FLOW_CONTROL_STATE`, and call `consumer.add(value)` without ever
releasing those permits.
Since `InboundObserver.accept(int)` and `releaseOrParkFlowControlToken()`
already guard against `this.requestObserver == null`, consider always wiring
`setOnElementsConsumed(inboundObserver)` (inside `receivers.compute` once
registration succeeds) and always using the `FLOW_CONTROL_STATE` + `add(value)`
path for `BeamFnDataInboundObserver` in `forwardToConsumerForInstructionId` so
increments and decrements cannot get out of sync.
--
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]
