Shuffle flushes can resend buffered tuples
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### Feature Summary
Hash-based and range-based shuffle partitioners emit pending tuples during control and state flushes but leave those tuples in their buffers. A later flush sends the same tuples again. Round-robin, broadcast, and one-to-one partitioners already drain their flushed buffers.
Before: first flush sends data and control; second flush sends the same data again
After: first flush sends data and control; second flush sends only control
Expected behavior:
A successful flush should drain only the emitted receiver buffer so tuples cannot be duplicated across control or state boundaries.
Reproduction evidence:
On `50321e403c82df299a13deb50a7f9849dd93bdba`, add one tuple to a hash shuffle partitioner and flush the same receiver twice.
Observed result:
```text
first=['list', 'EmbeddedControlMessage']
second=['list', 'EmbeddedControlMessage']
```
The same retained-buffer behavior occurs for range shuffle and for state flushes.
Version and commit evidence:
1.3.0-incubating-SNAPSHOT (main)
**Commit Hash (Optional)**
`50321e403c82df299a13deb50a7f9849dd93bdba`
### Proposed Solution or Design
After: first flush sends data and control; second flush sends only control
Expected behavior:
### Affected Area
Workflow Engine (Amber)
Contributor guide
Assessment
This issue has not been assessed yet.