1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 package org.apache.hadoop.hbase.regionserver.wal;
22
23 import java.io.IOException;
24 import java.util.List;
25
26 import org.apache.hadoop.fs.Path;
27 import org.apache.hadoop.hbase.Coprocessor;
28 import org.apache.hadoop.hbase.HRegionInfo;
29 import org.apache.hadoop.hbase.coprocessor.*;
30 import org.apache.hadoop.hbase.classification.InterfaceAudience;
31 import org.apache.hadoop.conf.Configuration;
32 import org.apache.hadoop.hbase.wal.WAL;
33 import org.apache.hadoop.hbase.wal.WALKey;
34
35
36
37
38
39 @InterfaceAudience.Private
40 public class WALCoprocessorHost
41 extends CoprocessorHost<WALCoprocessorHost.WALEnvironment> {
42
43
44
45
46 static class WALEnvironment extends CoprocessorHost.Environment
47 implements WALCoprocessorEnvironment {
48
49 private final WAL wal;
50
51 final boolean useLegacyPre;
52 final boolean useLegacyPost;
53
54 @Override
55 public WAL getWAL() {
56 return wal;
57 }
58
59
60
61
62
63
64
65
66
67
68 public WALEnvironment(Class<?> implClass, final Coprocessor impl,
69 final int priority, final int seq, final Configuration conf,
70 final WAL wal) {
71 super(impl, priority, seq, conf);
72 this.wal = wal;
73
74
75
76
77 useLegacyPre = useLegacyMethod(impl.getClass(), "preWALWrite", ObserverContext.class,
78 HRegionInfo.class, WALKey.class, WALEdit.class);
79 useLegacyPost = useLegacyMethod(impl.getClass(), "postWALWrite", ObserverContext.class,
80 HRegionInfo.class, WALKey.class, WALEdit.class);
81 }
82 }
83
84 private final WAL wal;
85
86
87
88
89
90
91 public WALCoprocessorHost(final WAL log, final Configuration conf) {
92
93
94
95
96
97
98 super(null);
99 this.wal = log;
100
101 loadSystemCoprocessors(conf, WAL_COPROCESSOR_CONF_KEY);
102 }
103
104 @Override
105 public WALEnvironment createEnvironment(final Class<?> implClass,
106 final Coprocessor instance, final int priority, final int seq,
107 final Configuration conf) {
108 return new WALEnvironment(implClass, instance, priority, seq, conf,
109 this.wal);
110 }
111
112
113
114
115
116
117
118
119 public boolean preWALWrite(final HRegionInfo info, final WALKey logKey, final WALEdit logEdit)
120 throws IOException {
121 boolean bypass = false;
122 if (this.coprocessors == null || this.coprocessors.isEmpty()) return bypass;
123 ObserverContext<WALCoprocessorEnvironment> ctx = null;
124 List<WALEnvironment> envs = coprocessors.get();
125 for (int i = 0; i < envs.size(); i++) {
126 WALEnvironment env = envs.get(i);
127 if (env.getInstance() instanceof WALObserver) {
128 final WALObserver observer = (WALObserver)env.getInstance();
129 ctx = ObserverContext.createAndPrepare(env, ctx);
130 Thread currentThread = Thread.currentThread();
131 ClassLoader cl = currentThread.getContextClassLoader();
132 try {
133 currentThread.setContextClassLoader(env.getClassLoader());
134 if (env.useLegacyPre) {
135 if (logKey instanceof HLogKey) {
136 observer.preWALWrite(ctx, info, (HLogKey)logKey, logEdit);
137 } else {
138 legacyWarning(observer.getClass(),
139 "There are wal keys present that are not HLogKey.");
140 }
141 } else {
142 observer.preWALWrite(ctx, info, logKey, logEdit);
143 }
144 } catch (Throwable e) {
145 handleCoprocessorThrowable(env, e);
146 } finally {
147 currentThread.setContextClassLoader(cl);
148 }
149 bypass |= ctx.shouldBypass();
150 if (ctx.shouldComplete()) {
151 break;
152 }
153 }
154 }
155 return bypass;
156 }
157
158
159
160
161
162
163
164 public void postWALWrite(final HRegionInfo info, final WALKey logKey, final WALEdit logEdit)
165 throws IOException {
166 if (this.coprocessors == null || this.coprocessors.isEmpty()) return;
167 ObserverContext<WALCoprocessorEnvironment> ctx = null;
168 List<WALEnvironment> envs = coprocessors.get();
169 for (int i = 0; i < envs.size(); i++) {
170 WALEnvironment env = envs.get(i);
171 if (env.getInstance() instanceof WALObserver) {
172 final WALObserver observer = (WALObserver)env.getInstance();
173 ctx = ObserverContext.createAndPrepare(env, ctx);
174 Thread currentThread = Thread.currentThread();
175 ClassLoader cl = currentThread.getContextClassLoader();
176 try {
177 currentThread.setContextClassLoader(env.getClassLoader());
178 if (env.useLegacyPost) {
179 if (logKey instanceof HLogKey) {
180 observer.postWALWrite(ctx, info, (HLogKey)logKey, logEdit);
181 } else {
182 legacyWarning(observer.getClass(),
183 "There are wal keys present that are not HLogKey.");
184 }
185 } else {
186 observer.postWALWrite(ctx, info, logKey, logEdit);
187 }
188 } catch (Throwable e) {
189 handleCoprocessorThrowable(env, e);
190 } finally {
191 currentThread.setContextClassLoader(cl);
192 }
193 if (ctx.shouldComplete()) {
194 break;
195 }
196 }
197 }
198 }
199
200
201
202
203
204
205 public void preWALRoll(Path oldPath, Path newPath) throws IOException {
206 if (this.coprocessors == null || this.coprocessors.isEmpty()) return;
207 ObserverContext<WALCoprocessorEnvironment> ctx = null;
208 List<WALEnvironment> envs = coprocessors.get();
209 for (int i = 0; i < envs.size(); i++) {
210 WALEnvironment env = envs.get(i);
211 if (env.getInstance() instanceof WALObserver) {
212 final WALObserver observer = (WALObserver)env.getInstance();
213 ctx = ObserverContext.createAndPrepare(env, ctx);
214 Thread currentThread = Thread.currentThread();
215 ClassLoader cl = currentThread.getContextClassLoader();
216 try {
217 currentThread.setContextClassLoader(env.getClassLoader());
218 observer.preWALRoll(ctx, oldPath, newPath);
219 } catch (Throwable e) {
220 handleCoprocessorThrowable(env, e);
221 } finally {
222 currentThread.setContextClassLoader(cl);
223 }
224 if (ctx.shouldComplete()) {
225 break;
226 }
227 }
228 }
229 }
230
231
232
233
234
235
236 public void postWALRoll(Path oldPath, Path newPath) throws IOException {
237 if (this.coprocessors == null || this.coprocessors.isEmpty()) return;
238 ObserverContext<WALCoprocessorEnvironment> ctx = null;
239 List<WALEnvironment> envs = coprocessors.get();
240 for (int i = 0; i < envs.size(); i++) {
241 WALEnvironment env = envs.get(i);
242 if (env.getInstance() instanceof WALObserver) {
243 final WALObserver observer = (WALObserver)env.getInstance();
244 ctx = ObserverContext.createAndPrepare(env, ctx);
245 Thread currentThread = Thread.currentThread();
246 ClassLoader cl = currentThread.getContextClassLoader();
247 try {
248 currentThread.setContextClassLoader(env.getClassLoader());
249 observer.postWALRoll(ctx, oldPath, newPath);
250 } catch (Throwable e) {
251 handleCoprocessorThrowable(env, e);
252 } finally {
253 currentThread.setContextClassLoader(cl);
254 }
255 if (ctx.shouldComplete()) {
256 break;
257 }
258 }
259 }
260 }
261 }