-
Notifications
You must be signed in to change notification settings - Fork 11.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add logic of resuming half message check
- Loading branch information
chengxiangwang
committed
Jan 14, 2019
1 parent
8a2344b
commit 36aff5b
Showing
10 changed files
with
848 additions
and
241 deletions.
There are no files selected for viewing
725 changes: 485 additions & 240 deletions
725
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
Large diffs are not rendered by default.
Oops, something went wrong.
155 changes: 155 additions & 0 deletions
155
broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| /* | ||
| * 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.processor; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.mockito.ArgumentMatchers.any; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import io.netty.channel.ChannelHandlerContext; | ||
| import java.net.SocketAddress; | ||
| import java.net.UnknownHostException; | ||
| import java.nio.ByteBuffer; | ||
| import org.apache.rocketmq.broker.BrokerController; | ||
| import org.apache.rocketmq.common.BrokerConfig; | ||
| 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.message.MessageId; | ||
| import org.apache.rocketmq.common.protocol.RequestCode; | ||
| import org.apache.rocketmq.common.protocol.ResponseCode; | ||
| import org.apache.rocketmq.common.protocol.header.ResumeCheckHalfMessageRequestHeader; | ||
| import org.apache.rocketmq.remoting.exception.RemotingCommandException; | ||
| import org.apache.rocketmq.remoting.netty.NettyClientConfig; | ||
| import org.apache.rocketmq.remoting.netty.NettyServerConfig; | ||
| import org.apache.rocketmq.remoting.protocol.RemotingCommand; | ||
| import org.apache.rocketmq.store.AppendMessageResult; | ||
| import org.apache.rocketmq.store.AppendMessageStatus; | ||
| import org.apache.rocketmq.store.MappedFile; | ||
| 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.SelectMappedBufferResult; | ||
| import org.apache.rocketmq.store.config.MessageStoreConfig; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
| import org.junit.runner.RunWith; | ||
| import org.mockito.Mock; | ||
| import org.mockito.Spy; | ||
| import org.powermock.api.mockito.PowerMockito; | ||
| import org.powermock.core.classloader.annotations.PrepareForTest; | ||
| import org.powermock.modules.junit4.PowerMockRunner; | ||
|
|
||
| @RunWith(PowerMockRunner.class) | ||
| @PrepareForTest({MessageDecoder.class}) | ||
| public class AdminBrokerProcessorTest { | ||
|
|
||
| private AdminBrokerProcessor adminBrokerProcessor; | ||
|
|
||
| @Mock | ||
| private ChannelHandlerContext handlerContext; | ||
|
|
||
| @Spy | ||
| private BrokerController | ||
| brokerController = new BrokerController(new BrokerConfig(), new NettyServerConfig(), new NettyClientConfig(), | ||
| new MessageStoreConfig()); | ||
|
|
||
| @Mock | ||
| private MessageStore messageStore; | ||
|
|
||
|
|
||
| @Before | ||
| public void init() { | ||
| brokerController.setMessageStore(messageStore); | ||
| adminBrokerProcessor = new AdminBrokerProcessor(brokerController); | ||
| } | ||
|
|
||
| @Test | ||
| public void testProcessRequest_success() throws RemotingCommandException, UnknownHostException { | ||
| RemotingCommand request = createResumeCheckHalfMessageCommand(); | ||
| when(messageStore.selectOneMessageByOffset(any(Long.class))).thenReturn(createSelectMappedBufferResult()); | ||
| PowerMockito.mockStatic(MessageDecoder.class); | ||
| PowerMockito.when(MessageDecoder.decode(any(ByteBuffer.class))).thenReturn(createDefaultMessageExt()); | ||
| PowerMockito.when(MessageDecoder.decodeMessageId(any(String.class))).thenReturn(createMessageId()); | ||
| when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult | ||
| (PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))); | ||
| RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request); | ||
| assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS); | ||
| } | ||
|
|
||
| @Test | ||
| public void testProcessRequest_fail() throws RemotingCommandException, UnknownHostException { | ||
| RemotingCommand request = createResumeCheckHalfMessageCommand(); | ||
| when(messageStore.selectOneMessageByOffset(any(Long.class))).thenReturn(createSelectMappedBufferResult()); | ||
| PowerMockito.mockStatic(MessageDecoder.class); | ||
| PowerMockito.when(MessageDecoder.decode(any(ByteBuffer.class))).thenReturn(createDefaultMessageExt()); | ||
| PowerMockito.when(MessageDecoder.decodeMessageId(any(String.class))).thenReturn(createMessageId()); | ||
| when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult | ||
| (PutMessageStatus.UNKNOWN_ERROR, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR))); | ||
| RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request); | ||
| assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR); | ||
| } | ||
|
|
||
| @Test | ||
| public void testProcessRequest_exception() throws RemotingCommandException, UnknownHostException { | ||
| RemotingCommand request = createResumeCheckHalfMessageCommand(); | ||
| when(messageStore.selectOneMessageByOffset(any(Long.class))).thenReturn(createSelectMappedBufferResult()); | ||
| PowerMockito.mockStatic(MessageDecoder.class); | ||
| PowerMockito.when(MessageDecoder.decode(any(ByteBuffer.class))).thenReturn(createDefaultMessageExt()); | ||
| PowerMockito.when(MessageDecoder.decodeMessageId(any(String.class))).thenThrow(new UnknownHostException()); | ||
| when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult | ||
| (PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))); | ||
| RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request); | ||
| assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR); | ||
| } | ||
|
|
||
| private MessageExt createDefaultMessageExt() { | ||
| MessageExt messageExt = new MessageExt(); | ||
| messageExt.setMsgId("12345678"); | ||
| messageExt.setQueueId(0); | ||
| messageExt.setCommitLogOffset(123456789L); | ||
| messageExt.setQueueOffset(1234); | ||
| MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_REAL_QUEUE_ID, "0"); | ||
| MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_REAL_TOPIC, "testTopic"); | ||
| MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES, "15"); | ||
| return messageExt; | ||
| } | ||
|
|
||
| private MessageId createMessageId() { | ||
| return new MessageId(new SocketAddress() { | ||
| }, 0); | ||
| } | ||
|
|
||
| private SelectMappedBufferResult createSelectMappedBufferResult(){ | ||
| SelectMappedBufferResult result = new SelectMappedBufferResult(0, ByteBuffer.allocate(1024) ,0, new MappedFile()); | ||
| return result; | ||
| } | ||
| private ResumeCheckHalfMessageRequestHeader createResumeCheckHalfMessageRequestHeader() { | ||
| ResumeCheckHalfMessageRequestHeader header = new ResumeCheckHalfMessageRequestHeader(); | ||
| header.setMsgId("12345678"); | ||
| return header; | ||
| } | ||
|
|
||
| private RemotingCommand createResumeCheckHalfMessageCommand() { | ||
| ResumeCheckHalfMessageRequestHeader header = createResumeCheckHalfMessageRequestHeader(); | ||
| RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.RESUME_CHECK_HALF_MESSAGE, header); | ||
| request.makeCustomHeaderToNet(); | ||
| return request; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
.../src/main/java/org/apache/rocketmq/common/protocol/body/ResumeCheckHalfMessageResult.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.rocketmq.common.protocol.body; | ||
|
|
||
| import org.apache.rocketmq.remoting.protocol.RemotingSerializable; | ||
|
|
||
| public class ResumeCheckHalfMessageResult extends RemotingSerializable { | ||
|
|
||
| private ResumeResult resumeResult; | ||
| private String remark; | ||
| private long spentTimeMills; | ||
|
|
||
|
|
||
| public ResumeResult getResumeResult() { | ||
| return resumeResult; | ||
| } | ||
|
|
||
| public void setResumeResult(ResumeResult resumeResult) { | ||
| this.resumeResult = resumeResult; | ||
| } | ||
|
|
||
| public String getRemark() { | ||
| return remark; | ||
| } | ||
|
|
||
| public void setRemark(String remark) { | ||
| this.remark = remark; | ||
| } | ||
|
|
||
|
|
||
| public long getSpentTimeMills() { | ||
| return spentTimeMills; | ||
| } | ||
|
|
||
| public void setSpentTimeMills(long spentTimeMills) { | ||
| this.spentTimeMills = spentTimeMills; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "ConsumeMessageDirectlyResult [consumeResult=" + resumeResult + ", remark=" + remark | ||
| + ", spentTimeMills=" | ||
| + spentTimeMills + "]"; | ||
| } | ||
| } |
23 changes: 23 additions & 0 deletions
23
common/src/main/java/org/apache/rocketmq/common/protocol/body/ResumeResult.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.rocketmq.common.protocol.body; | ||
|
|
||
| public enum ResumeResult { | ||
| RESUME_SUCCESS, | ||
| RESUME_THROW_EXCEPTION | ||
| } |
45 changes: 45 additions & 0 deletions
45
.../java/org/apache/rocketmq/common/protocol/header/ResumeCheckHalfMessageRequestHeader.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.rocketmq.common.protocol.header; | ||
|
|
||
| import org.apache.rocketmq.remoting.CommandCustomHeader; | ||
| import org.apache.rocketmq.remoting.annotation.CFNullable; | ||
| import org.apache.rocketmq.remoting.exception.RemotingCommandException; | ||
|
|
||
| public class ResumeCheckHalfMessageRequestHeader implements CommandCustomHeader { | ||
| @CFNullable | ||
| private String msgId; | ||
|
|
||
| @Override | ||
| public void checkFields() throws RemotingCommandException { | ||
|
|
||
| } | ||
|
|
||
| public String getMsgId() { | ||
| return msgId; | ||
| } | ||
|
|
||
| public void setMsgId(String msgId) { | ||
| this.msgId = msgId; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "ResumeCheckHalfMessageRequestHeader [msgId=" + msgId + "]"; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.