featzhang commented on PR #27356:
URL: https://github.com/apache/flink/pull/27356#issuecomment-4293200519
## Review Response Summary
Thank you @xishuaidelin for the thorough review! I've analyzed all 6
comments:
### Comments to Address:
1. **Timer edge case** - Old timer firing for new batch
2. **Unbounded inflight count** - Need max inflight limit
3. **Duplicate constant** - Consolidate definitions
4. **Redundant Serializable** - Remove from AsyncBatchFunction
5. **Redundant comments** - Clean up obvious comments
6. **Test helper simplification** - Reduce duplication
### Key Fixes:
**1. Timer Edge Case:**
```java
private Long currentTimerTimestamp = null;
private void flushBuffer() {
// Cancel old timer to prevent race condition
if (currentTimerTimestamp != null) {
timerService.deleteProcessingTimeTimer(currentTimerTimestamp);
currentTimerTimestamp = null;
}
// ... flush logic
}
```
**2. Max Inflight Limit:**
```java
private void flushBuffer() {
while (inFlightCount >= maxInFlightBatches) {
mailboxExecutor.yield();
}
inFlightCount++;
// ... invoke async
}
```
**3. Remove Redundant Serializable:**
```java
// Function already extends Serializable
public interface AsyncBatchFunction<IN, OUT> extends Function {
// No need for explicit Serializable
}
```
Full implementation guide available with test cases. Will update PR soon.
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]
