Re: [PR] KAFKA-19479: at_least_once mode in Kafka Streams silently drops messages when the producer fails with MESSAGE_TOO_LARGE, violating delivery guarantees [kafka]

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

junrao commented on PR #20254:
URL: https://github.com/apache/kafka/pull/20254#issuecomment-3152755901

   @mjsax : Hmm, the splitting code has some logic to chain the response futures. When we append a record to a new batch after a split, we create a new future for the record and chain it to the old future.
   
   ```
       private boolean tryAppendForSplit(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers, Thunk thunk) {
   ...
   
               FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,
                                                                      timestamp,
                                                                      key == null ? -1 : key.remaining(),
                                                                      value == null ? -1 : value.remaining(),
                                                                      Time.SYSTEM);
               // Chain the future to the original thunk.
               thunk.future.chain(future);
   ```
   
   `FutureRecordMetadata.chain()` then adds the new future as a dependency.
   ```
       void chain(FutureRecordMetadata futureRecordMetadata) {
           if (nextRecordMetadata == null)
               nextRecordMetadata = futureRecordMetadata;
           else
               nextRecordMetadata.chain(futureRecordMetadata);
       }
   
   ```
   
   `FutureRecordMetadata.get()` will then wait until all chained response futures complete.
   ```
       public RecordMetadata get() throws InterruptedException, ExecutionException {
           this.result.await();
           if (nextRecordMetadata != null)
               return nextRecordMetadata.get();
           return valueOrError();
       }
   ```
   
   So, a pending `flush()` waiting for an original batch is not supposed to complete until all split batches have completed. Did you actually observe that `flush()` returns early?
   
   @lianetm : Currently, `KafkaProducer.doSend()` only rejects a record if its size is > max.request.size. It's ok if a record has a size larger than batch.size. We will just create a larger new batch to accommodate for it. So, during split, it's reasonable to follow the same approach to allow the first record in a new batch. 
   


-- 
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: jira-unsubscribe@kafka.apache.org

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