mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-21 05:44:03 +08:00
fix(store): close all consume queue file handles on ConsumeQueueStore shutdown (#10060)
* fix(store): close all consume queue file handles on ConsumeQueueStore shutdown * remove implementation --------- Co-authored-by: RongtongJin <user@example.com>
This commit is contained in:
@@ -37,12 +37,11 @@ import org.apache.rocketmq.store.logfile.MappedFile;
|
||||
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
|
||||
import org.apache.rocketmq.store.queue.ConsumeQueueStore;
|
||||
import org.apache.rocketmq.store.queue.CqUnit;
|
||||
import org.apache.rocketmq.store.queue.FileQueueLifeCycle;
|
||||
import org.apache.rocketmq.store.queue.MultiDispatchUtils;
|
||||
import org.apache.rocketmq.store.queue.QueueOffsetOperator;
|
||||
import org.apache.rocketmq.store.queue.ReferredIterator;
|
||||
|
||||
public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
|
||||
public class ConsumeQueue implements ConsumeQueueInterface {
|
||||
private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
|
||||
|
||||
/**
|
||||
@@ -1236,4 +1235,10 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
|
||||
|
||||
flush(0);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean shutdown() {
|
||||
this.mappedFileQueue.cleanResourcesAll();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1200,4 +1200,9 @@ public class BatchConsumeQueue implements ConsumeQueueInterface {
|
||||
public void initializeWithOffset(long offset, long minPhyOffset) {
|
||||
// not support now
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean shutdown() {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -171,6 +171,7 @@ public class ConsumeQueueStore extends AbstractConsumeQueueStore {
|
||||
log.error("Failed to flush all consume queues", e);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -864,4 +865,5 @@ public class ConsumeQueueStore extends AbstractConsumeQueueStore {
|
||||
return messageStore.getBrokerConfig().getIdentifier() + CleanConsumeQueueService.class.getSimpleName();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -78,7 +78,10 @@ public interface FileQueueLifeCycle extends Swappable {
|
||||
|
||||
/**
|
||||
* Does the first file exist?
|
||||
*
|
||||
* @return true if it exists
|
||||
*/
|
||||
boolean isFirstFileExist();
|
||||
|
||||
boolean shutdown();
|
||||
}
|
||||
|
||||
@@ -507,4 +507,9 @@ public class RocksDBConsumeQueue implements ConsumeQueueInterface {
|
||||
ERROR_LOG.error("RocksDBConsumeQueue initializeWithOffset Failed. topic={}, queueId={}, offset={}", topic, queueId, offset, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean shutdown() {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user