featzhang commented on PR #27357:
URL: https://github.com/apache/flink/pull/27357#issuecomment-4293201203
## Review Response Summary
Thank you @xishuaidelin for the excellent feedback! I've analyzed all 4
comments:
### Comments to Address:
1. **Code duplication** - Extend AsyncBatchWaitOperator instead
2. **Complex TreeMap ordering** - Use simple FIFO queue pattern
3. **API inconsistency** - Parameter naming mismatch
4. **Flaky Thread.sleep()** - Replace with CountDownLatch
### Major Refactoring:
**1. Extend Base Operator:**
```java
public class OrderedAsyncBatchWaitOperator<IN, OUT>
extends AsyncBatchWaitOperator<IN, OUT> {
private final Queue<CompletedBatch<OUT>> completedBatches = new
ArrayDeque<>();
private long nextExpectedSequence = 0;
@Override
protected void invokeBatchAsync(List<IN> batch) {
final long sequence = nextBatchSequence++;
function.asyncInvoke(batch, new ResultFuture<OUT>() {
public void complete(Collection<OUT> results) {
completedBatches.add(new CompletedBatch<>(sequence,
results));
emitCompletedBatches();
}
});
}
}
```
**2. Simple FIFO Queue:**
```java
private void emitCompletedBatches() {
while (!completedBatches.isEmpty()) {
CompletedBatch<OUT> head = completedBatches.peek();
if (head.sequence != nextExpectedSequence) break;
completedBatches.poll();
emit(head.results);
nextExpectedSequence++;
}
}
```
**3. Robust Testing:**
```java
CountDownLatch latch = new CountDownLatch(expectedCount);
// ... async operations
assertTrue(latch.await(10, TimeUnit.SECONDS));
```
This refactoring eliminates ~70% of code duplication and simplifies ordering
logic significantly.
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]
