mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-25 06:20:42 +01:00
filer: stop an oversized metadata log flush from wedging the change feed (#10430)
* filer: write the metadata log in pieces a volume server will accept A single oversized metadata event grows the log buffer past the volume server's fileSizeLimitMB, and the flush of that buffer is then rejected forever: the retry loop has no exit, so the blob at the head of the queue blocks every later flush and the metadata feed stalls until restart. Split the flushed buffer into BufferSize pieces, on record boundaries where possible so each piece still decodes on its own, and retry each piece separately so a partial success is not replayed. Log files are already read as a chunk stream, with a whole-file fallback when a chunk does not decode standalone, so a record may cross a piece boundary. * log_buffer: let go of a window array grown for an oversized entry An entry larger than BufferSize grows the window array to 2*size+4, and window arrays cycle through SealBuffer rather than being freed. One such entry therefore leaves every later window carrying its size, and currentSnapshotView allocates a snapshot as wide as the array on each window, so a few KB of metadata keeps paying for it. Drop the array when SealBuffer hands it back. Growth is on demand, so the next oversized entry just reallocates. * iceberg maintenance: store merged data files as chunks, not inline saveFilerFile had no size threshold, so compaction wrote whole merged parquet files -- hundreds of MB -- as Entry.Content. That puts the parquet bytes verbatim in the filer store and sends them through the metadata change log again as one event. Keep manifests and metadata JSON inline, upload anything larger to volume servers in chunks, assigning through the filer so the path's storage rules apply. * filer: follow the file size limit the volume servers report The starting piece size is a constant, so a cluster whose -fileSizeLimitMB is set below it would reject every piece and wedge just the same. The rejection names the limit, so take it from there and re-cut the rest of the flush to fit. Piece the buffer one at a time rather than up front, since the size can change partway through a flush.
This commit is contained in:
1 parent
cba2e5150c
commit
7b3462be6a
7 files changed
+572
-14
No files matched your search
@@ -616,6 +616,14 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush
|
||||
// CRITICAL: logBuffer.offset is the "next offset to assign", so last offset in buffer is offset-1
|
||||
lastOffsetInBuffer := logBuffer.offset - 1
|
||||
logBuffer.buf = logBuffer.prevBuffers.SealBuffer(logBuffer.startTime, logBuffer.stopTime, logBuffer.buf, logBuffer.pos, logBuffer.bufferStartOffset, lastOffsetInBuffer)
|
||||
// SealBuffer hands back the oldest window array to reuse. An entry larger
|
||||
// than BufferSize grew one of these arrays to fit it, and buffers cycle
|
||||
// forever, so without this a single oversized entry leaves every later
|
||||
// window carrying — and snapshotting — its size. Growth is on demand, so
|
||||
// the next oversized entry just reallocates.
|
||||
if len(logBuffer.buf) > BufferSize {
|
||||
logBuffer.buf = make([]byte, BufferSize)
|
||||
}
|
||||
// Hand a fully extended prefix snapshot to the sealed slot so sealed
|
||||
// readers reuse it instead of re-copying the window; reset for the next
|
||||
// window either way (holders keep their immutable prefix slices).
|
||||
|
||||
Reference in new issue
Block a user