opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

Delete useless code file

Browse Source
This commit is contained in:
RongtongJin
2022-03-14 14:51:49 +08:00
parent 00d2d1c40c
commit 190617e8b9
13 changed files with 0 additions and 2534 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/transaction/TransactionRecord.java
-43
View File
@@ -1,43 +0,0 @@
/*
* 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.transaction;
/**
* This class will be removed in the version 4.4.0 and {@link OperationResult} class is recommended.
*/
@Deprecated
public class TransactionRecord {
// Commit Log Offset
private long offset;
private String producerGroup;
public long getOffset() {
return offset;
}
public void setOffset(long offset) {
this.offset = offset;
}
public String getProducerGroup() {
return producerGroup;
}
public void setProducerGroup(String producerGroup) {
this.producerGroup = producerGroup;
}
}
broker/src/main/java/org/apache/rocketmq/broker/transaction/TransactionStore.java
-42
View File
@@ -1,42 +0,0 @@
/*
* 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.transaction;
import java.util.List;
/**
* This class will be removed in ther version 4.4.0, and {@link TransactionalMessageService} class is recommended.
*/
@Deprecated
public interface TransactionStore {
boolean open();
void close();
boolean put(final List<TransactionRecord> trs);
void remove(final List<Long> pks);
List<TransactionRecord> traverse(final long pk, final int nums);
long totalRecords();
long minPK();
long maxPK();
}
broker/src/main/java/org/apache/rocketmq/broker/transaction/jdbc/JDBCTransactionStore.java
-242
View File
@@ -1,242 +0,0 @@
/*
* 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.transaction.jdbc;
import java.net.URL;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.broker.transaction.TransactionRecord;
import org.apache.rocketmq.broker.transaction.TransactionStore;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class JDBCTransactionStore implements TransactionStore {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME);
private final JDBCTransactionStoreConfig jdbcTransactionStoreConfig;
private Connection connection;
private AtomicLong totalRecordsValue = new AtomicLong(0);
public JDBCTransactionStore(JDBCTransactionStoreConfig jdbcTransactionStoreConfig) {
this.jdbcTransactionStoreConfig = jdbcTransactionStoreConfig;
}
@Override
public boolean open() {
if (this.loadDriver()) {
Properties props = new Properties();
props.put("user", jdbcTransactionStoreConfig.getJdbcUser());
props.put("password", jdbcTransactionStoreConfig.getJdbcPassword());
try {
this.connection =
DriverManager.getConnection(this.jdbcTransactionStoreConfig.getJdbcURL(), props);
this.connection.setAutoCommit(false);
if (!this.computeTotalRecords()) {
return this.createDB();
}
return true;
} catch (SQLException e) {
log.info("Create JDBC Connection Exception", e);
}
}
return false;
}
private boolean loadDriver() {
try {
Class.forName(this.jdbcTransactionStoreConfig.getJdbcDriverClass()).newInstance();
log.info("Loaded the appropriate driver, {}",
this.jdbcTransactionStoreConfig.getJdbcDriverClass());
return true;
} catch (Exception e) {
log.info("Loaded the appropriate driver Exception", e);
}
return false;
}
private boolean computeTotalRecords() {
Statement statement = null;
ResultSet resultSet = null;
try {
statement = this.connection.createStatement();
resultSet = statement.executeQuery("select count(offset) as total from t_transaction");
if (!resultSet.next()) {
log.warn("computeTotalRecords ResultSet is empty");
return false;
}
this.totalRecordsValue.set(resultSet.getLong(1));
} catch (Exception e) {
log.warn("computeTotalRecords Exception", e);
return false;
} finally {
if (null != statement) {
try {
statement.close();
} catch (SQLException e) {
}
}
if (null != resultSet) {
try {
resultSet.close();
} catch (SQLException e) {
}
}
}
return true;
}
private boolean createDB() {
Statement statement = null;
try {
statement = this.connection.createStatement();
String sql = this.createTableSql();
log.info("createDB SQL:\n {}", sql);
statement.execute(sql);
this.connection.commit();
return true;
} catch (Exception e) {
log.warn("createDB Exception", e);
return false;
} finally {
if (null != statement) {
try {
statement.close();
} catch (SQLException e) {
log.warn("Close statement exception", e);
}
}
}
}
private String createTableSql() {
URL resource = JDBCTransactionStore.class.getClassLoader().getResource("transaction.sql");
String fileContent = MixAll.file2String(resource);
return fileContent;
}
@Override
public void close() {
try {
if (this.connection != null) {
this.connection.close();
}
} catch (SQLException e) {
}
}
@Override
public boolean put(List<TransactionRecord> trs) {
PreparedStatement statement = null;
try {
this.connection.setAutoCommit(false);
statement = this.connection.prepareStatement("insert into t_transaction values (?, ?)");
for (TransactionRecord tr : trs) {
statement.setLong(1, tr.getOffset());
statement.setString(2, tr.getProducerGroup());
statement.addBatch();
}
int[] executeBatch = statement.executeBatch();
this.connection.commit();
this.totalRecordsValue.addAndGet(updatedRows(executeBatch));
return true;
} catch (Exception e) {
log.warn("createDB Exception", e);
return false;
} finally {
if (null != statement) {
try {
statement.close();
} catch (SQLException e) {
log.warn("Close statement exception", e);
}
}
}
}
private long updatedRows(int[] rows) {
long res = 0;
for (int i : rows) {
res += i;
}
return res;
}
@Override
public void remove(List<Long> pks) {
PreparedStatement statement = null;
try {
this.connection.setAutoCommit(false);
statement = this.connection.prepareStatement("DELETE FROM t_transaction WHERE offset = ?");
for (long pk : pks) {
statement.setLong(1, pk);
statement.addBatch();
}
int[] executeBatch = statement.executeBatch();
this.connection.commit();
} catch (Exception e) {
log.warn("createDB Exception", e);
} finally {
if (null != statement) {
try {
statement.close();
} catch (SQLException e) {
}
}
}
}
@Override
public List<TransactionRecord> traverse(long pk, int nums) {
return null;
}
@Override
public long totalRecords() {
return this.totalRecordsValue.get();
}
@Override
public long minPK() {
return 0;
}
@Override
public long maxPK() {
return 0;
}
}
broker/src/main/java/org/apache/rocketmq/broker/transaction/jdbc/JDBCTransactionStoreConfig.java
-57
View File
@@ -1,57 +0,0 @@
/*
* 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.transaction.jdbc;
public class JDBCTransactionStoreConfig {
private String jdbcDriverClass = "com.mysql.jdbc.Driver";
private String jdbcURL = "jdbc:mysql://xxx.xxx.xxx.xxx:1000/xxx?useUnicode=true&characterEncoding=UTF-8";
private String jdbcUser = "xxx";
private String jdbcPassword = "xxx";
public String getJdbcDriverClass() {
return jdbcDriverClass;
}
public void setJdbcDriverClass(String jdbcDriverClass) {
this.jdbcDriverClass = jdbcDriverClass;
}
public String getJdbcURL() {
return jdbcURL;
}
public void setJdbcURL(String jdbcURL) {
this.jdbcURL = jdbcURL;
}
public String getJdbcUser() {
return jdbcUser;
}
public void setJdbcUser(String jdbcUser) {
this.jdbcUser = jdbcUser;
}
public String getJdbcPassword() {
return jdbcPassword;
}
public void setJdbcPassword(String jdbcPassword) {
this.jdbcPassword = jdbcPassword;
}
}
broker/src/main/java/org/apache/rocketmq/broker/util/ServiceProvider.java
-201
View File
@@ -1,201 +0,0 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
public class ServiceProvider {
private final static Logger LOG = LoggerFactory
.getLogger(ServiceProvider.class);
/**
* A reference to the classloader that loaded this class. It's more efficient to compute it once and cache it here.
*/
private static ClassLoader thisClassLoader;
/**
* JDK1.3+ <a href= "http://java.sun.com/j2se/1.3/docs/guide/jar/jar.html#Service%20Provider" > 'Service Provider' specification</a>.
*/
public static final String TRANSACTION_SERVICE_ID = "META-INF/service/org.apache.rocketmq.broker.transaction.TransactionalMessageService";
public static final String TRANSACTION_LISTENER_ID = "META-INF/service/org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener";
public static final String RPC_HOOK_ID = "META-INF/service/org.apache.rocketmq.remoting.RPCHook";
public static final String ACL_VALIDATOR_ID = "META-INF/service/org.apache.rocketmq.acl.AccessValidator";
static {
thisClassLoader = getClassLoader(ServiceProvider.class);
}
/**
* Returns a string that uniquely identifies the specified object, including its class.
* <p>
* The returned string is of form "classname@hashcode", ie is the same as the return value of the Object.toString() method, but works even when the specified object's class has overidden the toString method.
*
* @param o may be null.
* @return a string of form classname@hashcode, or "null" if param o is null.
*/
protected static String objectId(Object o) {
if (o == null) {
return "null";
} else {
return o.getClass().getName() + "@" + System.identityHashCode(o);
}
}
protected static ClassLoader getClassLoader(Class<?> clazz) {
try {
return clazz.getClassLoader();
} catch (SecurityException e) {
LOG.error("Unable to get classloader for class {} due to security restrictions !",
clazz, e.getMessage());
throw e;
}
}
protected static ClassLoader getContextClassLoader() {
ClassLoader classLoader = null;
try {
classLoader = Thread.currentThread().getContextClassLoader();
} catch (SecurityException ex) {
/**
* The getContextClassLoader() method throws SecurityException when the context
* class loader isn't an ancestor of the calling class's class
* loader, or if security permissions are restricted.
*/
}
return classLoader;
}
protected static InputStream getResourceAsStream(ClassLoader loader, String name) {
if (loader != null) {
return loader.getResourceAsStream(name);
} else {
return ClassLoader.getSystemResourceAsStream(name);
}
}
public static <T> List<T> load(String name, Class<?> clazz) {
LOG.info("Looking for a resource file of name [{}] ...", name);
List<T> services = new ArrayList<T>();
try {
ArrayList<String> names = new ArrayList<String>();
final InputStream is = getResourceAsStream(getContextClassLoader(), name);
if (is != null) {
BufferedReader reader;
try {
reader = new BufferedReader(new InputStreamReader(is, "UTF-8"));
} catch (java.io.UnsupportedEncodingException e) {
reader = new BufferedReader(new InputStreamReader(is));
}
String serviceName = reader.readLine();
while (serviceName != null && !"".equals(serviceName)) {
LOG.info(
"Creating an instance as specified by file {} which was present in the path of the context classloader.",
name);
if (!names.contains(serviceName)) {
names.add(serviceName);
}
services.add((T)initService(getContextClassLoader(), serviceName, clazz));
serviceName = reader.readLine();
}
reader.close();
} else {
// is == null
LOG.warn("No resource file with name [{}] found.", name);
}
} catch (Exception e) {
LOG.error("Error occured when looking for resource file " + name, e);
}
return services;
}
public static <T> T loadClass(String name, Class<?> clazz) {
final InputStream is = getResourceAsStream(getContextClassLoader(), name);
if (is != null) {
BufferedReader reader;
try {
try {
reader = new BufferedReader(new InputStreamReader(is, "UTF-8"));
} catch (java.io.UnsupportedEncodingException e) {
reader = new BufferedReader(new InputStreamReader(is));
}
String serviceName = reader.readLine();
reader.close();
if (serviceName != null && !"".equals(serviceName)) {
return initService(getContextClassLoader(), serviceName, clazz);
} else {
LOG.warn("ServiceName is empty!");
return null;
}
} catch (Exception e) {
LOG.warn("Error occurred when looking for resource file " + name, e);
}
}
return null;
}
protected static <T> T initService(ClassLoader classLoader, String serviceName, Class<?> clazz) {
Class<?> serviceClazz = null;
try {
if (classLoader != null) {
try {
// Warning: must typecast here & allow exception to be generated/caught & recast properly
serviceClazz = classLoader.loadClass(serviceName);
if (clazz.isAssignableFrom(serviceClazz)) {
LOG.info("Loaded class {} from classloader {}", serviceClazz.getName(),
objectId(classLoader));
} else {
// This indicates a problem with the ClassLoader tree. An incompatible ClassLoader was used to load the implementation.
LOG.error(
"Class {} loaded from classloader {} does not extend {} as loaded by this classloader.",
new Object[] {serviceClazz.getName(),
objectId(serviceClazz.getClassLoader()), clazz.getName()});
}
return (T)serviceClazz.newInstance();
} catch (ClassNotFoundException ex) {
if (classLoader == thisClassLoader) {
// Nothing more to try, onwards.
LOG.warn("Unable to locate any class {} via classloader", serviceName,
objectId(classLoader));
throw ex;
}
// Ignore exception, continue
} catch (NoClassDefFoundError e) {
if (classLoader == thisClassLoader) {
// Nothing more to try, onwards.
LOG.warn(
"Class {} cannot be loaded via classloader {}.it depends on some other class that cannot be found.",
serviceClazz, objectId(classLoader));
throw e;
}
// Ignore exception, continue
}
}
} catch (Exception e) {
LOG.error("Unable to init service.", e);
}
return (T)serviceClazz;
}
}
namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java
-523
View File
@@ -1,523 +0,0 @@
/*
* 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.namesrv.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.common.protocol.header.namesrv.DeleteKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.PutKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.RegisterBrokerRequestHeader;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.assertj.core.util.Maps;
import org.junit.Before;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class DefaultRequestProcessorTest {
private DefaultRequestProcessor defaultRequestProcessor;
private NamesrvController namesrvController;
private NamesrvConfig namesrvConfig;
private NettyServerConfig nettyServerConfig;
private RouteInfoManager routeInfoManager;
private InternalLogger logger;
@Before
public void init() throws Exception {
namesrvConfig = new NamesrvConfig();
nettyServerConfig = new NettyServerConfig();
routeInfoManager = new RouteInfoManager();
namesrvController = new NamesrvController(namesrvConfig, nettyServerConfig);
Field field = NamesrvController.class.getDeclaredField("routeInfoManager");
field.setAccessible(true);
field.set(namesrvController, routeInfoManager);
defaultRequestProcessor = new DefaultRequestProcessor(namesrvController);
registerRouteInfoManager();
logger = mock(InternalLogger.class);
setFinalStatic(DefaultRequestProcessor.class.getDeclaredField("log"), logger);
}
@Test
public void testProcessRequest_PutKVConfig() throws RemotingCommandException {
PutKVConfigRequestHeader header = new PutKVConfigRequestHeader();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PUT_KV_CONFIG,
header);
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
request.addExtField("value", "value");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
assertThat(namesrvController.getKvConfigManager().getKVConfig("namespace", "key"))
.isEqualTo("value");
}
@Test
public void testProcessRequest_GetKVConfigReturnNotNull() throws RemotingCommandException {
namesrvController.getKvConfigManager().putKVConfig("namespace", "key", "value");
GetKVConfigRequestHeader header = new GetKVConfigRequestHeader();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_KV_CONFIG,
header);
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
GetKVConfigResponseHeader responseHeader = (GetKVConfigResponseHeader) response
.readCustomHeader();
assertThat(responseHeader.getValue()).isEqualTo("value");
}
@Test
public void testProcessRequest_GetKVConfigReturnNull() throws RemotingCommandException {
GetKVConfigRequestHeader header = new GetKVConfigRequestHeader();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_KV_CONFIG,
header);
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);
assertThat(response.getRemark()).isEqualTo("No config item, Namespace: namespace Key: key");
GetKVConfigResponseHeader responseHeader = (GetKVConfigResponseHeader) response
.readCustomHeader();
assertThat(responseHeader.getValue()).isNull();
}
@Test
public void testProcessRequest_DeleteKVConfig() throws RemotingCommandException {
namesrvController.getKvConfigManager().putKVConfig("namespace", "key", "value");
DeleteKVConfigRequestHeader header = new DeleteKVConfigRequestHeader();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_KV_CONFIG,
header);
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
assertThat(namesrvController.getKvConfigManager().getKVConfig("namespace", "key"))
.isNull();
}
@Test
public void testProcessRequest_RegisterBroker() throws RemotingCommandException,
NoSuchFieldException, IllegalAccessException {
RemotingCommand request = genSampleRegisterCmd(true);
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand response = defaultRequestProcessor.processRequest(ctx, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
RouteInfoManager routes = namesrvController.getRouteInfoManager();
Field brokerAddrTable = RouteInfoManager.class.getDeclaredField("brokerAddrTable");
brokerAddrTable.setAccessible(true);
BrokerData broker = new BrokerData();
broker.setBrokerName("broker");
broker.setBrokerAddrs((HashMap) Maps.newHashMap(new Long(2333), "10.10.1.1"));
assertThat((Map) brokerAddrTable.get(routes))
.contains(new HashMap.SimpleEntry("broker", broker));
}
/*@Test
public void testProcessRequest_RegisterBrokerLogicalQueue() throws Exception {
String cluster = "cluster";
String broker1Name = "broker1";
String broker1Addr = "10.10.1.1";
String broker2Name = "broker2";
String broker2Addr = "10.10.1.2";
String topic = "foobar";
LogicalQueueRouteData queueRouteData1 = new LogicalQueueRouteData(0, 0, new MessageQueue(topic, broker1Name, 0), MessageQueueRouteState.ReadOnly, 0, 10, 100, 100, broker1Addr);
{
RegisterBrokerRequestHeader header = new RegisterBrokerRequestHeader();
header.setBrokerName(broker1Name);
RemotingCommand request = RemotingCommand.createRequestCommand(
RequestCode.REGISTER_BROKER, header);
request.addExtField("brokerName", broker1Name);
request.addExtField("brokerAddr", broker1Addr);
request.addExtField("clusterName", cluster);
request.addExtField("haServerAddr", "10.10.2.1");
request.addExtField("brokerId", String.valueOf(MixAll.MASTER_ID));
request.setVersion(MQVersion.CURRENT_VERSION);
TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
topicConfigSerializeWrapper.setTopicConfigTable(new ConcurrentHashMap<>(Collections.singletonMap(topic, new TopicConfig(topic))));
topicConfigSerializeWrapper.setLogicalQueuesInfoMap(Maps.newHashMap(topic, new LogicalQueuesInfo(Collections.singletonMap(0, Lists.newArrayList(
queueRouteData1
)))));
topicConfigSerializeWrapper.setDataVersion(new DataVersion());
RegisterBrokerBody requestBody = new RegisterBrokerBody();
requestBody.setTopicConfigSerializeWrapper(topicConfigSerializeWrapper);
requestBody.setFilterServerList(Lists.<String>newArrayList());
request.setBody(requestBody.encode());
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand response = defaultRequestProcessor.processRequest(ctx, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
}
LogicalQueueRouteData queueRouteData2 = new LogicalQueueRouteData(0, 100, new MessageQueue(topic, broker2Name, 0), MessageQueueRouteState.Normal, 0, -1, -1, -1, broker2Addr);
LogicalQueueRouteData queueRouteData3 = new LogicalQueueRouteData(1, 100, new MessageQueue(topic, broker2Name, 0), MessageQueueRouteState.Normal, 0, -1, -1, -1, broker2Addr);
{
RegisterBrokerRequestHeader header = new RegisterBrokerRequestHeader();
header.setBrokerName(broker2Name);
RemotingCommand request = RemotingCommand.createRequestCommand(
RequestCode.REGISTER_BROKER, header);
request.addExtField("brokerName", broker2Name);
request.addExtField("brokerAddr", broker2Addr);
request.addExtField("clusterName", cluster);
request.addExtField("haServerAddr", "10.10.2.1");
request.addExtField("brokerId", String.valueOf(MixAll.MASTER_ID));
request.setVersion(MQVersion.CURRENT_VERSION);
TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
topicConfigSerializeWrapper.setTopicConfigTable(new ConcurrentHashMap<>(Collections.singletonMap(topic, new TopicConfig(topic))));
topicConfigSerializeWrapper.setLogicalQueuesInfoMap(Maps.newHashMap(topic, new LogicalQueuesInfo(ImmutableMap.of(
0, Collections.singletonList(queueRouteData2),
1, Collections.singletonList(queueRouteData3)
))));
topicConfigSerializeWrapper.setDataVersion(new DataVersion());
RegisterBrokerBody requestBody = new RegisterBrokerBody();
requestBody.setTopicConfigSerializeWrapper(topicConfigSerializeWrapper);
requestBody.setFilterServerList(Lists.<String>newArrayList());
request.setBody(requestBody.encode());
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand response = defaultRequestProcessor.processRequest(ctx, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
}
{
GetRouteInfoRequestHeader header = new GetRouteInfoRequestHeader();
header.setTopic(topic);
header.setSysFlag(MessageSysFlag.LOGICAL_QUEUE_FLAG);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, header);
request.makeCustomHeaderToNet();
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand response = defaultRequestProcessor.processRequest(ctx, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
TopicRouteDataNameSrv topicRouteDataNameSrv = JSON.parseObject(response.getBody(), TopicRouteDataNameSrv.class);
assertThat(topicRouteDataNameSrv).isNotNull();
LogicalQueuesInfoUnordered logicalQueuesInfoUnordered = new LogicalQueuesInfoUnordered();
logicalQueuesInfoUnordered.put(0, ImmutableMap.of(
new LogicalQueuesInfoUnordered.Key(queueRouteData1.getBrokerName(), queueRouteData1.getQueueId(), queueRouteData1.getOffsetDelta()), queueRouteData1,
new LogicalQueuesInfoUnordered.Key(queueRouteData2.getBrokerName(), queueRouteData2.getQueueId(), queueRouteData2.getOffsetDelta()), queueRouteData2
));
logicalQueuesInfoUnordered.put(1, ImmutableMap.of(new LogicalQueuesInfoUnordered.Key(queueRouteData3.getBrokerName(), queueRouteData3.getQueueId(), queueRouteData3.getOffsetDelta()), queueRouteData3));
assertThat(topicRouteDataNameSrv.getLogicalQueuesInfoUnordered()).isEqualTo(logicalQueuesInfoUnordered);
}
}
*/
@Test
public void testProcessRequest_RegisterBrokerWithFilterServer() throws RemotingCommandException,
NoSuchFieldException, IllegalAccessException {
RemotingCommand request = genSampleRegisterCmd(true);
// version >= MQVersion.Version.V3_0_11.ordinal() to register with filter server
request.setVersion(100);
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand response = defaultRequestProcessor.processRequest(ctx, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
RouteInfoManager routes = namesrvController.getRouteInfoManager();
Field brokerAddrTable = RouteInfoManager.class.getDeclaredField("brokerAddrTable");
brokerAddrTable.setAccessible(true);
BrokerData broker = new BrokerData();
broker.setBrokerName("broker");
broker.setBrokerAddrs((HashMap) Maps.newHashMap(new Long(2333), "10.10.1.1"));
assertThat((Map) brokerAddrTable.get(routes))
.contains(new HashMap.SimpleEntry("broker", broker));
}
@Test
public void testProcessRequest_UnregisterBroker() throws RemotingCommandException, NoSuchFieldException, IllegalAccessException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
//Register broker
RemotingCommand regRequest = genSampleRegisterCmd(true);
defaultRequestProcessor.processRequest(ctx, regRequest);
//Unregister broker
RemotingCommand unregRequest = genSampleRegisterCmd(false);
RemotingCommand unregResponse = defaultRequestProcessor.processRequest(ctx, unregRequest);
assertThat(unregResponse.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(unregResponse.getRemark()).isNull();
RouteInfoManager routes = namesrvController.getRouteInfoManager();
Field brokerAddrTable = RouteInfoManager.class.getDeclaredField("brokerAddrTable");
brokerAddrTable.setAccessible(true);
assertThat((Map) brokerAddrTable.get(routes)).isNotEmpty();
}
@Test
public void testGetRouteInfoByTopic() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC);
RemotingCommand remotingCommandSuccess = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommandSuccess.getCode()).isEqualTo(ResponseCode.SUCCESS);
request.getExtFields().put("topic", "test");
RemotingCommand remotingCommandNoTopicRouteInfo = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommandNoTopicRouteInfo.getCode()).isEqualTo(ResponseCode.TOPIC_NOT_EXIST);
}
@Test
public void testGetBrokerClusterInfo() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_BROKER_CLUSTER_INFO);
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testWipeWritePermOfBroker() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.WIPE_WRITE_PERM_OF_BROKER);
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetAllTopicListFromNameserver() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER);
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testDeleteTopicInNamesrv() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.DELETE_TOPIC_IN_NAMESRV);
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetKVListByNamespace() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_KVLIST_BY_NAMESPACE);
request.addExtField("namespace", "default-namespace-1");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);
namesrvController.getKvConfigManager().putKVConfig("default-namespace-1", "key", "value");
RemotingCommand remotingCommandSuccess = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommandSuccess.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetTopicsByCluster() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_TOPICS_BY_CLUSTER);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetSystemTopicListFromNs() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetUnitTopicList() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_UNIT_TOPIC_LIST);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetHasUnitSubTopicList() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetHasUnitSubUnUnitTopicList() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testUpdateConfig() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.UPDATE_NAMESRV_CONFIG);
request.addExtField("cluster", "default-cluster");
Map<String, String> propertiesMap = new HashMap<>();
propertiesMap.put("key", "value");
request.setBody(propertiesMap.toString().getBytes());
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testGetConfig() throws RemotingCommandException {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
when(ctx.channel()).thenReturn(null);
RemotingCommand request = getRemotingCommand(RequestCode.GET_NAMESRV_CONFIG);
request.addExtField("cluster", "default-cluster");
RemotingCommand remotingCommand = defaultRequestProcessor.processRequest(ctx, request);
assertThat(remotingCommand.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
private RemotingCommand getRemotingCommand(int code) {
RegisterBrokerRequestHeader header = new RegisterBrokerRequestHeader();
header.setBrokerName("broker");
RemotingCommand request = RemotingCommand.createRequestCommand(code, header);
request.addExtField("brokerName", "broker");
request.addExtField("brokerAddr", "10.10.1.1");
request.addExtField("clusterName", "cluster");
request.addExtField("haServerAddr", "10.10.2.1");
request.addExtField("brokerId", "2333");
request.addExtField("topic", "unit-test");
return request;
}
private static RemotingCommand genSampleRegisterCmd(boolean reg) {
RegisterBrokerRequestHeader header = new RegisterBrokerRequestHeader();
header.setBrokerName("broker");
RemotingCommand request = RemotingCommand.createRequestCommand(
reg ? RequestCode.REGISTER_BROKER : RequestCode.UNREGISTER_BROKER, header);
request.addExtField("brokerName", "broker");
request.addExtField("brokerAddr", "10.10.1.1");
request.addExtField("clusterName", "cluster");
request.addExtField("haServerAddr", "10.10.2.1");
request.addExtField("brokerId", "2333");
return request;
}
private static void setFinalStatic(Field field, Object newValue) throws Exception {
field.setAccessible(true);
Field modifiersField = Field.class.getDeclaredField("modifiers");
modifiersField.setAccessible(true);
modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);
field.set(null, newValue);
}
private void registerRouteInfoManager() {
TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
ConcurrentHashMap<String, TopicConfig> topicConfigConcurrentHashMap = new ConcurrentHashMap<>();
TopicConfig topicConfig = new TopicConfig();
topicConfig.setWriteQueueNums(8);
topicConfig.setTopicName("unit-test");
topicConfig.setPerm(6);
topicConfig.setReadQueueNums(8);
topicConfig.setOrder(false);
topicConfigConcurrentHashMap.put("unit-test", topicConfig);
topicConfigSerializeWrapper.setTopicConfigTable(topicConfigConcurrentHashMap);
Channel channel = mock(Channel.class);
RegisterBrokerResult registerBrokerResult = routeInfoManager.registerBroker("default-cluster", "127.0.0.1:10911", "default-broker", 0, "127.0.0.1:1001",
topicConfigSerializeWrapper, new ArrayList<String>(), channel);
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/netty/AsyncNettyRequestProcessor.java
-29
View File
@@ -1,29 +0,0 @@
/*
* 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.remoting.netty;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public abstract class AsyncNettyRequestProcessor implements NettyRequestProcessor {
public void asyncProcessRequest(ChannelHandlerContext ctx, RemotingCommand request, RemotingResponseCallback responseCallback) throws Exception {
RemotingCommand response = processRequest(ctx, request);
responseCallback.callback(response);
}
}
store/src/main/java/org/apache/rocketmq/store/MessageExtBatch.java
-51
View File
@@ -1,51 +0,0 @@
/**
* 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;
import java.nio.ByteBuffer;
public class MessageExtBatch extends MessageExtBrokerInner {
private static final long serialVersionUID = -2353110995348498537L;
/**
* Inner batch means the batch dose not need to be unwrapped
*/
private boolean isInnerBatch = false;
public ByteBuffer wrap() {
assert getBody() != null;
return ByteBuffer.wrap(getBody(), 0, getBody().length);
}
public boolean isInnerBatch() {
return isInnerBatch;
}
public void setInnerBatch(boolean innerBatch) {
isInnerBatch = innerBatch;
}
private ByteBuffer encodedBuff;
public ByteBuffer getEncodedBuff() {
return encodedBuff;
}
public void setEncodedBuff(ByteBuffer encodedBuff) {
this.encodedBuff = encodedBuff;
}
}
store/src/main/java/org/apache/rocketmq/store/MessageExtBrokerInner.java
-64
View File
@@ -1,64 +0,0 @@
/*
* 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;
import java.nio.ByteBuffer;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.message.MessageExt;
public class MessageExtBrokerInner extends MessageExt {
private static final long serialVersionUID = 7256001576878700634L;
private String propertiesString;
private long tagsCode;
private ByteBuffer encodedBuff;
public ByteBuffer getEncodedBuff() {
return encodedBuff;
}
public void setEncodedBuff(ByteBuffer encodedBuff) {
this.encodedBuff = encodedBuff;
}
public static long tagsString2tagsCode(final TopicFilterType filter, final String tags) {
if (null == tags || tags.length() == 0) { return 0; }
return tags.hashCode();
}
public static long tagsString2tagsCode(final String tags) {
return tagsString2tagsCode(null, tags);
}
public String getPropertiesString() {
return propertiesString;
}
public void setPropertiesString(String propertiesString) {
this.propertiesString = propertiesString;
}
public long getTagsCode() {
return tagsCode;
}
public void setTagsCode(long tagsCode) {
this.tagsCode = tagsCode;
}
}
store/src/main/java/org/apache/rocketmq/store/schedule/DelayOffsetSerializeWrapper.java
-34
View File
@@ -1,34 +0,0 @@
/*
* 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.schedule;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
public class DelayOffsetSerializeWrapper extends RemotingSerializable {
private ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
new ConcurrentHashMap<Integer, Long>(32);
public ConcurrentMap<Integer, Long> getOffsetTable() {
return offsetTable;
}
public void setOffsetTable(ConcurrentMap<Integer, Long> offsetTable) {
this.offsetTable = offsetTable;
}
}
store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
-819
View File
@@ -1,819 +0,0 @@
/*
* 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.schedule;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.running.RunningStats;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.ReferredIterator;
public class ScheduleMessageService extends ConfigManager {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
private static final long FIRST_DELAY_TIME = 1000L;
private static final long DELAY_FOR_A_WHILE = 100L;
private static final long DELAY_FOR_A_PERIOD = 10000L;
private static final long WAIT_FOR_SHUTDOWN = 5000L;
private static final long DELAY_FOR_A_SLEEP = 10L;
private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =
new ConcurrentHashMap<Integer, Long>(32);
private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
new ConcurrentHashMap<Integer, Long>(32);
private final DefaultMessageStore defaultMessageStore;
private final AtomicBoolean started = new AtomicBoolean(false);
private ScheduledExecutorService deliverExecutorService;
private MessageStore writeMessageStore;
private int maxDelayLevel;
private boolean enableAsyncDeliver = false;
private ScheduledExecutorService handleExecutorService;
private final Map<Integer /* level */, LinkedBlockingQueue<PutResultProcess>> deliverPendingTable =
new ConcurrentHashMap<>(32);
public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {
this.defaultMessageStore = defaultMessageStore;
this.writeMessageStore = defaultMessageStore;
if (defaultMessageStore != null) {
this.enableAsyncDeliver = defaultMessageStore.getMessageStoreConfig().isEnableScheduleAsyncDeliver();
}
}
public static int queueId2DelayLevel(final int queueId) {
return queueId + 1;
}
public static int delayLevel2QueueId(final int delayLevel) {
return delayLevel - 1;
}
/**
* @param writeMessageStore the writeMessageStore to set
*/
public void setWriteMessageStore(MessageStore writeMessageStore) {
this.writeMessageStore = writeMessageStore;
}
public void buildRunningStats(HashMap<String, String> stats) {
Iterator<Map.Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<Integer, Long> next = it.next();
int queueId = delayLevel2QueueId(next.getKey());
long delayOffset = next.getValue();
long maxOffset = this.defaultMessageStore.getMaxOffsetInQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, queueId);
String value = String.format("%d,%d", delayOffset, maxOffset);
String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey());
stats.put(key, value);
}
}
private void updateOffset(int delayLevel, long offset) {
this.offsetTable.put(delayLevel, offset);
}
public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
Long time = this.delayLevelTable.get(delayLevel);
if (time != null) {
return time + storeTimestamp;
}
return storeTimestamp + 1000;
}
public void start() {
if (started.compareAndSet(false, true)) {
super.load();
this.deliverExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
if (this.enableAsyncDeliver) {
this.handleExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
}
for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
Integer level = entry.getKey();
Long timeDelay = entry.getValue();
Long offset = this.offsetTable.get(level);
if (null == offset) {
offset = 0L;
}
if (timeDelay != null) {
if (this.enableAsyncDeliver) {
this.handleExecutorService.schedule(new HandlePutResultTask(level), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
}
this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
}
}
this.deliverExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
if (started.get()) {
ScheduleMessageService.this.persist();
}
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}
}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
}
}
public void shutdown() {
if (this.started.compareAndSet(true, false) && null != this.deliverExecutorService) {
this.deliverExecutorService.shutdown();
try {
this.deliverExecutorService.awaitTermination(WAIT_FOR_SHUTDOWN, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("deliverExecutorService awaitTermination error", e);
}
if (this.handleExecutorService != null) {
this.handleExecutorService.shutdown();
try {
this.handleExecutorService.awaitTermination(WAIT_FOR_SHUTDOWN, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("handleExecutorService awaitTermination error", e);
}
}
if (this.deliverPendingTable != null) {
for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
}
}
this.persist();
}
}
public boolean isStarted() {
return started.get();
}
public int getMaxDelayLevel() {
return maxDelayLevel;
}
@Override
public String encode() {
return this.encode(false);
}
@Override
public boolean load() {
boolean result = super.load();
result = result && this.parseDelayLevel();
result = result && this.correctDelayOffset();
return result;
}
public boolean correctDelayOffset() {
try {
for (int delayLevel : delayLevelTable.keySet()) {
ConsumeQueueInterface cq =
ScheduleMessageService.this.defaultMessageStore.getQueueStore().findOrCreateConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
delayLevel2QueueId(delayLevel));
Long currentDelayOffset = offsetTable.get(delayLevel);
if (currentDelayOffset == null || cq == null) {
continue;
}
long correctDelayOffset = currentDelayOffset;
long cqMinOffset = cq.getMinOffsetInQueue();
long cqMaxOffset = cq.getMaxOffsetInQueue();
if (currentDelayOffset < cqMinOffset) {
correctDelayOffset = cqMinOffset;
log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}",
currentDelayOffset, cqMinOffset, cqMaxOffset, cq.getQueueId());
}
if (currentDelayOffset > cqMaxOffset) {
correctDelayOffset = cqMaxOffset;
log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}",
currentDelayOffset, cqMinOffset, cqMaxOffset, cq.getQueueId());
}
if (correctDelayOffset != currentDelayOffset) {
log.error("correct delay offset [ delayLevel {} ] from {} to {}", delayLevel, currentDelayOffset, correctDelayOffset);
offsetTable.put(delayLevel, correctDelayOffset);
}
}
} catch (Exception e) {
log.error("correctDelayOffset exception", e);
return false;
}
return true;
}
@Override
public String configFilePath() {
return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig()
.getStorePathRootDir());
}
@Override
public void decode(String jsonString) {
if (jsonString != null) {
DelayOffsetSerializeWrapper delayOffsetSerializeWrapper =
DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);
if (delayOffsetSerializeWrapper != null) {
this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());
}
}
}
@Override
public String encode(final boolean prettyFormat) {
DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper();
delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable);
return delayOffsetSerializeWrapper.toJson(prettyFormat);
}
public boolean parseDelayLevel() {
HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
timeUnitTable.put("s", 1000L);
timeUnitTable.put("m", 1000L * 60);
timeUnitTable.put("h", 1000L * 60 * 60);
timeUnitTable.put("d", 1000L * 60 * 60 * 24);
String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
try {
String[] levelArray = levelString.split(" ");
for (int i = 0; i < levelArray.length; i++) {
String value = levelArray[i];
String ch = value.substring(value.length() - 1);
Long tu = timeUnitTable.get(ch);
int level = i + 1;
if (level > this.maxDelayLevel) {
this.maxDelayLevel = level;
}
long num = Long.parseLong(value.substring(0, value.length() - 1));
long delayTimeMillis = tu * num;
this.delayLevelTable.put(level, delayTimeMillis);
if (this.enableAsyncDeliver) {
this.deliverPendingTable.put(level, new LinkedBlockingQueue<>());
}
}
} catch (Exception e) {
log.error("parseDelayLevel exception", e);
log.info("levelString String = {}", levelString);
return false;
}
return true;
}
private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setBody(msgExt.getBody());
msgInner.setFlag(msgExt.getFlag());
MessageAccessor.setProperties(msgInner, msgExt.getProperties());
TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
long tagsCodeValue =
MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
msgInner.setTagsCode(tagsCodeValue);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
msgInner.setSysFlag(msgExt.getSysFlag());
msgInner.setBornTimestamp(msgExt.getBornTimestamp());
msgInner.setBornHost(msgExt.getBornHost());
msgInner.setStoreHost(msgExt.getStoreHost());
msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());
msgInner.setWaitStoreMsgOK(false);
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));
String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
int queueId = Integer.parseInt(queueIdStr);
msgInner.setQueueId(queueId);
return msgInner;
}
public int computeDelayLevel(long timeMillis) {
long intervalMillis = timeMillis - System.currentTimeMillis();
List<Map.Entry<Integer, Long>> sortedLevels = delayLevelTable.entrySet().stream().sorted(Comparator.comparingLong(Map.Entry::getValue)).collect(Collectors.toList());
for (Map.Entry<Integer, Long> entry : sortedLevels) {
if (entry.getValue() > intervalMillis) {
return entry.getKey();
}
}
return sortedLevels.get(sortedLevels.size() - 1).getKey();
}
class DeliverDelayedMessageTimerTask implements Runnable {
private final int delayLevel;
private final long offset;
public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {
this.delayLevel = delayLevel;
this.offset = offset;
}
@Override
public void run() {
try {
if (isStarted()) {
this.executeOnTimeup();
}
} catch (Exception e) {
// XXX: warn and notify me
log.error("ScheduleMessageService, executeOnTimeup exception", e);
this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_PERIOD);
}
}
/**
* @return
*/
private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {
long result = deliverTimestamp;
long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel);
if (deliverTimestamp > maxTimestamp) {
result = now;
}
return result;
}
public void executeOnTimeup() {
ConsumeQueueInterface cq =
ScheduleMessageService.this.defaultMessageStore.getConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
delayLevel2QueueId(delayLevel));
if (cq == null) {
this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_WHILE);
return;
}
ReferredIterator<CqUnit> bufferCQ = cq.iterateFrom(this.offset);
if (bufferCQ == null) {
long resetOffset;
if ((resetOffset = cq.getMinOffsetInQueue()) > this.offset) {
log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, queueId={}",
this.offset, resetOffset, cq.getQueueId());
} else if ((resetOffset = cq.getMaxOffsetInQueue()) < this.offset) {
log.error("schedule CQ offset invalid. offset={}, cqMaxOffset={}, queueId={}",
this.offset, resetOffset, cq.getQueueId());
} else {
resetOffset = this.offset;
}
this.scheduleNextTimerTask(resetOffset, DELAY_FOR_A_WHILE);
return;
}
long nextOffset = this.offset;
try {
while (bufferCQ.hasNext() && isStarted()) {
CqUnit cqUnit = bufferCQ.next();
long offsetPy = cqUnit.getPos();
int sizePy = cqUnit.getSize();
long tagsCode = cqUnit.getTagsCode();
if (!cqUnit.isTagsCodeValid()) {
//can't find ext content.So re compute tags code.
log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
tagsCode, offsetPy, sizePy);
long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
}
long now = System.currentTimeMillis();
long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
long currOffset = cqUnit.getQueueOffset();
assert cqUnit.getBatchNum() == 1;
nextOffset = currOffset + cqUnit.getBatchNum();
long countdown = deliverTimestamp - now;
if (countdown > 0) {
this.scheduleNextTimerTask(currOffset, DELAY_FOR_A_WHILE);
ScheduleMessageService.this.updateOffset(this.delayLevel, currOffset);
return;
}
MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
if (msgExt == null) {
continue;
}
MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) {
log.error("[BUG] the real topic of schedule msg is {}, discard the msg. msg={}",
msgInner.getTopic(), msgInner);
continue;
}
boolean deliverSuc;
if (ScheduleMessageService.this.enableAsyncDeliver) {
deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
} else {
deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
}
if (!deliverSuc) {
this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
return;
}
}
} catch (Exception e) {
log.error("ScheduleMessageService, messageTimeup execute error, offset = {}", nextOffset, e);
} finally {
bufferCQ.release();
}
this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
}
public void scheduleNextTimerTask(long offset, long delay) {
ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(
this.delayLevel, offset), delay, TimeUnit.MILLISECONDS);
}
private boolean syncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
int sizePy) {
PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, false);
PutMessageResult result = resultProcess.get();
boolean sendStatus = result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK;
if (sendStatus) {
ScheduleMessageService.this.updateOffset(this.delayLevel, resultProcess.getNextOffset());
}
return sendStatus;
}
private boolean asyncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
int sizePy) {
Queue<PutResultProcess> processesQueue = ScheduleMessageService.this.deliverPendingTable.get(this.delayLevel);
//Flow Control
int currentPendingNum = processesQueue.size();
int maxPendingLimit = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
.getScheduleAsyncDeliverMaxPendingLimit();
if (currentPendingNum > maxPendingLimit) {
log.warn("Asynchronous deliver triggers flow control, " +
"currentPendingNum={}, maxPendingLimit={}", currentPendingNum, maxPendingLimit);
return false;
}
//Blocked
PutResultProcess firstProcess = processesQueue.peek();
if (firstProcess != null && firstProcess.need2Blocked()) {
log.warn("Asynchronous deliver block. info={}", firstProcess.toString());
return false;
}
PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, true);
processesQueue.add(resultProcess);
return true;
}
private PutResultProcess deliverMessage(MessageExtBrokerInner msgInner, String msgId, long offset,
long offsetPy, int sizePy, boolean autoResend) {
CompletableFuture<PutMessageResult> future =
ScheduleMessageService.this.writeMessageStore.asyncPutMessage(msgInner);
return new PutResultProcess()
.setTopic(msgInner.getTopic())
.setDelayLevel(this.delayLevel)
.setOffset(offset)
.setPhysicOffset(offsetPy)
.setPhysicSize(sizePy)
.setMsgId(msgId)
.setAutoResend(autoResend)
.setFuture(future)
.thenProcess();
}
}
public class HandlePutResultTask implements Runnable {
private final int delayLevel;
public HandlePutResultTask(int delayLevel) {
this.delayLevel = delayLevel;
}
@Override
public void run() {
LinkedBlockingQueue<PutResultProcess> pendingQueue =
ScheduleMessageService.this.deliverPendingTable.get(this.delayLevel);
PutResultProcess putResultProcess;
while ((putResultProcess = pendingQueue.peek()) != null) {
try {
switch (putResultProcess.getStatus()) {
case SUCCESS:
ScheduleMessageService.this.updateOffset(this.delayLevel, putResultProcess.getNextOffset());
pendingQueue.remove();
break;
case RUNNING:
break;
case EXCEPTION:
if (!isStarted()) {
log.warn("HandlePutResultTask shutdown, info={}", putResultProcess.toString());
return;
}
log.warn("putResultProcess error, info={}", putResultProcess.toString());
putResultProcess.onException();
break;
case SKIP:
log.warn("putResultProcess skip, info={}", putResultProcess.toString());
pendingQueue.remove();
break;
}
} catch (Exception e) {
log.error("HandlePutResultTask exception. info={}", putResultProcess.toString(), e);
putResultProcess.onException();
}
}
if (isStarted()) {
ScheduleMessageService.this.handleExecutorService
.schedule(new HandlePutResultTask(this.delayLevel), DELAY_FOR_A_SLEEP, TimeUnit.MILLISECONDS);
}
}
}
public class PutResultProcess {
private String topic;
private long offset;
private long physicOffset;
private int physicSize;
private int delayLevel;
private String msgId;
private boolean autoResend = false;
private CompletableFuture<PutMessageResult> future;
private volatile int resendCount = 0;
private volatile ProcessStatus status = ProcessStatus.RUNNING;
public PutResultProcess setTopic(String topic) {
this.topic = topic;
return this;
}
public PutResultProcess setOffset(long offset) {
this.offset = offset;
return this;
}
public PutResultProcess setPhysicOffset(long physicOffset) {
this.physicOffset = physicOffset;
return this;
}
public PutResultProcess setPhysicSize(int physicSize) {
this.physicSize = physicSize;
return this;
}
public PutResultProcess setDelayLevel(int delayLevel) {
this.delayLevel = delayLevel;
return this;
}
public PutResultProcess setMsgId(String msgId) {
this.msgId = msgId;
return this;
}
public PutResultProcess setAutoResend(boolean autoResend) {
this.autoResend = autoResend;
return this;
}
public PutResultProcess setFuture(CompletableFuture<PutMessageResult> future) {
this.future = future;
return this;
}
public String getTopic() {
return topic;
}
public long getOffset() {
return offset;
}
public long getNextOffset() {
return offset + 1;
}
public long getPhysicOffset() {
return physicOffset;
}
public int getPhysicSize() {
return physicSize;
}
public Integer getDelayLevel() {
return delayLevel;
}
public String getMsgId() {
return msgId;
}
public boolean isAutoResend() {
return autoResend;
}
public CompletableFuture<PutMessageResult> getFuture() {
return future;
}
public int getResendCount() {
return resendCount;
}
public PutResultProcess thenProcess() {
this.future.thenAccept(result -> {
this.handleResult(result);
});
this.future.exceptionally(e -> {
log.error("ScheduleMessageService put message exceptionally, info: {}",
PutResultProcess.this.toString(), e);
onException();
return null;
});
return this;
}
private void handleResult(PutMessageResult result) {
if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
onSuccess(result);
} else {
log.warn("ScheduleMessageService put message failed. info: {}.", result);
onException();
}
}
public void onSuccess(PutMessageResult result) {
this.status = ProcessStatus.SUCCESS;
if (ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig().isEnableScheduleMessageStats()) {
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, result.getAppendMessageResult().getMsgNum());
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, result.getAppendMessageResult().getWroteBytes());
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, result.getAppendMessageResult().getMsgNum());
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, result.getAppendMessageResult().getWroteBytes());
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutNums(this.topic, result.getAppendMessageResult().getMsgNum(), 1);
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutSize(this.topic, result.getAppendMessageResult().getWroteBytes());
ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incBrokerPutNums(result.getAppendMessageResult().getMsgNum());
}
}
public void onException() {
log.warn("ScheduleMessageService onException, info: {}", this.toString());
if (this.autoResend) {
this.resend();
} else {
this.status = ProcessStatus.SKIP;
}
}
public ProcessStatus getStatus() {
return this.status;
}
public PutMessageResult get() {
try {
return this.future.get();
} catch (InterruptedException | ExecutionException e) {
return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null);
}
}
private void resend() {
log.info("Resend message, info: {}", this.toString());
// Gradually increase the resend interval.
try {
Thread.sleep(Math.min(this.resendCount++ * 100, 60 * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(this.physicOffset, this.physicSize);
if (msgExt == null) {
log.warn("ScheduleMessageService resend not found message. info: {}", this.toString());
this.status = need2Skip() ? ProcessStatus.SKIP : ProcessStatus.EXCEPTION;
return;
}
MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
PutMessageResult result = ScheduleMessageService.this.writeMessageStore.putMessage(msgInner);
this.handleResult(result);
if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
log.info("Resend message success, info: {}", this.toString());
}
} catch (Exception e) {
this.status = ProcessStatus.EXCEPTION;
log.error("Resend message error, info: {}", this.toString(), e);
}
}
public boolean need2Blocked() {
int maxResendNum2Blocked = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
.getScheduleAsyncDeliverMaxResendNum2Blocked();
return this.resendCount > maxResendNum2Blocked;
}
public boolean need2Skip() {
int maxResendNum2Blocked = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
.getScheduleAsyncDeliverMaxResendNum2Blocked();
return this.resendCount > maxResendNum2Blocked * 2;
}
@Override
public String toString() {
return "PutResultProcess{" +
"topic='" + topic + '\'' +
", offset=" + offset +
", physicOffset=" + physicOffset +
", physicSize=" + physicSize +
", delayLevel=" + delayLevel +
", msgId='" + msgId + '\'' +
", autoResend=" + autoResend +
", resendCount=" + resendCount +
", status=" + status +
'}';
}
}
public enum ProcessStatus {
/**
* In process, the processing result has not yet been returned.
*/
RUNNING,
/**
* Put message success.
*/
SUCCESS,
/**
* Put message exception. When autoResend is true, the message will be resend.
*/
EXCEPTION,
/**
* Skip put message. When the message cannot be looked, the message will be skipped.
*/
SKIP,
}
}
store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java
-194
View File
@@ -1,194 +0,0 @@
/*
* 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;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.schedule.ScheduleMessageService;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.Assert;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class ScheduleMessageServiceTest {
private Random random = new Random();
@Test
public void testCorrectDelayOffset_whenInit() throws Exception {
ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable = null;
DefaultMessageStore defaultMessageStore = new DefaultMessageStore(buildMessageStoreConfig(),
new BrokerStatsManager("simpleTest", true), null, new BrokerConfig());
ScheduleMessageService scheduleMessageService = new ScheduleMessageService(defaultMessageStore);
scheduleMessageService.parseDelayLevel();
ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable1 = new ConcurrentHashMap<>();
for (int i = 1; i <= 18; i++) {
offsetTable1.put(i, random.nextLong());
}
Field field = scheduleMessageService.getClass().getDeclaredField("offsetTable");
field.setAccessible(true);
field.set(scheduleMessageService, offsetTable1);
String jsonStr = scheduleMessageService.encode();
scheduleMessageService.decode(jsonStr);
offsetTable = (ConcurrentMap<Integer, Long>) field.get(scheduleMessageService);
for (Map.Entry<Integer, Long> entry : offsetTable.entrySet()) {
assertEquals(entry.getValue(), offsetTable1.get(entry.getKey()));
}
scheduleMessageService.correctDelayOffset();
offsetTable = (ConcurrentMap<Integer, Long>) field.get(scheduleMessageService);
for (long offset : offsetTable.values()) {
assertEquals(0, offset);
}
}
private MessageStoreConfig buildMessageStoreConfig() throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
messageStoreConfig.setMaxHashSlotNum(10000);
messageStoreConfig.setMaxIndexNum(100 * 100);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setFlushIntervalConsumeQueue(1);
return messageStoreConfig;
}
@Test
public void testHandlePutResultTask() throws Exception {
DefaultMessageStore messageStore = mock(DefaultMessageStore.class);
MessageStoreConfig config = buildMessageStoreConfig();
config.setEnableScheduleMessageStats(false);
config.setEnableScheduleAsyncDeliver(true);
when(messageStore.getMessageStoreConfig()).thenReturn(config);
ScheduleMessageService scheduleMessageService = new ScheduleMessageService(messageStore);
scheduleMessageService.parseDelayLevel();
Field field = scheduleMessageService.getClass().getDeclaredField("deliverPendingTable");
field.setAccessible(true);
Map<Integer /* level */, LinkedBlockingQueue<ScheduleMessageService.PutResultProcess>> deliverPendingTable =
(Map<Integer, LinkedBlockingQueue<ScheduleMessageService.PutResultProcess>>) field.get(scheduleMessageService);
field = scheduleMessageService.getClass().getDeclaredField("offsetTable");
field.setAccessible(true);
ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
(ConcurrentMap<Integer /* level */, Long/* offset */>) field.get(scheduleMessageService);
for (int i = 1; i <= scheduleMessageService.getMaxDelayLevel(); i++) {
offsetTable.put(i, 0L);
}
int deliverThreadPoolNums = Runtime.getRuntime().availableProcessors();
ScheduledExecutorService handleExecutorService = new ScheduledThreadPoolExecutor(deliverThreadPoolNums,
new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
field = scheduleMessageService.getClass().getDeclaredField("handleExecutorService");
field.setAccessible(true);
field.set(scheduleMessageService, handleExecutorService);
field = scheduleMessageService.getClass().getDeclaredField("started");
field.setAccessible(true);
AtomicBoolean started = (AtomicBoolean) field.get(scheduleMessageService);
started.set(true);
for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
ScheduleMessageService.HandlePutResultTask handlePutResultTask = scheduleMessageService.new HandlePutResultTask(level);
handleExecutorService.schedule(handlePutResultTask, 10L, TimeUnit.MILLISECONDS);
}
MessageExt messageExt = new MessageExt();
messageExt.putUserProperty("init", "test");
messageExt.getProperties().put(MessageConst.PROPERTY_REAL_QUEUE_ID, "0");
when(messageStore.lookMessageByOffset(anyLong(), anyInt())).thenReturn(messageExt);
when(messageStore.putMessage(any())).thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK, null));
int msgNum = 100;
int totalMsgNum = msgNum * scheduleMessageService.getMaxDelayLevel();
List<CompletableFuture<PutMessageResult>> putMsgFutrueList = new ArrayList<>(totalMsgNum);
for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
for (int num = 0; num < msgNum; num++) {
CompletableFuture<PutMessageResult> future = new CompletableFuture<>();
ScheduleMessageService.PutResultProcess putResultProcess = scheduleMessageService.new PutResultProcess();
putResultProcess = putResultProcess
.setOffset(num)
.setAutoResend(true)
.setFuture(future)
.thenProcess();
deliverPendingTable.get(level).add(putResultProcess);
putMsgFutrueList.add(future);
}
}
Collections.shuffle(putMsgFutrueList);
Random random = new Random();
for (CompletableFuture<PutMessageResult> future : putMsgFutrueList) {
PutMessageStatus status;
if (random.nextInt(1000) % 2 == 0) {
status = PutMessageStatus.PUT_OK;
} else {
status = PutMessageStatus.OS_PAGECACHE_BUSY;
}
if (random.nextInt(1000) % 2 == 0) {
PutMessageResult result = new PutMessageResult(status, null);
future.complete(result);
} else {
future.completeExceptionally(new Throwable("complete exceptionally"));
}
}
Thread.sleep(1000);
for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
Assert.assertEquals(0, deliverPendingTable.get(level).size());
Assert.assertEquals(msgNum, offsetTable.get(level).longValue());
}
scheduleMessageService.shutdown();
}
}
store/src/test/java/org/apache/rocketmq/store/schedule/ScheduleMessageServiceTest.java
-235
View File
@@ -1,235 +0,0 @@
/*
* 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.schedule;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.*;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import static org.apache.rocketmq.store.stats.BrokerStatsManager.BROKER_PUT_NUMS;
import static org.apache.rocketmq.store.stats.BrokerStatsManager.TOPIC_PUT_NUMS;
import static org.apache.rocketmq.store.stats.BrokerStatsManager.TOPIC_PUT_SIZE;
import static org.assertj.core.api.Assertions.assertThat;
public class ScheduleMessageServiceTest {
/**
* t
* defaultMessageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
*/
String testMessageDelayLevel = "5s 8s";
/**
* choose delay level
*/
int delayLevel = 2;
private static final String storePath = System.getProperty("user.home") + File.separator + "schedule_test#" + UUID.randomUUID();
private static final int commitLogFileSize = 1024;
private static final int cqFileSize = 10;
private static final int cqExtFileSize = 10 * (ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE + 64);
private static SocketAddress bornHost;
private static SocketAddress storeHost;
DefaultMessageStore messageStore;
MessageStoreConfig messageStoreConfig;
BrokerConfig brokerConfig;
ScheduleMessageService scheduleMessageService;
static String sendMessage = " ------- schedule message test -------";
static String topic = "schedule_topic_test";
static String messageGroup = "delayGroupTest";
static {
try {
bornHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123);
} catch (UnknownHostException e) {
e.printStackTrace();
}
try {
storeHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
@Before
public void init() throws Exception {
messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMessageDelayLevel(testMessageDelayLevel);
messageStoreConfig.setMappedFileSizeCommitLog(commitLogFileSize);
messageStoreConfig.setMappedFileSizeConsumeQueue(cqFileSize);
messageStoreConfig.setMappedFileSizeConsumeQueueExt(cqExtFileSize);
messageStoreConfig.setMessageIndexEnable(false);
messageStoreConfig.setEnableConsumeQueueExt(true);
messageStoreConfig.setStorePathRootDir(storePath);
messageStoreConfig.setStorePathCommitLog(storePath + File.separator + "commitlog");
brokerConfig = new BrokerConfig();
BrokerStatsManager manager = new BrokerStatsManager(brokerConfig.getBrokerClusterName(), brokerConfig.isEnableDetailStat());
messageStore = new DefaultMessageStore(messageStoreConfig, manager, new MyMessageArrivingListener(), new BrokerConfig());
assertThat(messageStore.load()).isTrue();
messageStore.start();
scheduleMessageService = messageStore.getScheduleMessageService();
}
@Test
public void deliverDelayedMessageTimerTaskTest() throws Exception {
assertThat(messageStore.getMessageStoreConfig().isEnableScheduleMessageStats()).isTrue();
assertThat(messageStore.getBrokerStatsManager().getStatsItem(TOPIC_PUT_NUMS, topic)).isNull();
MessageExtBrokerInner msg = buildMessage();
int realQueueId = msg.getQueueId();
// set delayLevel,and send delay message
msg.setDelayTimeLevel(delayLevel);
PutMessageResult result = messageStore.putMessage(msg);
assertThat(result.isOk()).isTrue();
// make sure consumerQueue offset = commitLog offset
StoreTestUtil.waitCommitLogReput(messageStore);
// consumer message
int delayQueueId = ScheduleMessageService.delayLevel2QueueId(delayLevel);
assertThat(delayQueueId).isEqualTo(delayLevel - 1);
Long offset = result.getAppendMessageResult().getLogicsOffset();
// now, no message in queue,must wait > delayTime
GetMessageResult messageResult = getMessage(realQueueId, offset);
assertThat(messageResult.getStatus()).isEqualTo(GetMessageStatus.NO_MESSAGE_IN_QUEUE);
// timer run maybe delay, then consumer message again
// and wait offsetTable
TimeUnit.SECONDS.sleep(10);
scheduleMessageService.buildRunningStats(new HashMap<String, String>());
messageResult = getMessage(realQueueId, offset);
// now,found the message
assertThat(messageResult.getStatus()).isEqualTo(GetMessageStatus.FOUND);
// get the stats change
assertThat(messageStore.getBrokerStatsManager().getStatsItem(BROKER_PUT_NUMS, brokerConfig.getBrokerClusterName()).getValue().sum()).isEqualTo(1);
assertThat(messageStore.getBrokerStatsManager().getStatsItem(TOPIC_PUT_NUMS, topic).getValue().sum()).isEqualTo(1L);
assertThat(messageStore.getBrokerStatsManager().getStatsItem(TOPIC_PUT_SIZE, topic).getValue().sum()).isEqualTo(messageResult.getBufferTotalSize());
// get the message body
ByteBuffer byteBuffer = ByteBuffer.allocate(messageResult.getBufferTotalSize());
List<ByteBuffer> byteBufferList = messageResult.getMessageBufferList();
for (ByteBuffer bb : byteBufferList) {
byteBuffer.put(bb);
}
// warp and decode the message
byteBuffer = ByteBuffer.wrap(byteBuffer.array());
List<MessageExt> msgList = MessageDecoder.decodes(byteBuffer);
String retryMsg = new String(msgList.get(0).getBody());
assertThat(sendMessage).isEqualTo(retryMsg);
// method will wait 10s,so I run it by myself
scheduleMessageService.persist();
// add mapFile release
messageResult.release();
}
/**
* add some [error/no use] code test
*/
@Test
public void otherTest() {
// the method no use ,why need ?
int queueId = ScheduleMessageService.queueId2DelayLevel(delayLevel);
assertThat(queueId).isEqualTo(delayLevel + 1);
// error delayLevelTest
Long time = scheduleMessageService.computeDeliverTimestamp(999, 0);
assertThat(time).isEqualTo(1000);
// just decode
scheduleMessageService.decode(new DelayOffsetSerializeWrapper().toJson());
}
private GetMessageResult getMessage(int queueId, Long offset) {
return messageStore.getMessage(messageGroup, topic,
queueId, offset, 1, null);
}
@After
public void shutdown() throws InterruptedException {
messageStore.shutdown();
messageStore.destroy();
File file = new File(messageStoreConfig.getStorePathRootDir());
UtilAll.deleteFile(file);
}
public MessageExtBrokerInner buildMessage() {
byte[] msgBody = sendMessage.getBytes();
MessageExtBrokerInner msg = new MessageExtBrokerInner();
msg.setTopic(topic);
msg.setTags("schedule_tag");
msg.setKeys("schedule_key");
msg.setBody(msgBody);
msg.setSysFlag(0);
msg.setBornTimestamp(System.currentTimeMillis());
msg.setStoreHost(storeHost);
msg.setBornHost(bornHost);
return msg;
}
private class MyMessageArrivingListener implements MessageArrivingListener {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
byte[] filterBitMap, Map<String, String> properties) {
}
}
}