Skip to content

Commit bbe87f5

Browse files
CLFutureXCLFutureX
authored
Merge branch 'apache:develop' into develop
2 parents a7b82b6 + c6d3f26 commit bbe87f5

File tree

71 files changed

+3312
-739
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

71 files changed

+3312
-739
lines changed

‎.github/workflows/pr-e2e-test.yml

+5-2
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,11 @@ env:
1313

1414
jobs:
1515
docker:
16-
runs-on: ubuntu-latest
1716
if: >
17+
github.repository == 'apache/rocketmq' &&
1818
github.event.workflow_run.event == 'pull_request' &&
1919
github.event.workflow_run.conclusion == 'success'
20+
runs-on: ubuntu-latest
2021
timeout-minutes: 30
2122
strategy:
2223
matrix:
@@ -74,7 +75,9 @@ jobs:
7475
path: rocketmq-docker/image-build-ci/versionlist/*
7576

7677
list-version:
77-
if: always()
78+
if: >
79+
github.repository == 'apache/rocketmq' &&
80+
always()
7881
name: List version
7982
needs: [docker]
8083
runs-on: ubuntu-latest

‎.github/workflows/push-ci.yml

+4-1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ env:
1515

1616
jobs:
1717
dist-tar:
18+
if: github.repository == 'apache/rocketmq'
1819
name: Build dist tar
1920
runs-on: ubuntu-latest
2021
timeout-minutes: 30
@@ -79,7 +80,9 @@ jobs:
7980

8081

8182
list-version:
82-
if: always()
83+
if: >
84+
github.repository == 'apache/rocketmq' &&
85+
always()
8386
name: List version
8487
needs: [docker]
8588
runs-on: ubuntu-latest

‎.github/workflows/snapshot-automation.yml

+1
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ env:
3636

3737
jobs:
3838
dist-tar:
39+
if: github.repository == 'apache/rocketmq'
3940
name: Build dist tar
4041
runs-on: ubuntu-latest
4142
timeout-minutes: 30

‎acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java

+4-1
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import java.util.SortedMap;
4141
import java.util.TreeMap;
4242
import org.apache.commons.codec.DecoderException;
43+
import org.apache.commons.lang3.StringUtils;
4344
import org.apache.commons.lang3.builder.ToStringBuilder;
4445
import org.apache.rocketmq.acl.AccessResource;
4546
import org.apache.rocketmq.acl.common.AclException;
@@ -268,7 +269,9 @@ public static PlainAccessResource parse(GeneratedMessageV3 messageV3, Authentica
268269
}
269270
} else if (NotifyClientTerminationRequest.getDescriptor().getFullName().equals(rpcFullName)) {
270271
NotifyClientTerminationRequest request = (NotifyClientTerminationRequest) messageV3;
271-
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
272+
if (StringUtils.isNotBlank(request.getGroup().getName())) {
273+
accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
274+
}
272275
} else if (QueryRouteRequest.getDescriptor().getFullName().equals(rpcFullName)) {
273276
QueryRouteRequest request = (QueryRouteRequest) messageV3;
274277
accessResource.addResourceAndPerm(request.getTopic(), Permission.ANY);

‎auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -129,7 +129,9 @@ public List<DefaultAuthorizationContext> build(Metadata metadata, GeneratedMessa
129129
}
130130
if (message instanceof NotifyClientTerminationRequest) {
131131
NotifyClientTerminationRequest request = (NotifyClientTerminationRequest) message;
132-
result = newGroupSubContexts(metadata, request.getGroup());
132+
if (StringUtils.isNotBlank(request.getGroup().getName())) {
133+
result = newGroupSubContexts(metadata, request.getGroup());
134+
}
133135
}
134136
if (message instanceof ChangeInvisibleDurationRequest) {
135137
ChangeInvisibleDurationRequest request = (ChangeInvisibleDurationRequest) message;

‎broker/BUILD.bazel

+2
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ java_library(
2929
"//remoting",
3030
"//srvutil",
3131
"//store",
32+
"//tieredstore",
3233
"@maven//:ch_qos_logback_logback_classic",
3334
"@maven//:com_alibaba_fastjson",
3435
"@maven//:com_alibaba_fastjson2_fastjson2",
@@ -81,6 +82,7 @@ java_library(
8182
"//filter",
8283
"//remoting",
8384
"//store",
85+
"//tieredstore",
8486
"@maven//:com_alibaba_fastjson",
8587
"@maven//:com_alibaba_fastjson2_fastjson2",
8688
"@maven//:com_google_guava_guava",

‎broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java

+5-2
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
import java.util.concurrent.ScheduledExecutorService;
3131
import java.util.concurrent.ScheduledFuture;
3232
import java.util.concurrent.TimeUnit;
33-
33+
import org.apache.commons.collections.CollectionUtils;
3434
import org.apache.commons.lang3.StringUtils;
3535
import org.apache.rocketmq.broker.BrokerController;
3636
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
@@ -803,7 +803,10 @@ private void scanAvailableControllerAddresses() {
803803

804804
private void updateControllerAddr() {
805805
if (brokerConfig.isFetchControllerAddrByDnsLookup()) {
806-
this.controllerAddresses = brokerOuterAPI.dnsLookupAddressByDomain(this.brokerConfig.getControllerAddr());
806+
List<String> adders = brokerOuterAPI.dnsLookupAddressByDomain(this.brokerConfig.getControllerAddr());
807+
if (CollectionUtils.isNotEmpty(adders)) {
808+
this.controllerAddresses = adders;
809+
}
807810
} else {
808811
final String controllerPaths = this.brokerConfig.getControllerAddr();
809812
final String[] controllers = controllerPaths.split(";");

‎broker/src/main/java/org/apache/rocketmq/broker/latency/BrokerFastFailure.java

+3
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,9 @@ private void cleanExpiredRequest() {
115115

116116
cleanExpiredRequestInQueue(this.brokerController.getAckThreadPoolQueue(),
117117
brokerController.getBrokerConfig().getWaitTimeMillsInAckQueue());
118+
119+
cleanExpiredRequestInQueue(this.brokerController.getAdminBrokerThreadPoolQueue(),
120+
brokerController.getBrokerConfig().getWaitTimeMillsInAdminBrokerQueue());
118121
}
119122

120123
void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) {

‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/NotifyMessageArrivingListener.java

+7-4
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,12 @@ public NotifyMessageArrivingListener(final PullRequestHoldService pullRequestHol
3636
@Override
3737
public void arriving(String topic, int queueId, long logicOffset, long tagsCode,
3838
long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
39-
this.pullRequestHoldService.notifyMessageArriving(topic, queueId, logicOffset, tagsCode,
40-
msgStoreTime, filterBitMap, properties);
41-
this.popMessageProcessor.notifyMessageArriving(topic, queueId);
42-
this.notificationProcessor.notifyMessageArriving(topic, queueId);
39+
40+
this.pullRequestHoldService.notifyMessageArriving(
41+
topic, queueId, logicOffset, tagsCode, msgStoreTime, filterBitMap, properties);
42+
this.popMessageProcessor.notifyMessageArriving(
43+
topic, queueId, tagsCode, msgStoreTime, filterBitMap, properties);
44+
this.notificationProcessor.notifyMessageArriving(
45+
topic, queueId, tagsCode, msgStoreTime, filterBitMap, properties);
4346
}
4447
}

‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java

+37-7
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,9 @@
3535
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
3636
import org.apache.rocketmq.remoting.netty.RequestTask;
3737
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
38+
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
39+
import org.apache.rocketmq.store.ConsumeQueueExt;
40+
import org.apache.rocketmq.store.MessageFilter;
3841

3942
import static org.apache.rocketmq.broker.longpolling.PollingResult.NOT_POLLING;
4043
import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_FULL;
@@ -147,39 +150,61 @@ public void run() {
147150
}
148151

149152
public void notifyMessageArrivingWithRetryTopic(final String topic, final int queueId) {
153+
this.notifyMessageArrivingWithRetryTopic(topic, queueId, null, 0L, null, null);
154+
}
155+
156+
public void notifyMessageArrivingWithRetryTopic(final String topic, final int queueId,
157+
Long tagsCode, long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
150158
String notifyTopic;
151159
if (KeyBuilder.isPopRetryTopicV2(topic)) {
152160
notifyTopic = KeyBuilder.parseNormalTopic(topic);
153161
} else {
154162
notifyTopic = topic;
155163
}
156-
notifyMessageArriving(notifyTopic, queueId);
164+
notifyMessageArriving(notifyTopic, queueId, tagsCode, msgStoreTime, filterBitMap, properties);
157165
}
158166

159-
public void notifyMessageArriving(final String topic, final int queueId) {
167+
public void notifyMessageArriving(final String topic, final int queueId,
168+
Long tagsCode, long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
160169
ConcurrentHashMap<String, Byte> cids = topicCidMap.get(topic);
161170
if (cids == null) {
162171
return;
163172
}
164173
for (Map.Entry<String, Byte> cid : cids.entrySet()) {
165174
if (queueId >= 0) {
166-
notifyMessageArriving(topic, cid.getKey(), -1);
175+
notifyMessageArriving(topic, -1, cid.getKey(), tagsCode, msgStoreTime, filterBitMap, properties);
167176
}
168-
notifyMessageArriving(topic, cid.getKey(), queueId);
177+
notifyMessageArriving(topic, queueId, cid.getKey(), tagsCode, msgStoreTime, filterBitMap, properties);
169178
}
170179
}
171180

172-
public boolean notifyMessageArriving(final String topic, final String cid, final int queueId) {
181+
public boolean notifyMessageArriving(final String topic, final int queueId, final String cid,
182+
Long tagsCode, long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
173183
ConcurrentSkipListSet<PopRequest> remotingCommands = pollingMap.get(KeyBuilder.buildPollingKey(topic, cid, queueId));
174184
if (remotingCommands == null || remotingCommands.isEmpty()) {
175185
return false;
176186
}
187+
177188
PopRequest popRequest = pollRemotingCommands(remotingCommands);
178189
if (popRequest == null) {
179190
return false;
180191
}
192+
193+
if (popRequest.getMessageFilter() != null && popRequest.getSubscriptionData() != null) {
194+
boolean match = popRequest.getMessageFilter().isMatchedByConsumeQueue(tagsCode,
195+
new ConsumeQueueExt.CqExtUnit(tagsCode, msgStoreTime, filterBitMap));
196+
if (match && properties != null) {
197+
match = popRequest.getMessageFilter().isMatchedByCommitLog(null, properties);
198+
}
199+
if (!match) {
200+
remotingCommands.add(popRequest);
201+
totalPollingNum.incrementAndGet();
202+
return false;
203+
}
204+
}
205+
181206
if (brokerController.getBrokerConfig().isEnablePopLog()) {
182-
POP_LOGGER.info("lock release , new msg arrive , wakeUp : {}", popRequest);
207+
POP_LOGGER.info("lock release, new msg arrive, wakeUp: {}", popRequest);
183208
}
184209
return wakeUp(popRequest);
185210
}
@@ -221,6 +246,11 @@ public boolean wakeUp(final PopRequest request) {
221246
*/
222247
public PollingResult polling(final ChannelHandlerContext ctx, RemotingCommand remotingCommand,
223248
final PollingHeader requestHeader) {
249+
return this.polling(ctx, remotingCommand, requestHeader, null, null);
250+
}
251+
252+
public PollingResult polling(final ChannelHandlerContext ctx, RemotingCommand remotingCommand,
253+
final PollingHeader requestHeader, SubscriptionData subscriptionData, MessageFilter messageFilter) {
224254
if (requestHeader.getPollTime() <= 0 || this.isStopped()) {
225255
return NOT_POLLING;
226256
}
@@ -234,7 +264,7 @@ public PollingResult polling(final ChannelHandlerContext ctx, RemotingCommand re
234264
}
235265
cids.putIfAbsent(requestHeader.getConsumerGroup(), Byte.MIN_VALUE);
236266
long expired = requestHeader.getBornTime() + requestHeader.getPollTime();
237-
final PopRequest request = new PopRequest(remotingCommand, ctx, expired);
267+
final PopRequest request = new PopRequest(remotingCommand, ctx, expired, subscriptionData, messageFilter);
238268
boolean isFull = totalPollingNum.get() >= this.brokerController.getBrokerConfig().getMaxPopPollingSize();
239269
if (isFull) {
240270
POP_LOGGER.info("polling {}, result POLLING_FULL, total:{}", remotingCommand, totalPollingNum.get());

‎broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopRequest.java

+20-5
Original file line numberDiff line numberDiff line change
@@ -16,28 +16,35 @@
1616
*/
1717
package org.apache.rocketmq.broker.longpolling;
1818

19+
import io.netty.channel.Channel;
1920
import io.netty.channel.ChannelHandlerContext;
2021
import java.util.Comparator;
2122
import java.util.concurrent.atomic.AtomicBoolean;
2223
import java.util.concurrent.atomic.AtomicLong;
23-
2424
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
25-
26-
import io.netty.channel.Channel;
25+
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
26+
import org.apache.rocketmq.store.MessageFilter;
2727

2828
public class PopRequest {
2929
private static final AtomicLong COUNTER = new AtomicLong(Long.MIN_VALUE);
3030

3131
private final RemotingCommand remotingCommand;
3232
private final ChannelHandlerContext ctx;
33-
private final long expired;
3433
private final AtomicBoolean complete = new AtomicBoolean(false);
3534
private final long op = COUNTER.getAndIncrement();
3635

37-
public PopRequest(RemotingCommand remotingCommand, ChannelHandlerContext ctx, long expired) {
36+
private final long expired;
37+
private final SubscriptionData subscriptionData;
38+
private final MessageFilter messageFilter;
39+
40+
public PopRequest(RemotingCommand remotingCommand, ChannelHandlerContext ctx,
41+
long expired, SubscriptionData subscriptionData, MessageFilter messageFilter) {
42+
3843
this.ctx = ctx;
3944
this.remotingCommand = remotingCommand;
4045
this.expired = expired;
46+
this.subscriptionData = subscriptionData;
47+
this.messageFilter = messageFilter;
4148
}
4249

4350
public Channel getChannel() {
@@ -64,6 +71,14 @@ public long getExpired() {
6471
return expired;
6572
}
6673

74+
public SubscriptionData getSubscriptionData() {
75+
return subscriptionData;
76+
}
77+
78+
public MessageFilter getMessageFilter() {
79+
return messageFilter;
80+
}
81+
6782
@Override
6883
public String toString() {
6984
final StringBuilder sb = new StringBuilder("PopRequest{");

‎broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsConstant.java

+5
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,16 @@ public class BrokerMetricsConstant {
2121

2222
public static final String GAUGE_PROCESSOR_WATERMARK = "rocketmq_processor_watermark";
2323
public static final String GAUGE_BROKER_PERMISSION = "rocketmq_broker_permission";
24+
public static final String GAUGE_TOPIC_NUM = "rocketmq_topic_number";
25+
public static final String GAUGE_CONSUMER_GROUP_NUM = "rocketmq_consumer_group_number";
2426

2527
public static final String COUNTER_MESSAGES_IN_TOTAL = "rocketmq_messages_in_total";
2628
public static final String COUNTER_MESSAGES_OUT_TOTAL = "rocketmq_messages_out_total";
2729
public static final String COUNTER_THROUGHPUT_IN_TOTAL = "rocketmq_throughput_in_total";
2830
public static final String COUNTER_THROUGHPUT_OUT_TOTAL = "rocketmq_throughput_out_total";
2931
public static final String HISTOGRAM_MESSAGE_SIZE = "rocketmq_message_size";
32+
public static final String HISTOGRAM_TOPIC_CREATE_EXECUTE_TIME = "rocketmq_topic_create_execution_time";
33+
public static final String HISTOGRAM_CONSUMER_GROUP_CREATE_EXECUTE_TIME = "rocketmq_consumer_group_create_execution_time";
3034

3135
public static final String GAUGE_PRODUCER_CONNECTIONS = "rocketmq_producer_connections";
3236
public static final String GAUGE_CONSUMER_CONNECTIONS = "rocketmq_consumer_connections";
@@ -52,6 +56,7 @@ public class BrokerMetricsConstant {
5256
public static final String LABEL_PROCESSOR = "processor";
5357

5458
public static final String LABEL_TOPIC = "topic";
59+
public static final String LABEL_INVOCATION_STATUS = "invocation_status";
5560
public static final String LABEL_IS_RETRY = "is_retry";
5661
public static final String LABEL_IS_SYSTEM = "is_system";
5762
public static final String LABEL_CONSUMER_GROUP = "consumer_group";

0 commit comments

Comments
 (0)