Skip to content

[ISSUE #10980] Encode and decode POP checkpoint and ack messages without the intermediate JSON string #10981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wang-jiahua wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

[ISSUE #10980] Encode and decode POP checkpoint and ack messages without the intermediate JSON string #10981

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.nio.charset.StandardCharsets;
import java.util.BitSet;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
Expand Down Expand Up @@ -285,7 +284,7 @@ private void appendAck(final AckMessageRequestHeader requestHeader, final BatchA

MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setTopic(reviveTopic);
msgInner.setBody(JSON.toJSONString(ackMsg).getBytes(StandardCharsets.UTF_8));
msgInner.setBody(JSON.toJSONBytes(ackMsg));
msgInner.setQueueId(rqId);
if (ackMsg instanceof BatchAckMsg) {
msgInner.setTags(PopAckConstants.BATCH_ACK_TAG);
Expand Down
5 changes: 2 additions & 3 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
Expand Down Expand Up @@ -278,7 +277,7 @@ private CompletableFuture<Boolean> ackOrigin(final ChangeInvisibleTimeRequestHea
}

msgInner.setTopic(reviveTopic);
msgInner.setBody(JSON.toJSONString(ackMsg).getBytes(StandardCharsets.UTF_8));
msgInner.setBody(JSON.toJSONBytes(ackMsg));
msgInner.setQueueId(rqId);
msgInner.setTags(PopAckConstants.ACK_TAG);
msgInner.setBornTimestamp(System.currentTimeMillis());
Expand Down Expand Up @@ -322,7 +321,7 @@ private CompletableFuture<Boolean> appendCheckPointThenAckOrigin(
ck.setBrokerName(ExtraInfoUtil.getBrokerName(extraInfo));
ck.setSuspend(requestHeader.isSuspend());

msgInner.setBody(JSON.toJSONString(ck).getBytes(StandardCharsets.UTF_8));
msgInner.setBody(JSON.toJSONBytes(ck));
msgInner.setQueueId(reviveQid);
msgInner.setTags(PopAckConstants.CK_TAG);
msgInner.setBornTimestamp(System.currentTimeMillis());
Expand Down
4 changes: 2 additions & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -668,7 +668,7 @@ private void putAckToStore(final PopCheckPointWrapper pointWrapper, byte msgInde
ackMsg.setPopTime(point.getPopTime());
ackMsg.setBrokerName(point.getBrokerName());
msgInner.setTopic(popMessageProcessor.getReviveTopic());
msgInner.setBody(JSON.toJSONString(ackMsg).getBytes(DataConverter.CHARSET_UTF8));
msgInner.setBody(JSON.toJSONBytes(ackMsg));
msgInner.setQueueId(pointWrapper.getReviveQueueId());
msgInner.setTags(PopAckConstants.ACK_TAG);
msgInner.setBornTimestamp(System.currentTimeMillis());
Expand Down Expand Up @@ -724,7 +724,7 @@ private void putBatchAckToStore(final PopCheckPointWrapper pointWrapper, final L
batchAckMsg.setQueueId(point.getQueueId());
batchAckMsg.setPopTime(point.getPopTime());
msgInner.setTopic(popMessageProcessor.getReviveTopic());
msgInner.setBody(JSON.toJSONString(batchAckMsg).getBytes(DataConverter.CHARSET_UTF8));
msgInner.setBody(JSON.toJSONBytes(batchAckMsg));
msgInner.setQueueId(pointWrapper.getReviveQueueId());
msgInner.setTags(PopAckConstants.BATCH_ACK_TAG);
msgInner.setBornTimestamp(System.currentTimeMillis());
Expand Down
3 changes: 1 addition & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,6 @@
import org.apache.rocketmq.store.pop.PopCheckPoint;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -937,7 +936,7 @@ public MessageExtBrokerInner buildCkMsg(final PopCheckPoint ck, final int revive
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();

msgInner.setTopic(reviveTopic);
msgInner.setBody(JSON.toJSONString(ck).getBytes(StandardCharsets.UTF_8));
msgInner.setBody(JSON.toJSONBytes(ck));
msgInner.setQueueId(reviveQid);
msgInner.setTags(PopAckConstants.CK_TAG);
msgInner.setBornTimestamp(System.currentTimeMillis());
Expand Down
21 changes: 9 additions & 12 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,11 +385,10 @@ protected void consumeReviveMessage(ConsumeReviveObj consumeReviveObj) {
}
for (MessageExt messageExt : messageExts) {
if (PopAckConstants.CK_TAG.equals(messageExt.getTags())) {
String raw = new String(messageExt.getBody(), DataConverter.CHARSET_UTF8);
if (brokerController.getBrokerConfig().isEnablePopLog()) {
POP_LOGGER.info("reviveQueueId={},find ck, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), raw);
if (brokerController.getBrokerConfig().isEnablePopLog() && POP_LOGGER.isInfoEnabled()) {
POP_LOGGER.info("reviveQueueId={},find ck, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), new String(messageExt.getBody(), DataConverter.CHARSET_UTF8));
}
PopCheckPoint point = JSON.parseObject(raw, PopCheckPoint.class);
PopCheckPoint point = JSON.parseObject(messageExt.getBody(), PopCheckPoint.class);
if (point.getTopic() == null || point.getCId() == null) {
continue;
}
Expand All @@ -400,11 +399,10 @@ protected void consumeReviveMessage(ConsumeReviveObj consumeReviveObj) {
firstRt = point.getReviveTime();
}
} else if (PopAckConstants.ACK_TAG.equals(messageExt.getTags())) {
String raw = new String(messageExt.getBody(), StandardCharsets.UTF_8);
if (brokerController.getBrokerConfig().isEnablePopLog()) {
POP_LOGGER.info("reviveQueueId={}, find ack, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), raw);
if (brokerController.getBrokerConfig().isEnablePopLog() && POP_LOGGER.isInfoEnabled()) {
POP_LOGGER.info("reviveQueueId={}, find ack, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), new String(messageExt.getBody(), StandardCharsets.UTF_8));
Comment thread
wang-jiahua marked this conversation as resolved.
}
AckMsg ackMsg = JSON.parseObject(raw, AckMsg.class);
AckMsg ackMsg = JSON.parseObject(messageExt.getBody(), AckMsg.class);
brokerController.getBrokerMetricsManager().getPopMetricsManager().incPopReviveAckGetCount(ackMsg, queueId);
String brokerName = StringUtils.isNotBlank(ackMsg.getBrokerName()) ?
ackMsg.getBrokerName() : brokerController.getBrokerConfig().getBrokerName();
Expand All @@ -426,12 +424,11 @@ protected void consumeReviveMessage(ConsumeReviveObj consumeReviveObj) {
}
}
} else if (PopAckConstants.BATCH_ACK_TAG.equals(messageExt.getTags())) {
String raw = new String(messageExt.getBody(), StandardCharsets.UTF_8);
if (brokerController.getBrokerConfig().isEnablePopLog()) {
POP_LOGGER.info("reviveQueueId={}, find batch ack, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), raw);
if (brokerController.getBrokerConfig().isEnablePopLog() && POP_LOGGER.isInfoEnabled()) {
POP_LOGGER.info("reviveQueueId={}, find batch ack, offset:{}, raw : {}", messageExt.getQueueId(), messageExt.getQueueOffset(), new String(messageExt.getBody(), StandardCharsets.UTF_8));
Comment thread
wang-jiahua marked this conversation as resolved.
}

BatchAckMsg bAckMsg = JSON.parseObject(raw, BatchAckMsg.class);
BatchAckMsg bAckMsg = JSON.parseObject(messageExt.getBody(), BatchAckMsg.class);
brokerController.getBrokerMetricsManager().getPopMetricsManager().incPopReviveAckGetCount(bAckMsg, queueId);
String brokerName = StringUtils.isNotBlank(bAckMsg.getBrokerName()) ?
bAckMsg.getBrokerName() : brokerController.getBrokerConfig().getBrokerName();
Expand Down
21 changes: 21 additions & 0 deletions store/src/test/java/org/apache/rocketmq/store/pop/AckMsgTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.rocketmq.store.pop;

import com.alibaba.fastjson2.JSON;
import java.nio.charset.StandardCharsets;
import org.junit.Assert;
import org.junit.Test;

Expand Down Expand Up @@ -48,4 +49,24 @@ public void testSerializeAndDeSerialize() {
Assert.assertEquals(ackMsg1.getAckOffset(), ackMsg2.getAckOffset());
Assert.assertEquals(ackMsg1.getPopTime(), ackMsg2.getPopTime());
}

@Test
public void testToJsonBytesMatchesJsonStringBytes() {
AckMsg ackMsg = new AckMsg();
ackMsg.setBrokerName("broker-a");
ackMsg.setTopic("topic-\u4e2d\u6587");
ackMsg.setConsumerGroup("group");
ackMsg.setQueueId(3);
ackMsg.setStartOffset(200L);
ackMsg.setAckOffset(100L);
ackMsg.setPopTime(1670212915531L);

byte[] direct = JSON.toJSONBytes(ackMsg);
Assert.assertArrayEquals(JSON.toJSONString(ackMsg).getBytes(StandardCharsets.UTF_8), direct);

AckMsg decoded = JSON.parseObject(direct, AckMsg.class);
Assert.assertEquals(ackMsg.getTopic(), decoded.getTopic());
Assert.assertEquals(ackMsg.getAckOffset(), decoded.getAckOffset());
Assert.assertEquals(ackMsg.getPopTime(), decoded.getPopTime());
}
}
27 changes: 25 additions & 2 deletions store/src/test/java/org/apache/rocketmq/store/pop/BatchAckMsgTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@
package org.apache.rocketmq.store.pop;

import com.alibaba.fastjson2.JSON;
import org.junit.Assert;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

import org.junit.Assert;
import org.junit.Test;

public class BatchAckMsgTest {

@Test
Expand Down Expand Up @@ -54,4 +56,25 @@ public void testSerializeAndDeSerialize() {
Assert.assertEquals(batchAckMsg1.getStartOffset(), batchAckMsg2.getStartOffset());
Assert.assertEquals(batchAckMsg1.getPopTime(), batchAckMsg2.getPopTime());
}

@Test
public void testToJsonBytesMatchesJsonStringBytes() {
BatchAckMsg batchAckMsg = new BatchAckMsg();
List<Long> aol = new ArrayList<>(2);
aol.add(100L);
aol.add(101L);
batchAckMsg.setAckOffsetList(aol);
batchAckMsg.setStartOffset(200L);
batchAckMsg.setConsumerGroup("group");
batchAckMsg.setTopic("topic-\u4e2d\u6587");
batchAckMsg.setQueueId(3);
batchAckMsg.setPopTime(1679454922000L);

byte[] direct = JSON.toJSONBytes(batchAckMsg);
Assert.assertArrayEquals(JSON.toJSONString(batchAckMsg).getBytes(StandardCharsets.UTF_8), direct);

BatchAckMsg decoded = JSON.parseObject(direct, BatchAckMsg.class);
Assert.assertEquals(batchAckMsg.getAckOffsetList(), decoded.getAckOffsetList());
Assert.assertEquals(batchAckMsg.getTopic(), decoded.getTopic());
}
}
53 changes: 53 additions & 0 deletions store/src/test/java/org/apache/rocketmq/store/pop/PopCheckPointTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.store.pop;

import com.alibaba.fastjson2.JSON;
import java.nio.charset.StandardCharsets;
import org.junit.Assert;
import org.junit.Test;

public class PopCheckPointTest {

@Test
public void testToJsonBytesMatchesJsonStringBytes() {
PopCheckPoint ck = new PopCheckPoint();
ck.setTopic("topic-\u4e2d\u6587");
ck.setCId("group");
ck.setQueueId(3);
ck.setStartOffset(200L);
ck.setPopTime(1670212915531L);
ck.setInvisibleTime(60000L);
ck.setBitMap(5);
ck.setNum((byte) 2);
ck.setBrokerName("broker-a");
ck.addDiff(1);
ck.addDiff(3);
ck.setRePutTimes("1");

byte[] direct = JSON.toJSONBytes(ck);
Assert.assertArrayEquals(JSON.toJSONString(ck).getBytes(StandardCharsets.UTF_8), direct);

PopCheckPoint decoded = JSON.parseObject(direct, PopCheckPoint.class);
Assert.assertEquals(ck.getTopic(), decoded.getTopic());
Assert.assertEquals(ck.getCId(), decoded.getCId());
Assert.assertEquals(ck.getStartOffset(), decoded.getStartOffset());
Assert.assertEquals(ck.getPopTime(), decoded.getPopTime());
Assert.assertEquals(ck.getQueueOffsetDiff(), decoded.getQueueOffsetDiff());
Assert.assertEquals(ck.getBitMap(), decoded.getBitMap());
}
}
Loading