featzhang commented on PR #27355:
URL: https://github.com/apache/flink/pull/27355#issuecomment-4293199814
## Review Response Summary
Thank you @xishuaidelin for the detailed review! I've analyzed all 3
comments and prepared implementation guidance:
### Comments to Address:
1. **Buffer data loss on failover** - Need to add state snapshotting
2. **No backpressure mechanism** - Need to add capacity-based blocking
3. **Missing object reuse support** - Need to add isObjectReuseEnabled
### Detailed Implementation Guide:
I've created a comprehensive guide with code examples for all changes. The
key implementations needed:
**1. State Snapshotting:**
```java
private transient ListState<StreamElement> recoveredStreamElements;
@Override
public void snapshotState(StateSnapshotContext context) {
recoveredStreamElements.clear();
for (StreamElement element : buffer) {
recoveredStreamElements.add(element);
}
}
```
**2. Backpressure:**
```java
private void addToWorkQueue(StreamRecord<IN> element) {
while (buffer.size() + inFlightCount >= maxCapacity) {
mailboxExecutor.yield();
}
buffer.add(new StreamElement(element));
}
```
**3. Object Reuse:**
```java
private StreamElement copyStreamElement(StreamRecord<IN> element) {
if (isObjectReuseEnabled) {
IN copied = typeSerializer.copy(element.getValue());
return new StreamElement(new StreamRecord<>(copied,
element.getTimestamp()));
}
return new StreamElement(element);
}
```
I'll update the PR with these changes soon. Full guide available with test
cases and best practices.
cc @featzhang
--
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]
