GenerousMan commented on code in PR #6507: URL: https://github.com/apache/rocketmq/pull/6507#discussion_r1163568199
########## broker/src/main/java/org/apache/rocketmq/broker/coldctr/ColdDataPullRequestHoldService.java: ########## @@ -0,0 +1,105 @@ +/* + * 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.broker.coldctr; + +import java.util.Iterator; +import java.util.concurrent.LinkedBlockingQueue; + +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.longpolling.PullRequest; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.SystemClock; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; + +/** + * just requests are type of pull have the qualification to be put into this hold queue. + * if the pull request is reading cold data and that request will be cold at the first time, + * then the pull request will be cold in this @code pullRequestLinkedBlockingQueue, + * in @code coldTimeoutMillis later the pull request will be warm and marked holded + */ +public class ColdDataPullRequestHoldService extends ServiceThread { + + private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_COLDCTR_LOGGER_NAME); + public static final String NO_SUSPEND_KEY = "_noSuspend_"; + + private final long coldHoldTimeoutMillis = 3000; + private final SystemClock systemClock = new SystemClock(); + private final BrokerController brokerController; + private final LinkedBlockingQueue<PullRequest> pullRequestColdHoldQueue = new LinkedBlockingQueue<>(10000); + + public void suspendColdDataReadRequest(PullRequest pullRequest) { + if (this.brokerController.getMessageStoreConfig().isColdDataFlowControlEnable()) { + pullRequestColdHoldQueue.offer(pullRequest); + } + } + + public ColdDataPullRequestHoldService(BrokerController brokerController) { + this.brokerController = brokerController; + } + + @Override + public String getServiceName() { + return ColdDataPullRequestHoldService.class.getSimpleName(); + } + + @Override + public void run() { + log.info("{} service started", this.getServiceName()); + while (!this.isStopped()) { + try { + if (!this.brokerController.getMessageStoreConfig().isColdDataFlowControlEnable()) { + this.waitForRunning(20 * 1000); Review Comment: Is it possible to continue after waiting for 20 seconds here? If cold control is not enabled, can the subsequent process of processing pullRequest be skipped? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
