diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java index 65f5f79aec4..036a7040e41 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java @@ -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; @@ -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); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java index 02deeb18a7a..fce4a74bb63 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java @@ -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; @@ -278,7 +277,7 @@ private CompletableFuture 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()); @@ -322,7 +321,7 @@ private CompletableFuture 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()); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java index 5373eaea333..13869e958e7 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java @@ -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()); @@ -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()); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java index 55cabe6f5e5..779e6fef2de 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java @@ -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; @@ -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()); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java index 07f16e98965..67abcfa9071 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java @@ -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; } @@ -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)); } - 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(); @@ -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)); } - 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(); diff --git a/store/src/test/java/org/apache/rocketmq/store/pop/AckMsgTest.java b/store/src/test/java/org/apache/rocketmq/store/pop/AckMsgTest.java index 13df028e83d..14a147ea627 100644 --- a/store/src/test/java/org/apache/rocketmq/store/pop/AckMsgTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/pop/AckMsgTest.java @@ -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; @@ -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()); + } } \ No newline at end of file diff --git a/store/src/test/java/org/apache/rocketmq/store/pop/BatchAckMsgTest.java b/store/src/test/java/org/apache/rocketmq/store/pop/BatchAckMsgTest.java index 0a1bc714cfa..53b16b4dd1b 100644 --- a/store/src/test/java/org/apache/rocketmq/store/pop/BatchAckMsgTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/pop/BatchAckMsgTest.java @@ -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 @@ -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 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()); + } } diff --git a/store/src/test/java/org/apache/rocketmq/store/pop/PopCheckPointTest.java b/store/src/test/java/org/apache/rocketmq/store/pop/PopCheckPointTest.java new file mode 100644 index 00000000000..4b74a42b674 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/pop/PopCheckPointTest.java @@ -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()); + } +}