fhan688 commented on issue #19664:
URL: https://github.com/apache/hudi/issues/19664#issuecomment-5354747705
> I validated the failure path against the current Hudi code and Flink
2.1.1's `BinaryInMemorySortBuffer.write()` implementation. The report is
correct: on an `EOFException`, Flink returns `false` before updating
`currentDataBufferOffset`, although the paged output may already have advanced;
Hudi can then retry that same buffer when a different bucket is selected as the
largest bucket.
>
> The PR 1 approach looks right. A few implementation details are important:
>
> * Distinguish “could not create a new buffer” from “an existing buffer's
`writeRow` returned false.” In the latter case, make the bucket unavailable for
any further write immediately.
> * If the diverged bucket has committed records, flush only those indexed
records and then dispose it. If it has no committed records, skip `flushBucket`
(the current method rejects an empty bucket) and just dispose it.
> * Put disposal/page return on an exception-safe path. If flushing the
committed records or the one fresh-buffer retry throws/fails, the failed buffer
must still be disposed before propagating the error.
> * A focused `RowDataBucket` test should verify that a second write after
`write() == false` is rejected, and recovery tests should verify page
restoration in addition to record count and variable-length values.
>
> Keeping owner-aware preemption in PR 2 is a good separation. Calling back
into the writer from `MemorySegmentPool.nextSegment()` introduces a re-entrant
flush while serialization is in progress, so that path deserves the separate
owner/progress/cleanup tests proposed here.
Thanks Danny for validating the failure path and for the detailed
implementation suggestions.
I agree with all four points. I will keep PR 1 focused on the correctness
guard and handle the two failure modes explicitly:
1. Failure to create a new buffer because the minimum number of pages is
unavailable.
2. Failure from `writeRow()` on an existing buffer, after which the bucket
is immediately marked as diverged and cannot accept any
further writes.
For a diverged bucket, PR 1 will:
- flush only the indexed/committed records when the bucket is non-empty;
- skip `flushBucket` and directly dispose it when no record was committed;
- guarantee disposal and page return through exception-safe cleanup paths;
- retry the record only once and only on a fresh buffer;
- dispose the fresh buffer as well if the retry fails.
I will also add:
- a focused `RowDataBucket` test verifying that a second write to a
diverged buffer is rejected;
- recovery tests that validate record count and variable-length field
values;
- explicit page-restoration checks for both on-heap and managed memory
pools.
I will keep owner-aware preemption in PR 2. As you noted, invoking a
writer flush from `MemorySegmentPool.nextSegment()` is re-entrant
while serialization is still in progress, so that PR will include
dedicated owner-exclusion, progress, exception-cleanup, and page-
return tests.
Thanks again. I will start with the correctness PR first.
--
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]
