1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.monitoring;
20
21 import static org.junit.Assert.*;
22
23 import java.util.concurrent.atomic.AtomicBoolean;
24
25 import org.apache.hadoop.conf.Configuration;
26 import org.apache.hadoop.hbase.testclassification.SmallTests;
27 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
28 import org.junit.Test;
29 import org.junit.experimental.categories.Category;
30
31 @Category(SmallTests.class)
32 public class TestTaskMonitor {
33
34 @Test
35 public void testTaskMonitorBasics() {
36 TaskMonitor tm = new TaskMonitor(new Configuration());
37 assertTrue("Task monitor should start empty",
38 tm.getTasks().isEmpty());
39
40
41 MonitoredTask task = tm.createStatus("Test task");
42 MonitoredTask taskFromTm = tm.getTasks().get(0);
43
44
45 assertEquals(task.getDescription(), taskFromTm.getDescription());
46 assertEquals(-1, taskFromTm.getCompletionTimestamp());
47 assertEquals(MonitoredTask.State.RUNNING, taskFromTm.getState());
48
49
50 task.markComplete("Finished!");
51 assertEquals(MonitoredTask.State.COMPLETE, task.getState());
52
53
54 assertEquals(1, tm.getTasks().size());
55
56
57 task.expireNow();
58 assertEquals(0, tm.getTasks().size());
59
60 tm.shutdown();
61 }
62
63 @Test
64 public void testTasksGetAbortedOnLeak() throws InterruptedException {
65 final TaskMonitor tm = new TaskMonitor(new Configuration());
66 assertTrue("Task monitor should start empty",
67 tm.getTasks().isEmpty());
68
69 final AtomicBoolean threadSuccess = new AtomicBoolean(false);
70
71 Thread t = new Thread() {
72 @Override
73 public void run() {
74 MonitoredTask task = tm.createStatus("Test task");
75 assertEquals(MonitoredTask.State.RUNNING, task.getState());
76 threadSuccess.set(true);
77 }
78 };
79 t.start();
80 t.join();
81
82 assertTrue(threadSuccess.get());
83
84
85 System.gc();
86 System.gc();
87 System.gc();
88
89
90 MonitoredTask taskFromTm = tm.getTasks().get(0);
91 assertEquals(MonitoredTask.State.ABORTED, taskFromTm.getState());
92
93 tm.shutdown();
94 }
95
96 @Test
97 public void testTaskLimit() throws Exception {
98 TaskMonitor tm = new TaskMonitor(new Configuration());
99 for (int i = 0; i < TaskMonitor.DEFAULT_MAX_TASKS + 10; i++) {
100 tm.createStatus("task " + i);
101 }
102
103 assertEquals(TaskMonitor.DEFAULT_MAX_TASKS, tm.getTasks().size());
104
105
106 assertEquals("task 10", tm.getTasks().get(0).getDescription());
107 tm.shutdown();
108 }
109
110 @Test
111 public void testDoNotPurgeRPCTask() throws Exception {
112 int RPCTaskNums = 10;
113 TaskMonitor tm = TaskMonitor.get();
114 for(int i = 0; i < RPCTaskNums; i++) {
115 tm.createRPCStatus("PRCTask" + i);
116 }
117 for(int i = 0; i < TaskMonitor.DEFAULT_MAX_TASKS; i++) {
118 tm.createStatus("otherTask" + i);
119 }
120 int remainRPCTask = 0;
121 for(MonitoredTask task: tm.getTasks()) {
122 if(task instanceof MonitoredRPCHandler) {
123 remainRPCTask++;
124 }
125 }
126 assertEquals("RPC Tasks have been purged!", RPCTaskNums, remainRPCTask);
127 tm.shutdown();
128 }
129
130 @Test
131 public void testWarnStuckTasks() throws Exception {
132 final int INTERVAL = 1000;
133 Configuration conf = new Configuration();
134 conf.setLong(TaskMonitor.RPC_WARN_TIME_KEY, INTERVAL);
135 conf.setLong(TaskMonitor.MONITOR_INTERVAL_KEY, INTERVAL);
136 final TaskMonitor tm = new TaskMonitor(conf);
137 MonitoredRPCHandler t = tm.createRPCStatus("test task");
138 long then = EnvironmentEdgeManager.currentTime();
139 t.setRPC("testMethod", new Object[0], then);
140 Thread.sleep(INTERVAL * 2);
141 assertTrue("We did not warn", t.getWarnTime() > then);
142 tm.shutdown();
143 }
144
145 @Test
146 public void testStatusJournal() {
147 TaskMonitor tm = new TaskMonitor(new Configuration());
148 MonitoredTask task = tm.createStatus("Test task");
149 assertTrue(task.getStatusJournal().isEmpty());
150 task.disableStatusJournal();
151 task.setStatus("status1");
152
153 assertTrue(task.getStatusJournal().isEmpty());
154 task.enableStatusJournal(true);
155
156 assertEquals("status1", task.getStatusJournal().get(0).getStatus());
157 assertTrue(task.getStatusJournal().get(0).getTimeStamp() > 0);
158 task.disableStatusJournal();
159 task.setStatus("status2");
160
161 assertEquals(1, task.getStatusJournal().size());
162 task.enableStatusJournal(false);
163
164 assertEquals(1, task.getStatusJournal().size());
165 task.setStatus("status3");
166 assertEquals("status3", task.getStatusJournal().get(1).getStatus());
167 tm.shutdown();
168 }
169
170 }
171