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.wal;
22
23 import java.io.IOException;
24 import java.util.Arrays;
25 import java.io.InterruptedIOException;
26 import java.util.Collections;
27 import java.util.List;
28 import java.util.concurrent.atomic.AtomicReference;
29
30 import com.google.common.annotations.VisibleForTesting;
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.hadoop.hbase.classification.InterfaceAudience;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.fs.FSDataInputStream;
36 import org.apache.hadoop.fs.FileSystem;
37 import org.apache.hadoop.fs.Path;
38 import org.apache.hadoop.hbase.HConstants;
39 import org.apache.hadoop.hbase.wal.WAL.Reader;
40 import org.apache.hadoop.hbase.wal.WALProvider.Writer;
41 import org.apache.hadoop.hbase.util.CancelableProgressable;
42 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
43 import org.apache.hadoop.hbase.util.LeaseNotRecoveredException;
44
45
46 import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
47 import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogReader;
48 import org.apache.hadoop.hbase.regionserver.wal.SequenceFileLogReader;
49 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69 @InterfaceAudience.Private
70 public class WALFactory {
71
72 private static final Log LOG = LogFactory.getLog(WALFactory.class);
73
74
75
76
77 static enum Providers {
78 defaultProvider(DefaultWALProvider.class),
79 filesystem(DefaultWALProvider.class),
80 multiwal(RegionGroupingProvider.class);
81
82 Class<? extends WALProvider> clazz;
83 Providers(Class<? extends WALProvider> clazz) {
84 this.clazz = clazz;
85 }
86 }
87
88 public static final String WAL_PROVIDER = "hbase.wal.provider";
89 static final String DEFAULT_WAL_PROVIDER = Providers.defaultProvider.name();
90
91 static final String META_WAL_PROVIDER = "hbase.wal.meta_provider";
92 static final String DEFAULT_META_WAL_PROVIDER = Providers.defaultProvider.name();
93
94 final String factoryId;
95 final WALProvider provider;
96
97
98
99 final AtomicReference<WALProvider> metaProvider = new AtomicReference<WALProvider>();
100
101
102
103
104 private final Class<? extends DefaultWALProvider.Reader> logReaderClass;
105
106
107
108
109 private final int timeoutMillis;
110
111 private final Configuration conf;
112
113
114 private WALFactory(Configuration conf) {
115
116
117
118 timeoutMillis = conf.getInt("hbase.hlog.open.timeout", 300000);
119
120 logReaderClass = conf.getClass("hbase.regionserver.hlog.reader.impl", ProtobufLogReader.class,
121 DefaultWALProvider.Reader.class);
122 this.conf = conf;
123
124
125
126 provider = null;
127 factoryId = SINGLETON_ID;
128 }
129
130 Class<? extends WALProvider> getProviderClass(String key, String defaultValue) {
131 try {
132 return Providers.valueOf(conf.get(key, defaultValue)).clazz;
133 } catch (IllegalArgumentException exception) {
134
135
136
137 return conf.getClass(key, DefaultWALProvider.class, WALProvider.class);
138 }
139 }
140
141 WALProvider createProvider(Class<? extends WALProvider> clazz,
142 List<WALActionsListener> listeners, String providerId) throws IOException {
143 LOG.info("Instantiating WALProvider of type " + clazz);
144 try {
145 final WALProvider result = clazz.newInstance();
146 result.init(this, conf, listeners, providerId);
147 return result;
148 } catch (InstantiationException exception) {
149 LOG.error("couldn't set up WALProvider, the configured class is " + clazz);
150 LOG.debug("Exception details for failure to load WALProvider.", exception);
151 throw new IOException("couldn't set up WALProvider", exception);
152 } catch (IllegalAccessException exception) {
153 LOG.error("couldn't set up WALProvider, the configured class is " + clazz);
154 LOG.debug("Exception details for failure to load WALProvider.", exception);
155 throw new IOException("couldn't set up WALProvider", exception);
156 }
157 }
158
159
160
161
162
163 WALProvider getProvider(final String key, final String defaultValue,
164 final List<WALActionsListener> listeners, final String providerId) throws IOException {
165 Class<? extends WALProvider> clazz = getProviderClass(key, defaultValue);
166 return createProvider(clazz, listeners, providerId);
167 }
168
169
170
171
172
173
174
175
176 public WALFactory(final Configuration conf, final List<WALActionsListener> listeners,
177 final String factoryId) throws IOException {
178
179
180 timeoutMillis = conf.getInt("hbase.hlog.open.timeout", 300000);
181
182 logReaderClass = conf.getClass("hbase.regionserver.hlog.reader.impl", ProtobufLogReader.class,
183 DefaultWALProvider.Reader.class);
184 this.conf = conf;
185 this.factoryId = factoryId;
186
187 if (conf.getBoolean("hbase.regionserver.hlog.enabled", true)) {
188 provider = getProvider(WAL_PROVIDER, DEFAULT_WAL_PROVIDER, listeners, null);
189 } else {
190
191 LOG.warn("Running with WAL disabled.");
192 provider = new DisabledWALProvider();
193 provider.init(this, conf, null, factoryId);
194 }
195 }
196
197
198
199
200
201
202 public void close() throws IOException {
203 final WALProvider metaProvider = this.metaProvider.get();
204 if (null != metaProvider) {
205 metaProvider.close();
206 }
207
208
209 if (null != provider) {
210 provider.close();
211 }
212 }
213
214
215
216
217
218
219 public void shutdown() throws IOException {
220 IOException exception = null;
221 final WALProvider metaProvider = this.metaProvider.get();
222 if (null != metaProvider) {
223 try {
224 metaProvider.shutdown();
225 } catch(IOException ioe) {
226 exception = ioe;
227 }
228 }
229 provider.shutdown();
230 if (null != exception) {
231 throw exception;
232 }
233 }
234
235
236
237
238
239 public WAL getWAL(final byte[] identifier, final byte[] namespace) throws IOException {
240 return provider.getWAL(identifier, namespace);
241 }
242
243
244
245
246 public WAL getMetaWAL(final byte[] identifier) throws IOException {
247 WALProvider metaProvider = this.metaProvider.get();
248 if (null == metaProvider) {
249 final WALProvider temp = getProvider(META_WAL_PROVIDER, DEFAULT_META_WAL_PROVIDER,
250 Collections.<WALActionsListener>singletonList(new MetricsWAL()),
251 DefaultWALProvider.META_WAL_PROVIDER_ID);
252 if (this.metaProvider.compareAndSet(null, temp)) {
253 metaProvider = temp;
254 } else {
255
256 temp.close();
257 metaProvider = this.metaProvider.get();
258 }
259 }
260 return metaProvider.getWAL(identifier, null);
261 }
262
263 public Reader createReader(final FileSystem fs, final Path path) throws IOException {
264 return createReader(fs, path, (CancelableProgressable)null);
265 }
266
267
268
269
270
271
272
273
274 public Reader createReader(final FileSystem fs, final Path path,
275 CancelableProgressable reporter) throws IOException {
276 return createReader(fs, path, reporter, true);
277 }
278
279 public Reader createReader(final FileSystem fs, final Path path,
280 CancelableProgressable reporter, boolean allowCustom)
281 throws IOException {
282 Class<? extends DefaultWALProvider.Reader> lrClass =
283 allowCustom ? logReaderClass : ProtobufLogReader.class;
284
285 try {
286
287
288
289 long startWaiting = EnvironmentEdgeManager.currentTime();
290 long openTimeout = timeoutMillis + startWaiting;
291 int nbAttempt = 0;
292 FSDataInputStream stream = null;
293 DefaultWALProvider.Reader reader = null;
294 while (true) {
295 try {
296 if (lrClass != ProtobufLogReader.class) {
297
298 reader = lrClass.newInstance();
299 reader.init(fs, path, conf, null);
300 return reader;
301 } else {
302 stream = fs.open(path);
303
304
305
306
307 byte[] magic = new byte[ProtobufLogReader.PB_WAL_MAGIC.length];
308 boolean isPbWal = (stream.read(magic) == magic.length)
309 && Arrays.equals(magic, ProtobufLogReader.PB_WAL_MAGIC);
310 reader =
311 isPbWal ? new ProtobufLogReader() : new SequenceFileLogReader();
312 reader.init(fs, path, conf, stream);
313 return reader;
314 }
315 } catch (IOException e) {
316 if (stream != null) {
317 try {
318 stream.close();
319 } catch (IOException exception) {
320 LOG.warn("Could not close DefaultWALProvider.Reader" + exception.getMessage());
321 LOG.debug("exception details", exception);
322 }
323 }
324 if (reader != null) {
325 try {
326 reader.close();
327 } catch (IOException exception) {
328 LOG.warn("Could not close FSDataInputStream" + exception.getMessage());
329 LOG.debug("exception details", exception);
330 }
331 }
332 String msg = e.getMessage();
333 if (msg != null && (msg.contains("Cannot obtain block length")
334 || msg.contains("Could not obtain the last block")
335 || msg.matches("Blocklist for [^ ]* has changed.*"))) {
336 if (++nbAttempt == 1) {
337 LOG.warn("Lease should have recovered. This is not expected. Will retry", e);
338 }
339 if (reporter != null && !reporter.progress()) {
340 throw new InterruptedIOException("Operation is cancelled");
341 }
342 if (nbAttempt > 2 && openTimeout < EnvironmentEdgeManager.currentTime()) {
343 LOG.error("Can't open after " + nbAttempt + " attempts and "
344 + (EnvironmentEdgeManager.currentTime() - startWaiting) + "ms " + " for " + path);
345 } else {
346 try {
347 Thread.sleep(nbAttempt < 3 ? 500 : 1000);
348 continue;
349 } catch (InterruptedException ie) {
350 InterruptedIOException iioe = new InterruptedIOException();
351 iioe.initCause(ie);
352 throw iioe;
353 }
354 }
355 throw new LeaseNotRecoveredException(e);
356 } else {
357 throw e;
358 }
359 }
360 }
361 } catch (IOException ie) {
362 throw ie;
363 } catch (Exception e) {
364 throw new IOException("Cannot get log reader", e);
365 }
366 }
367
368
369
370
371
372
373
374
375 public Writer createWALWriter(final FileSystem fs, final Path path) throws IOException {
376 return DefaultWALProvider.createWriter(conf, fs, path, false);
377 }
378
379
380
381
382
383 @VisibleForTesting
384 public Writer createRecoveredEditsWriter(final FileSystem fs, final Path path)
385 throws IOException {
386 return DefaultWALProvider.createWriter(conf, fs, path, true);
387 }
388
389
390
391
392
393 private static final AtomicReference<WALFactory> singleton = new AtomicReference<WALFactory>();
394 private static final String SINGLETON_ID = WALFactory.class.getName();
395
396
397 public static WALFactory getInstance(Configuration configuration) {
398 WALFactory factory = singleton.get();
399 if (null == factory) {
400 WALFactory temp = new WALFactory(configuration);
401 if (singleton.compareAndSet(null, temp)) {
402 factory = temp;
403 } else {
404
405 try {
406 temp.close();
407 } catch (IOException exception) {
408 LOG.debug("failed to close temporary singleton. ignoring.", exception);
409 }
410 factory = singleton.get();
411 }
412 }
413 return factory;
414 }
415
416
417
418
419
420
421 public static Reader createReader(final FileSystem fs, final Path path,
422 final Configuration configuration) throws IOException {
423 return getInstance(configuration).createReader(fs, path);
424 }
425
426
427
428
429
430
431 static Reader createReader(final FileSystem fs, final Path path,
432 final Configuration configuration, final CancelableProgressable reporter) throws IOException {
433 return getInstance(configuration).createReader(fs, path, reporter);
434 }
435
436
437
438
439
440
441
442 public static Reader createReaderIgnoreCustomClass(final FileSystem fs, final Path path,
443 final Configuration configuration) throws IOException {
444 return getInstance(configuration).createReader(fs, path, null, false);
445 }
446
447
448
449
450
451 static Writer createRecoveredEditsWriter(final FileSystem fs, final Path path,
452 final Configuration configuration)
453 throws IOException {
454 return DefaultWALProvider.createWriter(configuration, fs, path, true);
455 }
456
457
458
459
460
461 @VisibleForTesting
462 public static Writer createWALWriter(final FileSystem fs, final Path path,
463 final Configuration configuration)
464 throws IOException {
465 return DefaultWALProvider.createWriter(configuration, fs, path, false);
466 }
467
468 public final WALProvider getWALProvider() {
469 return this.provider;
470 }
471
472 public final WALProvider getMetaWALProvider() {
473 return this.metaProvider.get();
474 }
475 }