leizhiyuan commented on code in PR #9048:
URL: https://github.com/apache/rocketmq/pull/9048#discussion_r1881504478
##########
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);
Review Comment:
long startOffset = ExtraInfoUtil.getCkQueueOffset(extraInfo);
brokerName = ExtraInfoUtil.getBrokerName(extraInfo);
seems no used by any place
##########
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:
This exception message should not be confirmed to be get max offset, right?
--
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
