processResponseCommand used get+remove (non-atomic) to retrieve ResponseFuture
from responseTable. scanResponseTable used iterator.remove(). If the response
arrived between scanResponseTable's remove and processResponseCommand's get,
the response would be logged as 'not matched any request' and silently dropped.
The callback would fire with timeout instead of success, even though the broker
had successfully processed the request.
Fix: Use ConcurrentHashMap.remove(key) in both methods. This is atomic:
either processResponseCommand gets the future (and executes success callback),
or scanResponseTable gets it (and executes timeout callback), but never both
and never neither.
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
* chore: bump DLedger to 0.3.3.4 (maintenance line)
Update the DLedger dependency from 0.3.2 to 0.3.3.4, the latest
release on the 0.3.x maintenance line published on Maven Central.
The 0.3.3.4 release is API-compatible with 0.3.2 and only carries
dependency hygiene fixes (aligning fastjson2 and excluding fastjson1
from its rocketmq-remoting transitive path).
Also update the Bazel WORKSPACE artifact coordinate to keep the two
build systems in sync.
* chore: fully remove fastjson1 from the build
DLedger 0.3.3.4 already excludes com.alibaba:fastjson from its
rocketmq-remoting dependency and its bytecode only references
fastjson2, but rocketmq still declared fastjson1 itself, so it kept
leaking onto every module compile classpath through rocketmq-common.
- drop the com.alibaba:fastjson property and dependencyManagement
entry from the root pom
- drop the unused com.alibaba:fastjson dependency from rocketmq-common
(no main source imports com.alibaba.fastjson.*)
- drop the fastjson1 pin from WORKSPACE and the fastjson1 dep from the
remoting test target in remoting/BUILD.bazel
- migrate RemotingSerializableCompatTest to fastjson2: use the
fastjson2 JSONField annotation (which is what the protocol classes
actually carry) and round trip through RemotingSerializable instead
of com.alibaba.fastjson.JSON
fastjson1 wire-format coverage is retained by the frozen fastjson1
payload already asserted in testCompatibilityCheckWithBitSet.
* fix(bazel): exclude rocketmq-remoting from coursier resolution
dledger 0.3.3.4 depends on rocketmq-remoting:5.5.0, which Coursier
fetches from Maven Central. That artifact transitively brings
grpc-netty-shaded:1.53.0 with a strict [1.53.0] range on grpc-core,
conflicting with the WORKSPACE-declared grpc 1.47.0 artifacts.
Since rocketmq-remoting is built from source in this repo (//remoting),
exclude it from external resolution — matching what PR #10947 already
does for the DLedger 0.4.x line.
---------
Co-authored-by: 通融 <rongtong.jrt@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>
- Refactor filter data index from topic-based to consumerGroup-based (SubscriptionFilterHandler)
- Add topic existence check before registering filter
- Generate BloomFilterData only when enableCalcFilterBitMap is enabled
- Fix thread safety: use ConcurrentHashMap for topicSqlFilterData
- Fix TOCTOU race conditions: replace containsKey+get with single get
- Rebuild subscriptionFilterData from filterDataByTopic in decode
- Add test for subscription shrink marking removed topics as dead
* [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>
- Broker: add PopLiteLongPollingService to NotificationProcessor for lite consumer notification polling
- Broker: LiteEventDispatcher notifies both PopLiteMessageProcessor and NotificationProcessor polling services
- Broker: add hasEvents(clientId) API to LiteEventDispatcher for message availability check
- Proxy: unify lite/normal pop paths into single popMessage call, route by ProxyContext.isLiteConsumer()
- Proxy: register LITE_SIMPLE_CONSUMER in ClientActivity and GrpcClientSettingsManager lifecycle
- Proxy: pass liteTopic property only for lite consumers in filter and response writer
- Proxy: remove standalone popLiteMessage from MessagingProcessor/ConsumerProcessor
- Remoting: add isLiteConsumer and clientId fields to NotificationRequestHeader
- Remove cached consumerOffsetManager field from LiteEventDispatcher
- Replace all usages with brokerController.getConsumerOffsetManager() for lazy resolution
- Allows downstream projects to swap ConsumerOffsetManager after broker init
- DefaultMQProducerImpl#request(Message, RequestCallback, long): drop the
executeRequestCallback() call in the async send onSuccess so a send success
no longer delivers a premature onSuccess(null); align with the other async
request overloads which only set sendRequestOk here.
- RequestResponseFuture: add an AtomicBoolean executeCallbackOnlyOnce guard so
the callback fires at most once even if the reply and timeout paths race.
- RequestFutureHolder#scanExpiredRequest: use ConcurrentHashMap.remove(key) to
atomically claim ownership instead of iterator.remove(); also fix the log
placeholder concatenation.
- ClientRemotingProcessor#processReplyMessage: use atomic remove(correlationId)
and route the reply through executeRequestCallback so the single-shot guard
covers the reply-success path too.
- Add RequestResponseFutureTest cases for success-then-timeout and concurrent
single-callback semantics.
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
DefaultElectPolicy sorted broker candidates by subtracting maxOffset values and casting the long delta to int. Large offset gaps can overflow the comparator result and rank a lower-offset broker first. This replaces subtraction with safe comparator helpers and adds a focused overflow regression test.
Constraint: Preserve existing election order: higher epoch, higher maxOffset, lower electionPriority
Rejected: Keep subtraction comparator | unsafe for long offset deltas greater than Integer.MAX_VALUE
Confidence: high
Scope-risk: narrow
Tested: mvn -q -pl controller -DskipTests=false -Dtest=DefaultElectPolicyTest -Djacoco.skip=true test
Tested: mvn -q -pl controller -DskipTests compile -Dspotbugs.skip=true -Dcheckstyle.skip=true
Tested: mvn -q -DskipTests compile -Dspotbugs.skip=true -Dcheckstyle.skip=true
Not-tested: Full controller suite on local JDK due existing JaCoCo/Hessian module-access failures
Related: #10578
- Replace consumerService.start() with consumerService.getPopConsumerStore().start()
- Replace consumerService.shutdown() with consumerService.getPopConsumerStore().shutdown()
- Avoid starting revive/cache background threads that are not needed for transferToFsStore verification
- Reduces test execution time from ~60s to ~1.3s
* [ISSUE #10560] Remove enableLiteEventMode config switch
- Remove enableLiteEventMode field, getter and setter from BrokerConfig
- Remove 5 early-return guard checks in LiteEventDispatcher
- Remove dead condition in PopLiteMessageProcessor.popLiteTopic
- Simplify getEventIterator to always use event-set path
- Delete unused LiteSubscriptionIterator inner class
- Remove disabled-mode test cases and stale Javadoc references
* chore: retrigger CI
Normalize horizontal rules, blank lines around headings and code
blocks, list markers, nested list indentation, and trailing whitespace
without changing content.
* [ISSUE #10549] Fix lite topic reset offset: memory leak, FIFO block bypass, and offset-0 reset failure
- Fix memory leak in removeResetOffset: clean up empty inner map entries from resetOffsetTable
- Add eraseResetOffset for precise cleanup on lite topic removal
- Skip FIFO block check in isFifoBlocked when server-side reset offset is pending
- Fix ResetOffsetByTimeCommand: change resetOffset > 0 to >= 0 to allow resetting to offset 0
- Add unit tests for eraseResetOffset and isFifoBlocked reset bypass
* chore: empty commit to trigger CI pipeline
BrokerMetricsManager.getMessageType(SendMessageRequestHeader) is called
once per send to classify the message. It internally decodes the
properties String into a HashMap, but the typical caller
(SendMessageProcessor) has already decoded the same String moments
before. The result is a redundant decode allocation per send (one
HashMap + ~14 String substrings + one Node[]).
This commit adds a public overload getMessageType(Map<String, String>)
that lets callers pass an already-decoded Map and reuse it. The
existing SendMessageRequestHeader overload now delegates to the new
overload; behavior is unchanged for callers that don't have a decoded
Map. Downstream callers (e.g. SendMessageProcessor) can switch to the
new overload in a separate broker-layer commit.
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
Jiahua Wang