clean until round clean storage size is zero
@@ -125,6 +125,11 @@ public class StoreCleanManager extends Service {
if (LOG.isDebugEnabled()) {
LOG.info("Start scheduled StoreCleaningStrategy task use class: <{}>!!!", brokerStoreConfig.getCleanStrategyClass());
}
+ long roundTotalDeleteStoreSize=0;
+ long roundDeleteStoreSize;
+ long startMs=SystemClock.now();
+ do {
+ roundDeleteStoreSize=0;
List<TopicConfig> topicConfigs = clusterManager.getTopics();
if (topicConfigs != null && topicConfigs.size() > 0) {
for (TopicConfig topicConfig : topicConfigs) {
@@ -151,7 +156,7 @@ public class StoreCleanManager extends Service {
LOG.info("Begin store clean topic: <{}>, partition group: <{}>, partition ack map: <{}>",
topicConfig.getName().getFullName(), partitionGroup.getGroup(), partitionAckMap);
}
- cleaningStrategy.deleteIfNeeded(storeService.getStore(topicConfig.getName().getFullName(), partitionGroup.getGroup()), partitionAckMap);
+ roundDeleteStoreSize = cleaningStrategy.deleteIfNeeded(storeService.getStore(topicConfig.getName().getFullName(), partitionGroup.getGroup()), partitionAckMap);
}
}
} catch (Throwable t) {
@@ -161,5 +166,10 @@ public class StoreCleanManager extends Service {
}
}
}
+ roundTotalDeleteStoreSize+=roundDeleteStoreSize;
+ }while (roundDeleteStoreSize>0);
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Round total clean storage size {},elapsed time {}ms",roundTotalDeleteStoreSize,SystemClock.now()-startMs);
+ }
}
}
