* fix(timer): persist TimelineRollService checkpoint to avoid repeated rolls
Scan hour-sized windows from a RocksDB checkpoint instead of sleeping a fixed interval, so delayed scans do not skip or re-roll the same timer messages.
Co-authored-by: Cursor <cursoragent@cursor.com>
* Persist the roll checkpoint only after the messages are reput
TimelineRollService used to write the checkpoint right after a scan, so a crash
between the scan and the reput dropped the whole window. The reput service now
owns the write and takes the key to write, which lets the roll queue persist its
own checkpoint the same way the expired queue already does. Roll keeps its scan
frontier in memory and advances it per window, as TimelineForwardService does.
Without a persisted roll checkpoint the frontier is derived from the delivery
checkpoint plus one maximum delay, so a broker that ran without this key
re-scans its windows instead of skipping them.
Co-authored-by: Cursor <cursoragent@cursor.com>
* Drop timerRocksDBRollIntervalHours and back off on roll errors
A checkpoint driven loop advances one window per window width, so the width and
the cadence are the same quantity and only one config can define it. Keep
timerRocksDBRollRangeHours as the scan window width and remove
timerRocksDBRollIntervalHours, which no longer has a role of its own.
The outer catch used to log and retry immediately, spinning the thread while
rocksdb keeps throwing.
Co-authored-by: Cursor <cursoragent@cursor.com>
* fix
* Retrigger CI
Co-authored-by: Cursor <cursoragent@cursor.com>
* Honor timerEnableRetryUntilSuccess in the RocksDB timer reput path
Reuse the file-wheel flag so a recoverable put keeps retrying instead of
advancing the roll checkpoint after a fixed number of failures.
Co-authored-by: Cursor <cursoragent@cursor.com>
* Retrigger CI
Co-authored-by: Cursor <cursoragent@cursor.com>
---------
Co-authored-by: hqbfzwang <hqbfzwang@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
ConsumeQueue.putMessagePositionInfo() accesses MappedFile without
holding a reference count. A concurrent cleanExpiredConsumeQueue
can destroy() the MappedFile (unmapping its buffer) while the
writer is still using it, causing SIGSEGV in JIT-compiled
Unsafe.copyMemory.
Fix: Call mappedFile.hold() before accessing the buffer and
release() in a finally block, following the same pattern used
in DefaultMappedFile.getData(). Also protect the fillPreBlank
call in initializeWithOffset.
Reported crash signatures:
- Case 1 (v4.9.4): SEGV_MAPERR in ReputMessageService thread
at ConsumeQueue.putMessagePositionInfo
- Case 2 (v5.3.3): SEGV in AdminBrokerThread at
ConsumeQueueStore.cleanExpired
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
startIndex < minOffset was previously accepted as long as it stayed
within [0, maxOffset), returning a non-null LargeRocksDBConsumeQueueIterator
whose next() can yield a null CqUnit once the underlying data has been
purged. Callers like ScheduleMessageService rely on iterateFrom returning
null to detect and correct an out-of-range offset (matching ConsumeQueue's
getMinLogicOffset() check); without it they NPE dereferencing the null
CqUnit instead.
Add the same startIndex >= getMinOffsetInQueue() bound used by the
file-based ConsumeQueue to both iterateFrom overloads.
Co-authored-by: maowei.ymw <maowei.ymw@alibaba-inc.com>
* [ISSUE #10639] Reuse a per-thread scratch buffer in CommitLog.checkMessageAndReturnSize
* [ISSUE #10639] Reject corrupt negative totalSize before allocation in checkMessageAndReturnSize
* [ISSUE #10639] Add configurable reuse-buffer cap and unit tests for checkMessageAndReturnSize
* [ISSUE #10639] Default the check-message reuse-buffer cap to 1 MiB
---------
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
* When IndexRocksDBEnable or TransRocksDBEnable are enabled, we need to take these two offsets into account to accelerate recovery.
* Add UTs
* Refactor the code based on the review comments
* Revert "[ISSUE #8127]Optimize the metric calculation logic of the time wheel"
* Remove useless import
* Refactor Code
* Refactor Code
* Refactor Code
* Refactor Code
* Refactor Code
* Implement accelerated recovery for the file-based ConsumeQueue.
* Implement accelerated recovery for the file-based ConsumeQueue.
Change-Id: Ieac45d0582f2f83d977aeb8e6f5084268b7f8752
* Implement accelerated recovery for the file-based ConsumeQueue.
* Ignore testTruncateCQ UT
---------
Co-authored-by: RongtongJin <user@example.com>
* Fix StoreCheckpoint: set logicsMsgTimestamp only after CQ flush, use logicsMsgTempTimestamp for in-memory storetime
Change-Id: I6085bf6efaef84168ece31d080481717465f2b13
* rename to tmpLogicsMsgTimestamp
Change-Id: Ia65ca06751f765bdc2bf053c58e08789f4b2fb22
---------
Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
When the error 'failed to check message CRC, not found CRC in properties' occurs,
now it will print the full message content including topic, properties map,
properties length, and full message hex for debugging purposes.
Change-Id: I26d977e13802c5e69d01f16512e40d121dcff354
Co-developed-by: Cursor <noreply@cursor.com>
Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
bai mengxuan