You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2018/06/04 20:31:33 UTC
[1/2] hive git commit: HIVE-19663 : refactor LLAP IO report
generation (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master 2d3faead7 -> 43e331e35
HIVE-19663 : refactor LLAP IO report generation (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/43e331e3
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/43e331e3
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/43e331e3
Branch: refs/heads/master
Commit: 43e331e35ec17479c204ab6b310e49d68cb4a2e9
Parents: 283e51d
Author: sergey <se...@apache.org>
Authored: Mon Jun 4 11:28:50 2018 -0700
Committer: sergey <se...@apache.org>
Committed: Mon Jun 4 13:31:22 2018 -0700
----------------------------------------------------------------------
.../hadoop/hive/llap/cache/BuddyAllocator.java | 9 +--
.../hive/llap/cache/CacheContentsTracker.java | 11 ----
.../hive/llap/cache/EvictionDispatcher.java | 25 +-------
.../hadoop/hive/llap/cache/LlapIoDebugDump.java | 23 +++++++
.../hive/llap/cache/LlapOomDebugDump.java | 24 --------
.../hadoop/hive/llap/cache/LowLevelCache.java | 2 +-
.../hive/llap/cache/LowLevelCacheImpl.java | 28 +--------
.../llap/cache/LowLevelCacheMemoryManager.java | 12 ----
.../hive/llap/cache/LowLevelCachePolicy.java | 3 +-
.../llap/cache/LowLevelFifoCachePolicy.java | 28 ---------
.../llap/cache/LowLevelLrfuCachePolicy.java | 18 ------
.../hadoop/hive/llap/cache/MemoryManager.java | 2 +-
.../hive/llap/cache/SerDeLowLevelCacheImpl.java | 20 +------
.../hive/llap/cache/SimpleBufferManager.java | 10 ----
.../hive/llap/io/api/impl/LlapIoImpl.java | 63 ++++++++++++++------
.../hive/llap/io/metadata/MetadataCache.java | 10 +---
.../hive/llap/cache/TestBuddyAllocator.java | 9 ---
.../hive/llap/cache/TestLowLevelCacheImpl.java | 6 +-
.../hive/llap/cache/TestOrcMetadataCache.java | 15 +----
19 files changed, 78 insertions(+), 240 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java
index f4a549c..fcfc22a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/BuddyAllocator.java
@@ -47,7 +47,7 @@ import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
public final class BuddyAllocator
- implements EvictionAwareAllocator, BuddyAllocatorMXBean, LlapOomDebugDump {
+ implements EvictionAwareAllocator, BuddyAllocatorMXBean, LlapIoDebugDump {
private final Arena[] arenas;
private final AtomicInteger allocatedArenas = new AtomicInteger(0);
@@ -653,7 +653,6 @@ public final class BuddyAllocator
*/
@Override
public void debugDumpShort(StringBuilder sb) {
- memoryManager.debugDumpShort(sb);
sb.append("\nDefrag counters: ");
for (int i = 0; i < defragCounters.length; ++i) {
sb.append(defragCounters[i].get()).append(", ");
@@ -1558,12 +1557,6 @@ public final class BuddyAllocator
return sb.toString();
}
- @Override
- public String debugDumpForOom() {
- return "\nALLOCATOR STATE:\n" + debugDumpForOomInternal()
- + "\nPARENT STATE:\n" + memoryManager.debugDumpForOom();
- }
-
private String debugDumpForOomInternal() {
StringBuilder sb = new StringBuilder();
for (Arena a : arenas) {
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/CacheContentsTracker.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/CacheContentsTracker.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/CacheContentsTracker.java
index 6a361fa..64c0125 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/CacheContentsTracker.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/CacheContentsTracker.java
@@ -179,11 +179,6 @@ public class CacheContentsTracker implements LowLevelCachePolicy, EvictionListen
}
@Override
- public void setParentDebugDumper(LlapOomDebugDump dumper) {
- realPolicy.setParentDebugDumper(dumper);
- }
-
- @Override
public long purge() {
return realPolicy.purge();
}
@@ -195,11 +190,6 @@ public class CacheContentsTracker implements LowLevelCachePolicy, EvictionListen
}
@Override
- public String debugDumpForOom() {
- return realPolicy.debugDumpForOom();
- }
-
- @Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nCache state: ");
for (TagState state : tagInfo.values()) {
@@ -209,7 +199,6 @@ public class CacheContentsTracker implements LowLevelCachePolicy, EvictionListen
.append(state.maxSize);
}
}
- realPolicy.debugDumpShort(sb);
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
index 10a4bfb..6f5ac7f 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/EvictionDispatcher.java
@@ -25,7 +25,7 @@ import org.apache.hadoop.hive.llap.io.metadata.MetadataCache.LlapMetadataBuffer;
/**
* Eviction dispatcher - uses double dispatch to route eviction notifications to correct caches.
*/
-public final class EvictionDispatcher implements EvictionListener, LlapOomDebugDump {
+public final class EvictionDispatcher implements EvictionListener {
private final LowLevelCache dataCache;
private final SerDeLowLevelCacheImpl serdeCache;
private final MetadataCache metadataCache;
@@ -63,27 +63,4 @@ public final class EvictionDispatcher implements EvictionListener, LlapOomDebugD
public void notifyEvicted(OrcFileEstimateErrors buffer) {
metadataCache.notifyEvicted(buffer);
}
-
- @Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder(dataCache.debugDumpForOom());
- if (serdeCache != null) {
- sb.append(serdeCache.debugDumpForOom());
- }
- if (metadataCache != null) {
- sb.append(metadataCache.debugDumpForOom());
- }
- return sb.toString();
- }
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- dataCache.debugDumpShort(sb);
- if (serdeCache != null) {
- serdeCache.debugDumpShort(sb);
- }
- if (metadataCache != null) {
- metadataCache.debugDumpShort(sb);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapIoDebugDump.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapIoDebugDump.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapIoDebugDump.java
new file mode 100644
index 0000000..105b354
--- /dev/null
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapIoDebugDump.java
@@ -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.hadoop.hive.llap.cache;
+
+public interface LlapIoDebugDump {
+ void debugDumpShort(StringBuilder sb);
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapOomDebugDump.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapOomDebugDump.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapOomDebugDump.java
deleted file mode 100644
index 29ddf5c..0000000
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LlapOomDebugDump.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.hadoop.hive.llap.cache;
-
-public interface LlapOomDebugDump {
- String debugDumpForOom();
- void debugDumpShort(StringBuilder sb);
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
index af1b699..9591e48 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCache.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.hive.common.io.DataCache.BooleanRef;
import org.apache.hadoop.hive.common.io.DataCache.DiskRangeListFactory;
import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
-public interface LowLevelCache extends LlapOomDebugDump {
+public interface LowLevelCache {
public enum Priority {
NORMAL,
HIGH
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java
index 5e102d9..53bdc2a 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheImpl.java
@@ -41,7 +41,7 @@ import org.apache.hive.common.util.Ref;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
-public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, LlapOomDebugDump {
+public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, LlapIoDebugDump {
private static final int DEFAULT_CLEANUP_INTERVAL = 600;
private final Allocator allocator;
private final AtomicInteger newEvictions = new AtomicInteger(0);
@@ -448,32 +448,6 @@ public class LowLevelCacheImpl implements LowLevelCache, BufferUsageManager, Lla
}
@Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder("File cache state ");
- for (Map.Entry<Object, FileCache<ConcurrentSkipListMap<Long, LlapDataBuffer>>> e :
- cache.entrySet()) {
- if (!e.getValue().incRef()) continue;
- try {
- sb.append("\n file " + e.getKey());
- for (Map.Entry<Long, LlapDataBuffer> e2 : e.getValue().getCache().entrySet()) {
- if (e2.getValue().incRef() < 0) continue;
- try {
- sb.append("\n [").append(e2.getKey()).append(", ")
- .append(e2.getKey() + e2.getValue().declaredCachedLength)
- .append(") => ").append(e2.getValue().toString())
- .append(" alloc ").append(e2.getValue().byteBuffer.position());
- } finally {
- e2.getValue().decRef();
- }
- }
- } finally {
- e.getValue().decRef();
- }
- }
- return sb.toString();
- }
-
- @Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nORC cache state ");
int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0;
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
index 31cd45f..60d56d6 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCacheMemoryManager.java
@@ -130,18 +130,6 @@ public class LowLevelCacheMemoryManager implements MemoryManager {
}
@Override
- public String debugDumpForOom() {
- if (evictor == null) return null;
- return "\ncache state\n" + evictor.debugDumpForOom();
- }
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- if (evictor == null) return;
- evictor.debugDumpShort(sb);
- }
-
- @Override
public void updateMaxSize(long maxSize) {
this.maxSize = maxSize;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
index 3323636..45829dd 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelCachePolicy.java
@@ -20,12 +20,11 @@ package org.apache.hadoop.hive.llap.cache;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
-public interface LowLevelCachePolicy extends LlapOomDebugDump {
+public interface LowLevelCachePolicy extends LlapIoDebugDump {
void cache(LlapCacheableBuffer buffer, Priority priority);
void notifyLock(LlapCacheableBuffer buffer);
void notifyUnlock(LlapCacheableBuffer buffer);
long evictSomeBlocks(long memoryToReserve);
void setEvictionListener(EvictionListener listener);
- void setParentDebugDumper(LlapOomDebugDump dumper);
long purge();
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
index f7f80a8..bdc6721 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelFifoCachePolicy.java
@@ -32,7 +32,6 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
private final Lock lock = new ReentrantLock();
private final LinkedList<LlapCacheableBuffer> buffers;
private EvictionListener evictionListener;
- private LlapOomDebugDump parentDebugDump;
public LowLevelFifoCachePolicy() {
LlapIoImpl.LOG.info("FIFO cache policy");
@@ -66,11 +65,6 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
}
@Override
- public void setParentDebugDumper(LlapOomDebugDump dumper) {
- this.parentDebugDump = dumper;
- }
-
- @Override
public long purge() {
long evicted = evictSomeBlocks(Long.MAX_VALUE);
LlapIoImpl.LOG.info("PURGE: evicted {} from FIFO policy", LlapUtil.humanReadableByteCount(evicted));
@@ -105,25 +99,6 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
}
@Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder("FIFO eviction list: ");
- lock.lock();
- try {
- sb.append(buffers.size()).append(" elements): ");
- Iterator<LlapCacheableBuffer> iter = buffers.iterator();
- while (iter.hasNext()) {
- sb.append(iter.next().toStringForCache()).append(",\n");
- }
- } finally {
- lock.unlock();
- }
- if (parentDebugDump != null) {
- sb.append("\n").append(parentDebugDump.debugDumpForOom());
- }
- return sb.toString();
- }
-
- @Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nFIFO eviction list: ");
lock.lock();
@@ -132,8 +107,5 @@ public class LowLevelFifoCachePolicy implements LowLevelCachePolicy {
} finally {
lock.unlock();
}
- if (parentDebugDump != null) {
- parentDebugDump.debugDumpShort(sb);
- }
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
index 7787cb4..82e1934 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/LowLevelLrfuCachePolicy.java
@@ -63,7 +63,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
/** Number of elements. */
private int heapSize = 0;
private EvictionListener evictionListener;
- private LlapOomDebugDump parentDebugDump;
public LowLevelLrfuCachePolicy(int minBufferSize, long maxSize, Configuration conf) {
lambda = HiveConf.getFloatVar(conf, HiveConf.ConfVars.LLAP_LRFU_LAMBDA);
@@ -174,11 +173,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
}
@Override
- public void setParentDebugDumper(LlapOomDebugDump dumper) {
- this.parentDebugDump = dumper;
- }
-
- @Override
public long purge() {
long evicted = evictSomeBlocks(Long.MAX_VALUE);
LlapIoImpl.LOG.info("PURGE: evicted {} from LRFU policy", LlapUtil.humanReadableByteCount(evicted));
@@ -487,15 +481,6 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
}
@Override
- public String debugDumpForOom() {
- String result = debugDumpHeap();
- if (parentDebugDump != null) {
- result += "\n" + parentDebugDump.debugDumpForOom();
- }
- return result;
- }
-
- @Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nLRFU eviction list: ");
LlapCacheableBuffer listHeadLocal = listHead, listTailLocal = listTail;
@@ -512,8 +497,5 @@ public class LowLevelLrfuCachePolicy implements LowLevelCachePolicy {
sb.append(c + " items");
}
sb.append("\nLRFU eviction heap: " + heapSize + " items");
- if (parentDebugDump != null) {
- parentDebugDump.debugDumpShort(sb);
- }
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
index 65b1d4f..542041d 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/MemoryManager.java
@@ -18,7 +18,7 @@
package org.apache.hadoop.hive.llap.cache;
-public interface MemoryManager extends LlapOomDebugDump {
+public interface MemoryManager {
void releaseMemory(long memUsage);
void updateMaxSize(long maxSize);
void reserveMemory(long memoryToReserve);
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
index cb89d12..a8f89ef 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SerDeLowLevelCacheImpl.java
@@ -43,7 +43,7 @@ import org.apache.orc.OrcProto.ColumnEncoding;
import com.google.common.base.Function;
-public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapOomDebugDump {
+public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapIoDebugDump {
private static final int DEFAULT_CLEANUP_INTERVAL = 600;
private final Allocator allocator;
private final AtomicInteger newEvictions = new AtomicInteger(0);
@@ -722,24 +722,6 @@ public class SerDeLowLevelCacheImpl implements BufferUsageManager, LlapOomDebugD
}
@Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder("File cache state ");
- for (Map.Entry<Object, FileCache<FileData>> e : cache.entrySet()) {
- if (!e.getValue().incRef()) continue;
- try {
- sb.append("\n file " + e.getKey());
- sb.append("\n [");
- e.getValue().getCache().toString(sb);
- sb.append("]");
- } finally {
- e.getValue().decRef();
- }
- }
- return sb.toString();
- }
-
-
- @Override
public void debugDumpShort(StringBuilder sb) {
sb.append("\nSerDe cache state ");
int allLocked = 0, allUnlocked = 0, allEvicted = 0, allMoving = 0;
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
index a1b6cae..41855e1 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cache/SimpleBufferManager.java
@@ -101,14 +101,4 @@ public class SimpleBufferManager implements BufferUsageManager, LowLevelCache {
public void notifyEvicted(MemoryBuffer buffer) {
throw new UnsupportedOperationException("Buffer manager doesn't have cache");
}
-
- @Override
- public String debugDumpForOom() {
- return "";
- }
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- // TODO Auto-generated method stub
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
index 5003d9b..e1e8a32 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java
@@ -46,7 +46,7 @@ import org.apache.hadoop.hive.llap.cache.BufferUsageManager;
import org.apache.hadoop.hive.llap.cache.CacheContentsTracker;
import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
import org.apache.hadoop.hive.llap.cache.LlapDataBuffer;
-import org.apache.hadoop.hive.llap.cache.LlapOomDebugDump;
+import org.apache.hadoop.hive.llap.cache.LlapIoDebugDump;
import org.apache.hadoop.hive.llap.cache.LowLevelCache;
import org.apache.hadoop.hive.llap.cache.LowLevelCacheImpl;
import org.apache.hadoop.hive.llap.cache.LowLevelCacheMemoryManager;
@@ -78,10 +78,13 @@ import org.apache.hive.common.util.FixedSizedObjectPool;
+
+
+
import com.google.common.primitives.Ints;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
+public class LlapIoImpl implements LlapIo<VectorizedRowBatch>, LlapIoDebugDump {
public static final Logger LOG = LoggerFactory.getLogger("LlapIoImpl");
public static final Logger ORC_LOGGER = LoggerFactory.getLogger("LlapIoOrc");
public static final Logger CACHE_LOGGER = LoggerFactory.getLogger("LlapIoCache");
@@ -95,12 +98,13 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
private final LlapDaemonIOMetrics ioMetrics;
private ObjectName buddyAllocatorMXBean;
private final Allocator allocator;
- private final LlapOomDebugDump memoryDump;
private final FileMetadataCache fileMetadataCache;
private final LowLevelCache dataCache;
private final BufferUsageManager bufferManager;
private final Configuration daemonConf;
- private LowLevelCachePolicy cachePolicy;
+ private LowLevelCachePolicy cachePolicyWrapper;
+
+ private List<LlapIoDebugDump> debugDumpComponents = new ArrayList<>();
private LlapIoImpl(Configuration conf) throws IOException {
this.daemonConf = conf;
@@ -140,51 +144,66 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
boolean useLrfu = HiveConf.getBoolVar(conf, HiveConf.ConfVars.LLAP_USE_LRFU);
long totalMemorySize = HiveConf.getSizeVar(conf, ConfVars.LLAP_IO_MEMORY_MAX_SIZE);
int minAllocSize = (int)HiveConf.getSizeVar(conf, ConfVars.LLAP_ALLOCATOR_MIN_ALLOC);
- LowLevelCachePolicy cp = useLrfu ? new LowLevelLrfuCachePolicy(
+ LowLevelCachePolicy realCachePolicy = useLrfu ? new LowLevelLrfuCachePolicy(
minAllocSize, totalMemorySize, conf) : new LowLevelFifoCachePolicy();
boolean trackUsage = HiveConf.getBoolVar(conf, HiveConf.ConfVars.LLAP_TRACK_CACHE_USAGE);
if (trackUsage) {
- this.cachePolicy = new CacheContentsTracker(cp);
+ this.cachePolicyWrapper = new CacheContentsTracker(realCachePolicy);
} else {
- this.cachePolicy = cp;
+ this.cachePolicyWrapper = realCachePolicy;
}
// Allocator uses memory manager to request memory, so create the manager next.
LowLevelCacheMemoryManager memManager = new LowLevelCacheMemoryManager(
- totalMemorySize, cachePolicy, cacheMetrics);
+ totalMemorySize, cachePolicyWrapper, cacheMetrics);
cacheMetrics.setCacheCapacityTotal(totalMemorySize);
// Cache uses allocator to allocate and deallocate, create allocator and then caches.
BuddyAllocator allocator = new BuddyAllocator(conf, memManager, cacheMetrics);
this.allocator = allocator;
- this.memoryDump = allocator;
LowLevelCacheImpl cacheImpl = new LowLevelCacheImpl(
- cacheMetrics, cachePolicy, allocator, true);
+ cacheMetrics, cachePolicyWrapper, allocator, true);
dataCache = cacheImpl;
if (isEncodeEnabled) {
SerDeLowLevelCacheImpl serdeCacheImpl = new SerDeLowLevelCacheImpl(
- cacheMetrics, cachePolicy, allocator);
+ cacheMetrics, cachePolicyWrapper, allocator);
serdeCache = serdeCacheImpl;
}
boolean useGapCache = HiveConf.getBoolVar(conf, ConfVars.LLAP_CACHE_ENABLE_ORC_GAP_CACHE);
metadataCache = new MetadataCache(
- allocator, memManager, cachePolicy, useGapCache, cacheMetrics);
+ allocator, memManager, cachePolicyWrapper, useGapCache, cacheMetrics);
fileMetadataCache = metadataCache;
// And finally cache policy uses cache to notify it of eviction. The cycle is complete!
EvictionDispatcher e = new EvictionDispatcher(
dataCache, serdeCache, metadataCache, allocator);
- cachePolicy.setEvictionListener(e);
- cachePolicy.setParentDebugDumper(e);
+ cachePolicyWrapper.setEvictionListener(e);
cacheImpl.startThreads(); // Start the cache threads.
bufferManager = bufferManagerOrc = cacheImpl; // Cache also serves as buffer manager.
bufferManagerGeneric = serdeCache;
+ if (trackUsage) {
+ debugDumpComponents.add(cachePolicyWrapper); // Cache contents tracker.
+ }
+ debugDumpComponents.add(realCachePolicy);
+ debugDumpComponents.add(cacheImpl);
+ if (serdeCache != null) {
+ debugDumpComponents.add(serdeCache);
+ }
+ if (metadataCache != null) {
+ debugDumpComponents.add(metadataCache);
+ }
+ debugDumpComponents.add(allocator);
} else {
this.allocator = new SimpleAllocator(conf);
- memoryDump = null;
fileMetadataCache = null;
SimpleBufferManager sbm = new SimpleBufferManager(allocator, cacheMetrics);
bufferManager = bufferManagerOrc = bufferManagerGeneric = sbm;
dataCache = sbm;
+ debugDumpComponents.add(new LlapIoDebugDump() {
+ @Override
+ public void debugDumpShort(StringBuilder sb) {
+ sb.append("LLAP IO allocator is not in use!");
+ }
+ });
}
// IO thread pool. Listening is used for unhandled errors for now (TODO: remove?)
int numThreads = HiveConf.getIntVar(conf, HiveConf.ConfVars.LLAP_IO_THREADPOOL_SIZE);
@@ -208,16 +227,15 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
@Override
public String getMemoryInfo() {
- if (memoryDump == null) return "\nNot using the allocator";
StringBuilder sb = new StringBuilder();
- memoryDump.debugDumpShort(sb);
+ debugDumpShort(sb);
return sb.toString();
}
@Override
public long purge() {
- if (cachePolicy != null) {
- return cachePolicy.purge();
+ if (cachePolicyWrapper != null) {
+ return cachePolicyWrapper.purge();
}
return 0;
}
@@ -309,4 +327,11 @@ public class LlapIoImpl implements LlapIo<VectorizedRowBatch> {
return new LlapDataBuffer();
}
}
+
+ @Override
+ public void debugDumpShort(StringBuilder sb) {
+ for (LlapIoDebugDump child : debugDumpComponents) {
+ child.debugDumpShort(sb);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
index 0184e30..426d599 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java
@@ -36,7 +36,7 @@ import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
import org.apache.hadoop.hive.llap.cache.EvictionAwareAllocator;
import org.apache.hadoop.hive.llap.cache.EvictionDispatcher;
import org.apache.hadoop.hive.llap.cache.LlapAllocatorBuffer;
-import org.apache.hadoop.hive.llap.cache.LlapOomDebugDump;
+import org.apache.hadoop.hive.llap.cache.LlapIoDebugDump;
import org.apache.hadoop.hive.llap.cache.LowLevelCachePolicy;
import org.apache.hadoop.hive.llap.cache.MemoryManager;
import org.apache.hadoop.hive.llap.cache.LowLevelCache.Priority;
@@ -44,7 +44,7 @@ import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl;
import org.apache.hadoop.hive.llap.metrics.LlapDaemonCacheMetrics;
import org.apache.hadoop.hive.ql.io.orc.encoded.OrcBatchKey;
-public class MetadataCache implements LlapOomDebugDump, FileMetadataCache {
+public class MetadataCache implements LlapIoDebugDump, FileMetadataCache {
private final ConcurrentHashMap<Object, LlapBufferOrBuffers> metadata =
new ConcurrentHashMap<>();
@@ -121,12 +121,6 @@ public class MetadataCache implements LlapOomDebugDump, FileMetadataCache {
estimateErrors.remove(buffer.getFileKey());
}
- @Override
- public String debugDumpForOom() {
- StringBuilder sb = new StringBuilder();
- debugDumpShort(sb);
- return sb.toString();
- }
@Override
public void debugDumpShort(StringBuilder sb) {
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestBuddyAllocator.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestBuddyAllocator.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestBuddyAllocator.java
index 714efef..1e6f3ac 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestBuddyAllocator.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestBuddyAllocator.java
@@ -66,17 +66,8 @@ public class TestBuddyAllocator {
}
@Override
- public String debugDumpForOom() {
- return "";
- }
-
- @Override
public void updateMaxSize(long maxSize) {
}
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- }
}
@Test
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
index b19cdcf..1c2eef2 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestLowLevelCacheImpl.java
@@ -112,11 +112,7 @@ public class TestLowLevelCacheImpl {
public void setEvictionListener(EvictionListener listener) {
}
- public String debugDumpForOom() {
- return "";
- }
-
- public void setParentDebugDumper(LlapOomDebugDump dumper) {
+ public void setParentDebugDumper(LlapIoDebugDump dumper) {
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/43e331e3/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
index 58c918c..df20f20 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestOrcMetadataCache.java
@@ -56,11 +56,7 @@ public class TestOrcMetadataCache {
public void setEvictionListener(EvictionListener listener) {
}
- public String debugDumpForOom() {
- return "";
- }
-
- public void setParentDebugDumper(LlapOomDebugDump dumper) {
+ public void setParentDebugDumper(LlapIoDebugDump dumper) {
}
@Override
@@ -88,17 +84,8 @@ public class TestOrcMetadataCache {
}
@Override
- public String debugDumpForOom() {
- return "";
- }
-
- @Override
public void updateMaxSize(long maxSize) {
}
-
- @Override
- public void debugDumpShort(StringBuilder sb) {
- }
}
@Test
[2/2] hive git commit: HIVE-19597 : TestWorkloadManager sometimes
hangs (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Posted by se...@apache.org.
HIVE-19597 : TestWorkloadManager sometimes hangs (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/283e51d2
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/283e51d2
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/283e51d2
Branch: refs/heads/master
Commit: 283e51d25706bb8f7dfae051d346c16369465329
Parents: 2d3faea
Author: sergey <se...@apache.org>
Authored: Mon Jun 4 11:27:22 2018 -0700
Committer: sergey <se...@apache.org>
Committed: Mon Jun 4 13:31:22 2018 -0700
----------------------------------------------------------------------
.../hive/ql/exec/tez/WorkloadManager.java | 23 ++++++----
.../hive/ql/exec/tez/TestWorkloadManager.java | 45 +++++++++++---------
2 files changed, 40 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/283e51d2/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
index 97ba036..7137a17 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
@@ -339,7 +339,8 @@ public class WorkloadManager extends TezSessionPoolSession.AbstractTriggerValida
private WMFullResourcePlan resourcePlanToApply = null;
private boolean doClearResourcePlan = false;
private boolean hasClusterStateChanged = false;
- private SettableFuture<Boolean> testEvent, applyRpFuture;
+ private List<SettableFuture<Boolean>> testEvents = new LinkedList<>();
+ private SettableFuture<Boolean> applyRpFuture;
private SettableFuture<List<String>> dumpStateFuture;
private final List<MoveSession> moveSessions = new LinkedList<>();
}
@@ -401,10 +402,11 @@ public class WorkloadManager extends TezSessionPoolSession.AbstractTriggerValida
return;
} catch (Exception | AssertionError ex) {
LOG.error("WM thread encountered an error but will attempt to continue", ex);
- if (currentEvents.testEvent != null) {
- currentEvents.testEvent.setException(ex);
- currentEvents.testEvent = null;
+ for (SettableFuture<Boolean> testEvent : currentEvents.testEvents) {
+ LOG.info("Failing test event " + System.identityHashCode(testEvent));
+ testEvent.setException(ex);
}
+ currentEvents.testEvents.clear();
if (currentEvents.applyRpFuture != null) {
currentEvents.applyRpFuture.setException(ex);
currentEvents.applyRpFuture = null;
@@ -721,12 +723,14 @@ public class WorkloadManager extends TezSessionPoolSession.AbstractTriggerValida
e.dumpStateFuture.set(result);
e.dumpStateFuture = null;
}
-
+
// 15. Notify tests and global async ops.
- if (e.testEvent != null) {
- e.testEvent.set(true);
- e.testEvent = null;
+ for (SettableFuture<Boolean> testEvent : e.testEvents) {
+ LOG.info("Triggering test event " + System.identityHashCode(testEvent));
+ testEvent.set(null);
}
+ e.testEvents.clear();
+
if (e.applyRpFuture != null) {
e.applyRpFuture.set(true);
e.applyRpFuture = null;
@@ -1552,7 +1556,8 @@ public class WorkloadManager extends TezSessionPoolSession.AbstractTriggerValida
SettableFuture<Boolean> testEvent = SettableFuture.create();
currentLock.lock();
try {
- current.testEvent = testEvent;
+ LOG.info("Adding test event " + System.identityHashCode(testEvent));
+ current.testEvents.add(testEvent);
notifyWmThreadUnderLock();
} finally {
currentLock.unlock();
http://git-wip-us.apache.org/repos/asf/hive/blob/283e51d2/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestWorkloadManager.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestWorkloadManager.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestWorkloadManager.java
index 6e15b2c..30ad212 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestWorkloadManager.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestWorkloadManager.java
@@ -60,7 +60,6 @@ import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@RunWith(RetryTestRunner.class)
public class TestWorkloadManager {
@SuppressWarnings("unused")
private static final Logger LOG = LoggerFactory.getLogger(TestWorkloadManager.class);
@@ -90,8 +89,10 @@ public class TestWorkloadManager {
if (cdl != null) {
cdl.countDown();
}
+ LOG.info("About to call get with " + old);
try {
session.set((WmTezSession) wm.getSession(old, mappingInput(userName), conf));
+ LOG.info("Received " + session.get());
} catch (Throwable e) {
error.compareAndSet(null, e);
}
@@ -114,7 +115,7 @@ public class TestWorkloadManager {
isCalled = true;
return 0;
}
-
+
@Override
public void updateSessionAsync(WmTezSession session) {
}
@@ -206,7 +207,7 @@ public class TestWorkloadManager {
}
private static WMFullResourcePlan createDummyPlan(int numSessions) {
- WMFullResourcePlan plan = new WMFullResourcePlan(new WMResourcePlan("rp"),
+ WMFullResourcePlan plan = new WMFullResourcePlan(new WMResourcePlan("rp"),
Lists.newArrayList(pool("llap", numSessions, 1.0f)));
plan.getPlan().setDefaultPoolPath("llap");
return plan;
@@ -441,7 +442,7 @@ public class TestWorkloadManager {
session.returnToSessionManager();
}
-
+
@Test(timeout=10000)
@@ -477,12 +478,12 @@ public class TestWorkloadManager {
checkError(error);
// Now release a single session from A.
sessionA1.returnToSessionManager();
- t1.join();
+ joinThread(t1);
checkError(error);
assertNotNull(sessionA3.get());
assertNull(sessionA4.get());
sessionA3.get().returnToSessionManager();
- t2.join();
+ joinThread(t2);
checkError(error);
assertNotNull(sessionA4.get());
sessionA4.get().returnToSessionManager();
@@ -553,14 +554,14 @@ public class TestWorkloadManager {
assertNull(session4.get());
// We have released the session by trying to reuse it and going back into queue, s3 can start.
- t1.join();
+ joinThread(t1);
checkError(error);
assertNotNull(session3.get());
assertEquals(0.5, session3.get().getClusterFraction(), EPSILON);
// Now release another session; the thread that gave up on reuse can proceed.
session1.returnToSessionManager();
- t2.join();
+ joinThread(t2);
checkError(error);
assertNotNull(session4.get());
assertNotSame(session2, session4.get());
@@ -569,6 +570,12 @@ public class TestWorkloadManager {
session4.get().returnToSessionManager();
}
+ private static void joinThread(Thread t) throws InterruptedException {
+ LOG.debug("Joining " + t.getName());
+ t.join();
+ LOG.debug("Joined " + t.getName());
+ }
+
private void waitForThreadToBlock(CountDownLatch cdl, Thread t1) throws InterruptedException {
t1.start();
cdl.await();
@@ -619,7 +626,7 @@ public class TestWorkloadManager {
plan.setMappings(Lists.newArrayList(mapping("U", "A")));
final WorkloadManager wm = new WorkloadManagerForTest("test", conf, qam, plan);
wm.start();
-
+
// One session will be running, the other will be queued in "A"
WmTezSession sessionA1 = (WmTezSession) wm.getSession(null, mappingInput("U"), conf);
assertEquals("A", sessionA1.getPoolName());
@@ -639,7 +646,7 @@ public class TestWorkloadManager {
wm.updateResourcePlanAsync(plan);
// The session will go to B with the new mapping; check it.
- t1.join();
+ joinThread(t1);
checkError(error);
assertNotNull(sessionA2.get());
assertEquals("B", sessionA2.get().getPoolName());
@@ -667,7 +674,7 @@ public class TestWorkloadManager {
wm.start();
TezSessionPool<WmTezSession> tezAmPool = wm.getTezAmPool();
assertEquals(6, tezAmPool.getCurrentSize());
-
+
// A: 1/1 running, 1 queued; B: 2/2 running, C: 1/2 running, D: 1/1 running, 1 queued.
// Total: 5/6 running.
WmTezSession sessionA1 = (WmTezSession) wm.getSession(null, mappingInput("A"), conf),
@@ -700,8 +707,8 @@ public class TestWorkloadManager {
wm.updateResourcePlanAsync(plan);
wm.addTestEvent().get();
- t1.join();
- t2.join();
+ joinThread(t1);
+ joinThread(t2);
checkError(error);
assertNotNull(sessionA2.get());
assertNotNull(sessionD2.get());
@@ -743,7 +750,7 @@ public class TestWorkloadManager {
plan.getPlan().setDefaultPoolPath("A");
final WorkloadManager wm = new WorkloadManagerForTest("test", conf, qam, plan);
wm.start();
-
+
// 2 running.
WmTezSession sessionA1 = (WmTezSession) wm.getSession(
null, mappingInput("A", null), conf, null),
@@ -807,7 +814,7 @@ public class TestWorkloadManager {
// Remove the resource plan - disable WM. All the queries die.
wm.updateResourcePlanAsync(null).get();
- t1.join();
+ joinThread(t1);
assertNotNull(error.get());
assertNull(sessionA2.get());
assertKilledByWm(sessionA1);
@@ -843,7 +850,7 @@ public class TestWorkloadManager {
// Take away the only session, as if it was expiring.
TezSessionPool<WmTezSession> pool = wm.getTezAmPool();
WmTezSession oob = pool.getSession();
-
+
final AtomicReference<WmTezSession> sessionA1 = new AtomicReference<>();
final AtomicReference<Throwable> error = new AtomicReference<>();
final CountDownLatch cdl1 = new CountDownLatch(1);
@@ -852,7 +859,7 @@ public class TestWorkloadManager {
checkError(error);
// Replacing it directly in the pool should unblock get.
pool.replaceSession(oob);
- t1.join();
+ joinThread(t1);
assertNotNull(sessionA1.get());
assertEquals("A", sessionA1.get().getPoolName());
@@ -1136,7 +1143,7 @@ public class TestWorkloadManager {
wm.updateResourcePlanAsync(plan);
wm.addTestEvent().get();
blockedWait.set(true); // Meanwhile, the init succeeds!
- t1.join();
+ joinThread(t1);
try {
sessionA.get();
fail("Expected an error but got " + sessionA.get());
@@ -1191,7 +1198,7 @@ public class TestWorkloadManager {
wm.updateResourcePlanAsync(plan);
wm.addTestEvent().get();
failedWait.setException(new Exception("moo")); // Meanwhile, the init fails.
- t1.join();
+ joinThread(t1);
try {
sessionA.get();
fail("Expected an error but got " + sessionA.get());