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

[ISSUE #6525]Make ConsumeQueueInterface extends from FileQueueLifeCycle (#6534)

Browse Source 
* [#6525] Make ConsumeQueueInterface extends from FileQueueLifeCycle

* [apache#6525] forced typecast is removed

* [#6525] remove all the forced convert
This commit is contained in:
Abhijeet Mishra
2023-04-15 15:51:03 +05:30
committed by GitHub
parent d3a8fe0c5b
commit d1b14b0ff9
3 changed files with 4 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java
+1 -1
View File
@@ -41,7 +41,7 @@ import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.logfile.MappedFile;
public class BatchConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
public class BatchConsumeQueue implements ConsumeQueueInterface {
protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
/**
store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueInterface.java
+1 -1
View File
@@ -22,7 +22,7 @@ import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.MessageFilter;
public interface ConsumeQueueInterface {
public interface ConsumeQueueInterface extends FileQueueLifeCycle {
/**
* Get the topic name
* @return the topic this cq belongs to.
store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStore.java
+2 -2
View File
@@ -79,7 +79,7 @@ public class ConsumeQueueStore {
}
private FileQueueLifeCycle getLifeCycle(String topic, int queueId) {
return (FileQueueLifeCycle) findOrCreateConsumeQueue(topic, queueId);
return findOrCreateConsumeQueue(topic, queueId);
}
public long rollNextFile(ConsumeQueueInterface consumeQueue, final long offset) {
@@ -220,7 +220,7 @@ public class ConsumeQueueStore {
FutureTask<Boolean> futureTask = new FutureTask<>(() -> {
boolean ret = true;
try {
((FileQueueLifeCycle) logic).recover();
logic.recover();
} catch (Throwable e) {
ret = false;
log.error("Exception occurs while recover consume queue concurrently, " +