You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2025/11/27 20:17:51 UTC
[PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
1996fanrui opened a new pull request, #27282:
URL: https://github.com/apache/flink/pull/27282
## What is the purpose of the change
## Problem
`RecordsWindowBuffer.addElement()` catches `EOFException` and retries recursively, causing **StackOverflowError** when retry keeps failing.
## Root Cause
The original code assumes "flush will always free enough space for retry". This assumption fails when unrecoverable errors occur, leading to infinite recursion.
## Solution
Use `numKeys == 0` as the termination condition:
```java
while (true) {
LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
try {
recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
break;
} catch (EOFException e) {
if (recordsBuffer.getNumKeys() == 0) {
// Buffer is empty, retry won't help (unrecoverable error)
throw e;
}
// Buffer has data, flush and retry
flush();
checkState(recordsBuffer.getNumKeys() == 0, "The recordsBuffer should be empty after flushing.");
}
}
```
## Why This Works
| Scenario | numKeys before catch | Behavior |
|----------|---------------------|----------|
| Buffer full (recoverable) | > 0 | flush → numKeys=0 → retry → success |
| Unrecoverable error (1st attempt) | 0 | throw immediately |
| Unrecoverable error (after flush retry) | 0 | flush → numKeys=0 → retry → fail again → numKeys still 0 → throw |
Key point: `flush()` clears the buffer, making `numKeys == 0`. If retry still fails after flush, entering the catch block again with `numKeys == 0` indicates the problem is not caused by a full buffer, but an unrecoverable error. In this case, an exception should be thrown instead of continuing to retry.
## Benefits
1. **Prevents StackOverflowError**: The while loop executes at most twice (initial attempt + retry after flush), no infinite loop
2. **Preserves normal behavior**: Normal flush + retry still works when buffer is full
3. **Better diagnostics**: Unrecoverable errors throw a complete EOFException with full stack trace, making it easy to identify the root cause
4. **Minimal change**: Only checks numKeys, no additional state variables introduced
## Brief change log
- [hotfix] WindowBuffer implements AutoCloseable to avoid declare close explicitly
- [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement
## Verifying this change
- Added `RecordsWindowBufferTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #27282:
URL: https://github.com/apache/flink/pull/27282#discussion_r2585421585
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/buffers/RecordsWindowBuffer.java:
##########
@@ -84,14 +85,22 @@ public void addElement(RowData key, long sliceEnd, RowData element) throws Excep
minSliceEnd = Math.min(sliceEnd, minSliceEnd);
reuseWindowKey.replace(sliceEnd, key);
- LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
- try {
- recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
- } catch (EOFException e) {
- // buffer is full, flush it to state
- flush();
- // remember to add the input element again
- addElement(key, sliceEnd, element);
+ while (true) {
+ LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
+ try {
+ recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
+ break;
+ } catch (EOFException e) {
+ if (recordsBuffer.getNumKeys() == 0) {
+ // Buffer is empty, retry won't help (record is too large for the buffer)
+ throw e;
+ }
Review Comment:
They are equivalent.
- One element must have one key
- Every key has at least one element
- `numKeys == 0` ⟺ `numElements == 0` ⟺ buffer is empty
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/buffers/RecordsWindowBuffer.java:
##########
@@ -84,14 +85,22 @@ public void addElement(RowData key, long sliceEnd, RowData element) throws Excep
minSliceEnd = Math.min(sliceEnd, minSliceEnd);
reuseWindowKey.replace(sliceEnd, key);
- LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
- try {
- recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
- } catch (EOFException e) {
- // buffer is full, flush it to state
- flush();
- // remember to add the input element again
- addElement(key, sliceEnd, element);
+ while (true) {
Review Comment:
Good catch!
I have moved it between `append` and `break`, and introduced a new test to cover this case.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #27282:
URL: https://github.com/apache/flink/pull/27282#discussion_r2582756034
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/buffers/RecordsWindowBuffer.java:
##########
@@ -84,14 +85,22 @@ public void addElement(RowData key, long sliceEnd, RowData element) throws Excep
minSliceEnd = Math.min(sliceEnd, minSliceEnd);
reuseWindowKey.replace(sliceEnd, key);
- LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
- try {
- recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
- } catch (EOFException e) {
- // buffer is full, flush it to state
- flush();
- // remember to add the input element again
- addElement(key, sliceEnd, element);
+ while (true) {
Review Comment:
Do we also need to move the above line
`minSliceEnd = Math.min(sliceEnd, minSliceEnd);`
inside the loop?
In `flush()`, it might be reset to `Long.MAX_VALUE`, so we'll end up with the wrong `minSliceEnd` field
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/buffers/RecordsWindowBuffer.java:
##########
@@ -84,14 +85,22 @@ public void addElement(RowData key, long sliceEnd, RowData element) throws Excep
minSliceEnd = Math.min(sliceEnd, minSliceEnd);
reuseWindowKey.replace(sliceEnd, key);
- LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
- try {
- recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
- } catch (EOFException e) {
- // buffer is full, flush it to state
- flush();
- // remember to add the input element again
- addElement(key, sliceEnd, element);
+ while (true) {
+ LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
+ try {
+ recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
+ break;
+ } catch (EOFException e) {
+ if (recordsBuffer.getNumKeys() == 0) {
+ // Buffer is empty, retry won't help (record is too large for the buffer)
+ throw e;
+ }
Review Comment:
Is `getNumKeys` correct and sufficient?
Or should `getNumElements` be used instead or in addition?
In `append()`, I see that `numKeys` is not incremented if `lookupInfo.found`.
I guess there's an implicit invariant that numKeys > 0 in this case, but I'm not sure.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27282:
URL: https://github.com/apache/flink/pull/27282#issuecomment-3587190048
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "012be30220bb905882914d2f71f8ad5406aeafdf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "012be30220bb905882914d2f71f8ad5406aeafdf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 012be30220bb905882914d2f71f8ad5406aeafdf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #27282:
URL: https://github.com/apache/flink/pull/27282
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38746][table-runtime] Fix StackOverflowError in RecordsWindowBuffer.addElement [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #27282:
URL: https://github.com/apache/flink/pull/27282#issuecomment-3588598980
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org