Re: [PR] [FLINK-38825][streaming] Add ordered async batch support for AsyncBatchWaitOperator [flink]

["featzhang (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org