Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #4118 from ltamber/5.0.0-beta-ctopic
[RIP 30] Support Compaction topic
- Loading branch information
Showing
34 changed files
with
3,126 additions
and
330 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -13,3 +13,4 @@ devenv | |
| .DS_Store | ||
| localbin | ||
| nohup.out | ||
| docker/ | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
22 changes: 22 additions & 0 deletions
22
common/src/main/java/org/apache/rocketmq/common/attribute/DeletePolicy.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| /* | ||
| * 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.common.attribute; | ||
|
|
||
| public enum DeletePolicy { | ||
| NORMAL, | ||
| COMPACTION | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
50 changes: 50 additions & 0 deletions
50
common/src/main/java/org/apache/rocketmq/common/utils/DeletePolicyUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * 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.common.utils; | ||
|
|
||
| import org.apache.rocketmq.common.TopicAttributes; | ||
| import org.apache.rocketmq.common.TopicConfig; | ||
| import org.apache.rocketmq.common.attribute.DeletePolicy; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.Optional; | ||
|
|
||
| public class DeletePolicyUtils { | ||
| public static boolean isCompaction(Optional<TopicConfig> topicConfig) { | ||
| return Objects.equals(DeletePolicy.COMPACTION, getDeletePolicy(topicConfig)); | ||
| } | ||
|
|
||
| public static DeletePolicy getDeletePolicy(Optional<TopicConfig> topicConfig) { | ||
| if (!topicConfig.isPresent()) { | ||
| return DeletePolicy.valueOf(TopicAttributes.DELETE_POLICY_ATTRIBUTE.getDefaultValue()); | ||
| } | ||
|
|
||
| String attributeName = TopicAttributes.DELETE_POLICY_ATTRIBUTE.getName(); | ||
|
|
||
| Map<String, String> attributes = topicConfig.get().getAttributes(); | ||
| if (attributes == null || attributes.size() == 0) { | ||
| return DeletePolicy.valueOf(TopicAttributes.DELETE_POLICY_ATTRIBUTE.getDefaultValue()); | ||
| } | ||
|
|
||
| if (attributes.containsKey(attributeName)) { | ||
| return DeletePolicy.valueOf(attributes.get(attributeName)); | ||
| } else { | ||
| return DeletePolicy.valueOf(TopicAttributes.DELETE_POLICY_ATTRIBUTE.getDefaultValue()); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| # Compaction Topic | ||
|
|
||
| ## 使用方式 | ||
| ### 创建compaction topic | ||
| ```shell | ||
| $ bin/mqadmin updateTopic -w 8 -r 8 -a +delete.policy=COMPACTION -n localhost:9876 -t ctopic -c DefaultCluster | ||
| create topic to 127.0.0.1:10911 success. | ||
| TopicConfig [topicName=ctopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={+delete.policy=COMPACTION}] | ||
| ``` | ||
| ### 生产数据 | ||
| 与普通消息一样 | ||
| ```java | ||
| DefaultMQProducer producer = new DefaultMQProducer("CompactionTestGroup"); | ||
| producer.setNamesrvAddr("localhost:9876"); | ||
| producer.start(); | ||
|
|
||
| Message msg = new Message(topic, "tags", "keys", "bodys"getBytes(StandardCharsets.UTF_8)); | ||
| SendResult sendResult = producer.send(msg); | ||
|
|
||
| System.out.printf("%s%n", sendResult); | ||
| ``` | ||
| ### 消费数据 | ||
| 消费offset与compaction之前保持不变，如果指定offset消费，当指定的offset不存在时，返回后面最近的一条数据 | ||
| 在compaction场景下，大部分消费都是从0开始消费完整的数据 | ||
| ```java | ||
| DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("compactionTestGroup"); | ||
| consumer.setNamesrvAddr("localhost:9876"); | ||
| consumer.setPullThreadNums(4); | ||
| consumer.start(); | ||
|
|
||
| Collection<MessageQueue> messageQueueList = consumer.fetchMessageQueues("ctopic"); | ||
| consumer.assign(messageQueueList); | ||
| messageQueueList.forEach(mq -> { | ||
| try { | ||
| consumer.seekToBegin(mq); | ||
| } catch (MQClientException e) { | ||
| e.printStackTrace(); | ||
| } | ||
| }); | ||
|
|
||
| while (true) { | ||
| List<MessageExt> msgList = consumer.poll(1000); | ||
| if (msgList != null) { | ||
| msgList.forEach(msg -> System.out.println(msg.toString())); | ||
| } | ||
| } | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| # Compaction Topic | ||
|
|
||
| ## use example | ||
| ### create compaction topic | ||
| ```shell | ||
| $ bin/mqadmin updateTopic -w 8 -r 8 -a +delete.policy=COMPACTION -n localhost:9876 -t ctopic -c DefaultCluster | ||
| create topic to 127.0.0.1:10911 success. | ||
| TopicConfig [topicName=ctopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={+delete.policy=COMPACTION}] | ||
| ``` | ||
|
|
||
| ### produce message | ||
| the same with ordinary message | ||
| ```java | ||
| DefaultMQProducer producer = new DefaultMQProducer("CompactionTestGroup"); | ||
| producer.setNamesrvAddr("localhost:9876"); | ||
| producer.start(); | ||
|
|
||
| Message msg = new Message(topic, "tags", "keys", "bodys"getBytes(StandardCharsets.UTF_8)); | ||
| SendResult sendResult = producer.send(msg); | ||
|
|
||
| System.out.printf("%s%n", sendResult); | ||
| ``` | ||
| ### consume message | ||
| the message offset remains unchanged after compaction. If the consumer specified offset does not exist, return the most recent message after the offset. | ||
|
|
||
| In the compaction scenario, most consumption was started from the beginning of the queue. | ||
| ```java | ||
| DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("compactionTestGroup"); | ||
| consumer.setNamesrvAddr("localhost:9876"); | ||
| consumer.setPullThreadNums(4); | ||
| consumer.start(); | ||
|
|
||
| Collection<MessageQueue> messageQueueList = consumer.fetchMessageQueues("ctopic"); | ||
| consumer.assign(messageQueueList); | ||
| messageQueueList.forEach(mq -> { | ||
| try { | ||
| consumer.seekToBegin(mq); | ||
| } catch (MQClientException e) { | ||
| e.printStackTrace(); | ||
| } | ||
| }); | ||
|
|
||
| while (true) { | ||
| List<MessageExt> msgList = consumer.poll(1000); | ||
| if (msgList != null) { | ||
| msgList.forEach(msg -> System.out.println(msg.toString())); | ||
| } | ||
| } | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.