[Dataflow Streaming] Reduce contention on CommitQueue

[arunpandianp]

Replaces LinkedBlockingQueue with a ConcurrentLinkedQueue and a Semaphore. The harness threads don't block on the queue anymore.

Improves the internal reshuffle benchmark run time by ~10%.

#33578

[arunpandianp]

R: @scwhittle

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[bvolpato]

Is delayed visibility acceptable here? With concurrent producers, one can pause after queue.add() but before availableItems.release(). Another producer can enqueue and release, allowing consumer to dequeue first item using second permit and leaving second item unavailable until first producer resumes. Does commit queue require each completed put() to become immediately consumable?

[arunpandianp]

@bvolpato Thanks, that is a good observation. Such interleaving are possible and i think it is fine, with how the queue is used today in StreamingEngineWorkCommitter and StreamingApplianceCommitter.

It is true that a consumer can consume items using permits released after a different item. Generally the logic to stop consumers (or draining queue) should know and happen after all producer puts are complete. So that edge i think resolves the visibility problems here.

Having said that, changing to LinkedTransferQueue might be a simpler option with similar benefits. Testing it.

[scwhittle]

LinkedTransferQueue does seem like it would be simpler, so I'd prefer that.