lizhimins commented on code in PR #9048:
URL: https://github.com/apache/rocketmq/pull/9048#discussion_r1884966594
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java:
##########
@@ -296,6 +308,76 @@ private void appendAck(final AckMessageRequestHeader
requestHeader, final BatchA
}
}
+ private void appendAckNew(final AckMessageRequestHeader requestHeader,
final BatchAck batchAck,
+ final RemotingCommand response, final Channel channel, String
brokerName) throws RemotingCommandException {
+
+ if (requestHeader != null && batchAck == null) {
+ String[] extraInfo =
ExtraInfoUtil.split(requestHeader.getExtraInfo());
+ brokerName = ExtraInfoUtil.getBrokerName(extraInfo);
+ String groupId = requestHeader.getConsumerGroup();
+ String topicId = requestHeader.getTopic();
+ int queueId = requestHeader.getQueueId();
+ long startOffset = ExtraInfoUtil.getCkQueueOffset(extraInfo);
+ long ackOffset = requestHeader.getOffset();
+ long popTime = ExtraInfoUtil.getPopTime(extraInfo);
+ long invisibleTime = ExtraInfoUtil.getInvisibleTime(extraInfo);
+
+ int reviveQueueId = ExtraInfoUtil.getReviveQid(extraInfo);
+ if (reviveQueueId == KeyBuilder.POP_ORDER_REVIVE_QUEUE) {
+ ackOrderlyNew(topicId, groupId, queueId, ackOffset, popTime,
invisibleTime, channel, response);
+ } else {
+ this.brokerController.getPopConsumerService().ackAsync(
+ popTime, invisibleTime, groupId, topicId, queueId,
ackOffset);
+ }
+
+ this.brokerController.getBrokerStatsManager().incBrokerAckNums(1);
+
this.brokerController.getBrokerStatsManager().incGroupAckNums(groupId, topicId,
1);
+ } else {
+ String groupId = batchAck.getConsumerGroup();
+ String topicId = ExtraInfoUtil.getRealTopic(
+ batchAck.getTopic(), batchAck.getConsumerGroup(),
batchAck.getRetry());
+ int queueId = batchAck.getQueueId();
+ int reviveQueueId = batchAck.getReviveQueueId();
+ long startOffset = batchAck.getStartOffset();
+ long popTime = batchAck.getPopTime();
+ long invisibleTime = batchAck.getInvisibleTime();
+
+ try {
+ long minOffset =
this.brokerController.getMessageStore().getMinOffsetInQueue(topicId, queueId);
+ long maxOffset =
this.brokerController.getMessageStore().getMaxOffsetInQueue(topicId, queueId);
+ if (minOffset == -1 || maxOffset == -1) {
+ POP_LOGGER.error("Illegal topic or queue found when batch
ack {}", batchAck);
+ return;
+ }
+
+ int ackCount = 0;
+ // Maintain consistency with the old implementation code style
+ BitSet bitSet = batchAck.getBitSet();
+ for (int i = bitSet.nextSetBit(0); i >= 0; i =
bitSet.nextSetBit(i + 1)) {
+ if (i == Integer.MAX_VALUE) {
+ break;
+ }
+ long offset = startOffset + i;
+ if (offset < minOffset || offset > maxOffset) {
+ continue;
+ }
+ if (reviveQueueId == KeyBuilder.POP_ORDER_REVIVE_QUEUE) {
+ ackOrderlyNew(topicId, groupId, queueId, offset,
popTime, invisibleTime, channel, response);
+ } else {
+ this.brokerController.getPopConsumerService().ackAsync(
+ popTime, invisibleTime, groupId, topicId, queueId,
offset);
+ }
+ ackCount++;
+ }
+
+
this.brokerController.getBrokerStatsManager().incBrokerAckNums(ackCount);
+
this.brokerController.getBrokerStatsManager().incGroupAckNums(groupId, topicId,
ackCount);
+ } catch (ConsumeQueueException e) {
+ throw new RemotingCommandException("Failed to get max offset
in queue", e);
Review Comment:
Copied and pasted from the old method [doge]... I have already made the
modifications.
--
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: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
