Add more logs to ClusterMemoryManager
@@ -241,8 +241,29 @@ public class ClusterMemoryManager
if (!(lowMemoryKiller instanceof NoneLowMemoryKiller) &&
outOfMemory &&
!queryKilled &&
- nanosSince(lastTimeNotOutOfMemory).compareTo(killOnOutOfMemoryDelay) > 0 &&
- isLastKilledQueryGone()) {
+ nanosSince(lastTimeNotOutOfMemory).compareTo(killOnOutOfMemoryDelay) > 0) {
+ boolean lastKilledQueryGone = isLastKilledQueryGone();
+ if (lastKilledQueryGone) {
+ callOomKiller(queries);
+ }
+ else {
+ log.debug("Last killed query is still not gone: %s", lastKilledQuery);
+ }
+ }
+
+ Map<MemoryPoolId, Integer> countByPool = new HashMap<>();
+ for (QueryExecution query : queries) {
+ MemoryPoolId id = query.getMemoryPool().getId();
+ countByPool.put(id, countByPool.getOrDefault(id, 0) + 1);
+ }
+
+ updatePools(countByPool);
+
+ updateNodes(updateAssignments(queries));
+ }
+
+ private synchronized void callOomKiller(Iterable<QueryExecution> queries)
+ {
List<QueryMemoryInfo> queryMemoryInfoList = Streams.stream(queries)
.map(this::createQueryMemoryInfo)
.collect(toImmutableList());
@@ -253,6 +274,7 @@ public class ClusterMemoryManager
.collect(toImmutableList());
Optional<QueryId> chosenQueryId = lowMemoryKiller.chooseQueryToKill(queryMemoryInfoList, nodeMemoryInfos);
if (chosenQueryId.isPresent()) {
+ log.debug("Low memory killer chose %s", chosenQueryId.get());
Optional<QueryExecution> chosenQuery = Streams.stream(queries).filter(query -> chosenQueryId.get().equals(query.getQueryId())).collect(toOptional());
if (chosenQuery.isPresent()) {
// See comments in isLastKilledQueryGone for why chosenQuery might be absent.
@@ -264,17 +286,6 @@ public class ClusterMemoryManager
}
}
- Map<MemoryPoolId, Integer> countByPool = new HashMap<>();
- for (QueryExecution query : queries) {
- MemoryPoolId id = query.getMemoryPool().getId();
- countByPool.put(id, countByPool.getOrDefault(id, 0) + 1);
- }
-
- updatePools(countByPool);
-
- updateNodes(updateAssignments(queries));
- }
-
@GuardedBy("this")
private boolean isLastKilledQueryGone()
{
