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