yx9o commented on code in PR #11480:
URL: https://github.com/apache/rocketmq/pull/11480#discussion_r4407959337
##########
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyService.java:
##########
@@ -311,16 +452,178 @@ public int getShardingKeyIndex() {
return shardingKeyIndex;
}
+ private void mergeFrom(final ConsumeRequest other) {
+ PopConsumeBatch batch;
+ while ((batch = other.pendingBatches.poll()) != null) {
+ this.pendingBatches.add(batch);
+ }
+ }
+
+ private void releaseWaitAck(int count) {
+ if (count > 0) {
+ this.processQueue.decFoundMsg(-count);
+ }
+ }
+
+ private void releaseWaitAckForPendingBatches() {
+ PopConsumeBatch batch;
+ while ((batch = this.pendingBatches.poll()) != null) {
+ releaseWaitAck(batch.getMsgs().size());
+ }
+ }
+
+ /**
+ * Remove this exact request from the shared set. ConsumeRequest
equality is based on the
+ * queue/shard, so a plain remove could delete a successor that was
submitted after this
+ * request had already been unregistered. Callers must hold the
consume request lock.
+ */
+ private void removeFromConsumeRequestSet() {
+ for (ConsumeRequest request : consumeRequestSet) {
+ if (request == this) {
+ consumeRequestSet.remove(request);
+ return;
+ }
+ }
+ }
+
+ private void abortRemainingBatches() {
+ Object lock =
consumeRequestLock.fetchLockObject(this.messageQueue, this.shardingKeyIndex);
+ synchronized (lock) {
+ removeFromConsumeRequestSet();
+ releaseWaitAckForPendingBatches();
+ }
+ }
+
@Override
public void run() {
+ try {
+ for (;;) {
+ PopConsumeBatch batch = this.pendingBatches.poll();
+ if (batch == null) {
+ // Check for work that was merged in while the last
batch was being consumed,
+ // then remove the request atomically under the
submission lock. Without this
+ // lock a concurrently submitted request could be
appended after the empty
+ // check but before removal and would never be
consumed.
+ Object lock =
consumeRequestLock.fetchLockObject(this.messageQueue, this.shardingKeyIndex);
+ synchronized (lock) {
+ batch = this.pendingBatches.poll();
+ if (batch == null) {
+ removeFromConsumeRequestSet();
+ return;
+ }
+ }
+ }
+
+ if (!consumeBatch(batch)) {
+ abortRemainingBatches();
+ return;
+ }
+ }
+ } finally {
+ abortRemainingBatches();
+ }
+ }
+
+ private boolean consumeBatch(final PopConsumeBatch batch) {
+ final List<MessageExt> msgs = batch.getMsgs();
if (this.processQueue.isDropped()) {
log.warn("run, message queue not be able to consume, because
it's dropped. {}", this.messageQueue);
-
ConsumeMessagePopOrderlyService.this.removeConsumeRequest(this);
- return;
+ releaseWaitAck(msgs.size());
+ return false;
+ }
+
+ if (batch.isPopTimeout()) {
+ log.info("the pop message time out so abort consume.
popTime={} invisibleTime={}, group={} {}",
+ batch.getPopTime(), batch.getInvisibleTime(),
consumerGroup, this.messageQueue);
+ releaseWaitAck(msgs.size());
+ return false;
}
- // lock on sharding key index
- final Object objLock =
messageQueueLock.fetchLockObject(this.messageQueue, shardingKeyIndex);
+ // Ordered consumption must serialize all batches of the same
queue/shard. The sharding
+ // key index stays 0 because POP consumes the queue as a single
ordered lane.
+ final Object objLock =
messageQueueLock.fetchLockObject(this.messageQueue, this.shardingKeyIndex);
+ synchronized (objLock) {
+ if (this.processQueue.isDropped()) {
+ log.warn("message queue not be able to consume, because
it's dropped. {}", this.messageQueue);
+ releaseWaitAck(msgs.size());
+ return false;
+ }
+
+ if (batch.isPopTimeout()) {
+ log.info("the pop message time out before consume.
popTime={} invisibleTime={}, group={} {}",
+ batch.getPopTime(), batch.getInvisibleTime(),
consumerGroup, this.messageQueue);
+ releaseWaitAck(msgs.size());
+ return false;
+ }
+
+ defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs,
consumerGroup);
+
+ ConsumeOrderlyContext context = new
ConsumeOrderlyContext(this.messageQueue);
+ ConsumeMessageContext consumeMessageContext = null;
+ if (defaultMQPushConsumerImpl.hasHook()) {
+ consumeMessageContext = new ConsumeMessageContext();
+
consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
+
consumeMessageContext.setConsumerGroup(defaultMQPushConsumer.getConsumerGroup());
+ consumeMessageContext.setProps(new HashMap<>());
+ consumeMessageContext.setMq(this.messageQueue);
+ consumeMessageContext.setMsgList(msgs);
+ consumeMessageContext.setSuccess(false);
+
defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
+ }
+
+ long beginTimestamp = System.currentTimeMillis();
+ boolean hasException = false;
+ ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
+ ConsumeOrderlyStatus status = null;
+ try {
+ for (MessageExt msg : msgs) {
+ MessageAccessor.setConsumeStartTimeStamp(msg,
String.valueOf(System.currentTimeMillis()));
+ }
+ status =
messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
Review Comment:
Could we split the POP result by `consumeMessageBatchMaxSize` and consume
the sub-batches in order? With the default batch size of 1, a listener may
process only `msgs.get(0)`, but returning `SUCCESS` here acknowledges the
entire POP result.
--
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]
