cleanItemListMoreThanSecondGen() validates, for every queue whose leader
moved, that the new leader agrees. It reads the new leader's hosted
queues with get(qId) and calls isEmpty() on the result unconditionally:
when the remote broker's mapping does not contain that queue id at all —
the transient state this service exists to reconcile — the read returns
null and the pass dies with a NullPointerException.
The exception is swallowed by the per-topic catch, so the visible damage
is silent: every remaining queue of that topic is skipped for this cycle,
the mapping is left half-cleaned, and the failure repeats every five
minutes.
Treat a missing remote queue as "not confirmed yet" and continue, which
is how the surrounding reads already handle absent data.
Red/green: remoteLeaderMissingTheQueueIdDoesNotAbortTheRestOfTheTopic
fails on the old code (the second queue is never removed because the NPE
aborts the pass) and passes with the null guard. Broker module: 815
tests, the 7 failures are pre-existing (RocksDB/filter/schedule/pop-lite/
BrokerOuterAPI) and reproduce identically with this change stashed.
BrokerContainer.initialize() schedules the name-server refresh and the five-second
brokerOuterAPI.refreshMetadata() task on scheduledExecutorService, and shutdown()
stops the brokers, both remoting servers, the request executor and the outer API -
but not that scheduler. A shut-down container therefore keeps refreshing metadata
through an API it has already closed, logging "ScheduledTask refresh metadata
exception" every five seconds, and leaks the scheduler thread on every start/shutdown
cycle. Stop it before the API is closed, the way the request executor next to it is
stopped.
CheckpointFile.read builds the IOException for a crc32 mismatch with a "file=%s"
label but passes expectedCrc32 into it, so an operator who opens a corrupted
checkpoint reads e.g. "file=823222925" - a checksum where the path that failed to
load is expected - and the expected value itself is never named.
Pass the path that was read and label the expected checksum, and pin it with the
case that corrupts both the checkpoint and the .bak fallback read() falls back to.
* [ISSUE #11392] Ignore AI agent default files
Add common AI coding assistant configuration files (.claude/, CLAUDE.md,
.cursor/, .cursorrules, .windsurf/, .windsurfrules, .aider*, .continue/,
.codeium/, .gemini/, GEMINI.md, AGENTS.md) to .gitignore so contributors
do not accidentally commit local AI agent artifacts.
* Also ignore .qoder/ default config files
Covers the same rule proposed in #11391, making this PR a superset and
avoiding a merge conflict on .gitignore between the two PRs.
* Also ignore Qoder, Cline, Roo Code, Augment, Amazon Q, Trae, Junie, Cody and Copilot default files
Wildcard Lite consumer groups consume with clustering semantics, but the
registration path reported them as LITE_SELECTIVE, so the admin/console view
showed an incorrect consumption mode.
Instead of adding a new MessageModel value, derive the Lite identity from the
client type carried on ProxyContext and normalize the reported model:
- ClientActivity#registerConsumer writes the real client type into the context
via ctx.setClientType, so downstream no longer infers Lite identity from the
display-only LITE_SELECTIVE value.
- ClientProcessor replaces validateLiteMode with validateLiteConsumer, which
reads ctx.isLiteConsumer() plus the real (remapped) SubscriptionGroupConfig to
enforce that Lite clients only register to Lite groups and vice versa, and
normalizes a wildcard Lite group's model to CLUSTERING before registration.
The overridden method signatures kept by the commercial subclass
(registerConsumer, validateLiteSubTopic) are unchanged.
GroupConsumeInfo.compareTo, TagCountBean.compareTo and
QueryMsgByUniqueKeySubCommand.queryById subtracted two long values and cast
the difference to int, so a difference beyond Integer.MAX_VALUE reversed the
order and broke the comparator contract that TimSort checks. Use
Long.compare / Comparator.comparingLong instead.
Signed-off-by: zhang hongyuan <100520587+Metastarx@users.noreply.github.com>
encodeBody used mark()/get()/reset() on the caller's buffer, which
restored the position but silently overwrote any mark the caller had set,
so a later reset() jumped to the wrong position. Read the remaining bytes
from a duplicate() instead, leaving the original buffer's position, limit
and mark untouched, and add a regression test.
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
* [ISSUE #11250] Fix Lite topic ack re-dispatching the event while holding the pop lock
- Decide the re-dispatch under the pop lock and dispatch after releasing it
- Add a unit test asserting the pop lock is free when the ack dispatches
* [ISSUE #11250] Re-trigger CI
* [ISSUE #11248] Fix Lite topic consumer losing messages when notified before they are readable
- Mark a lite topic in the per-request dedup set only after messages are read
- Retry a re-dispatch within the same pop after an empty read instead of dropping it
- Add unit tests for the empty-read path of popLiteTopic and popByClientId
* [ISSUE #11248] Re-trigger CI
* 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>
- Mark the dedup entry only after a lite topic is handled under the pop lock, not when it is FIFO-blocked, so a re-dispatched event is retried instead of dropped
- Move the processed-set marking into popLiteTopic and narrow its visibility
- Add a regression test reproducing the concurrent ack-unblock event loss
- Correct the off-by-one offset check in LiteEventDispatcher.dispatch (offset == 0 -> == 1) so onLmqCreate fires on the first arriving message
- Restore store-dependent lifecycle tests to the concrete impl test classes, driven by real putMessage through the notify path
- Add MessageArrivingListener overloads to LiteTestUtil to wire the real dispatch path in tests
Implement the control-plane admin capability as a gRPC service
served from the proxy process (NOT connecting to broker remoting/grpc
directly), per the rocketmq-apis admin.proto contract.
---------
Co-authored-by: lizhimins <707364882@qq.com>
Extract the user property validation logic (property count limit,
reserved-key check, control character check) from buildMessageProperty()
into a standalone protected checkUserProperties() method. The property
size accumulation stays in buildMessageProperty().
* [ISSUE #10929] Add LmqPrefixIndex to accelerate wildcard dispatch and refine lite subscription model
- Add LmqPrefixIndex, a prefix-ordered in-memory index over lmq names, bootstrapped at startup and maintained via onLmqCreate/onLmqDelete hooks
- Rewrite wildcard full dispatch, getLiteTopicCount, collectByParentTopic and cleanByParentTopic on index-backed forEachLiteTopicByPrefix/ByParent, turning O(total lmqs) scans into O(matched lmqs)
- Refine lite subscription model: replace SubscriberWrapper with Map-based getAllSubscribers, simplify getWildcardGroupClients, rename getLiteTopicSet to getLmqSet, make LiteCtlListener callbacks default methods
* [ISSUE #10929] Retrigger CI
Move the ASF license comment to the top of the file, before the
package statement. It was squeezed onto the same line as the package
declaration (introduced by b6ff649291), which breaks the conventional
header placement used across the codebase.
Wrap the topic route-size assertions in awaitility polling so the tests
wait for broker->NameServer route propagation instead of asserting
immediately. This removes the race condition that intermittently caused
'Expected size: 3 but was: 2' failures in the CI pipeline.
Applied to testCreateOrUpdateTopic_EnableSingleTopicRegistration,
testStaticTopicNotAffected and testCreateOrUpdateTopic_EnableSplitRegistration,
consistent with the existing awaitility pattern in the same class.
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>
Felix-Gong