[Bug][Functions] Event-time window function never acks late messages under ATLEAST_ONCE
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Read release policy
- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.
### User environment
Pulsar Functions, Java window function WindowFunctionExecutor.
### Issue Description
What happened:
When an event-time window function receives a message whose extracted timestamp is behind the current watermark, WindowFunctionExecutor.process() treats it as a late tuple and keeps it out of the window. With lateDataTopic unset it logs Received a late tuple. This will not be processed. With lateDataTopic set it republishes the value there through sendAsync(). Neither path acks the record.
Under ATLEAST_ONCE the only ack happens in processWindow(), which walks the tuples of the window that just fired. A late message never enters a window, so it is never in that list. Window functions also rewrite functionDetails.processingGuarantees to MANUAL and delegate acking to WindowFunctionExecutor, so JavaInstanceRunnable does not ack it either.
Expected:
Under ATLEAST_ONCE, a message the runtime has decided not to process should still be acked, or explicitly failed, so the subscription cursor can move past it.
### Error messages
```text
There is no exception. The only signal is that log line, plus a growing backlog and unacked count on the function's input subscription when late event-time data arrives.
```
### Reproducing the issue
1. Deploy a Java window function with timestampExtractorClassName, windowLengthDurationMs, slidingIntervalDurationMs, maxLagMs and watermarkEmitIntervalMs set, processingGuarantees=ATLEAST_ONCE, and no lateDataTopic.
2. Produce messages with increasing event timestamps so the watermark advances.
3. Produce one message whose extracted timestamp is behind the current watermark.
4. The log shows the late tuple line, the message never appears in any window invocation, and the input subscription never acks it. Running the same steps with ATMOST_ONCE acks the message.
### Additional information
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
Research direction
Start with the Java WindowFunctionExecutor.process() path for event-time tuples, then compare its late-tuple handling with processWindow() and the JavaInstanceRunnable acknowledgement behavior. Trace both unset and configured lateDataTopic cases under ATLEAST_ONCE; done means a late message is acknowledged or explicitly failed rather than remaining unacked, without changing the ATMOST_ONCE behavior described.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100