1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.wal;
20
21 import java.io.EOFException;
22 import java.io.FileNotFoundException;
23 import java.io.IOException;
24 import java.io.InterruptedIOException;
25 import java.text.ParseException;
26 import java.util.ArrayList;
27 import java.util.Collections;
28 import java.util.HashSet;
29 import java.util.LinkedList;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.NavigableSet;
33 import java.util.Set;
34 import java.util.TreeMap;
35 import java.util.TreeSet;
36 import java.util.UUID;
37 import java.util.concurrent.Callable;
38 import java.util.concurrent.CompletionService;
39 import java.util.concurrent.ConcurrentHashMap;
40 import java.util.concurrent.ExecutionException;
41 import java.util.concurrent.ExecutorCompletionService;
42 import java.util.concurrent.Future;
43 import java.util.concurrent.ThreadFactory;
44 import java.util.concurrent.ThreadPoolExecutor;
45 import java.util.concurrent.TimeUnit;
46 import java.util.concurrent.atomic.AtomicBoolean;
47 import java.util.concurrent.atomic.AtomicLong;
48 import java.util.concurrent.atomic.AtomicReference;
49 import java.util.regex.Matcher;
50 import java.util.regex.Pattern;
51
52 import org.apache.commons.logging.Log;
53 import org.apache.commons.logging.LogFactory;
54 import org.apache.hadoop.conf.Configuration;
55 import org.apache.hadoop.fs.FileAlreadyExistsException;
56 import org.apache.hadoop.fs.FileStatus;
57 import org.apache.hadoop.fs.FileSystem;
58 import org.apache.hadoop.fs.Path;
59 import org.apache.hadoop.fs.PathFilter;
60 import org.apache.hadoop.hbase.Cell;
61 import org.apache.hadoop.hbase.CellScanner;
62 import org.apache.hadoop.hbase.CellUtil;
63 import org.apache.hadoop.hbase.CoordinatedStateException;
64 import org.apache.hadoop.hbase.CoordinatedStateManager;
65 import org.apache.hadoop.hbase.HBaseConfiguration;
66 import org.apache.hadoop.hbase.HConstants;
67 import org.apache.hadoop.hbase.HRegionInfo;
68 import org.apache.hadoop.hbase.HRegionLocation;
69 import org.apache.hadoop.hbase.RemoteExceptionHandler;
70 import org.apache.hadoop.hbase.ServerName;
71 import org.apache.hadoop.hbase.TableName;
72 import org.apache.hadoop.hbase.TableNotFoundException;
73 import org.apache.hadoop.hbase.TableStateManager;
74 import org.apache.hadoop.hbase.classification.InterfaceAudience;
75 import org.apache.hadoop.hbase.client.ConnectionUtils;
76 import org.apache.hadoop.hbase.client.Delete;
77 import org.apache.hadoop.hbase.client.Durability;
78 import org.apache.hadoop.hbase.client.HConnection;
79 import org.apache.hadoop.hbase.client.HConnectionManager;
80 import org.apache.hadoop.hbase.client.Mutation;
81 import org.apache.hadoop.hbase.client.Put;
82 import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
83 import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
84 import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
85 import org.apache.hadoop.hbase.io.HeapSize;
86 import org.apache.hadoop.hbase.master.SplitLogManager;
87 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
88 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
89 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
90 import org.apache.hadoop.hbase.protobuf.RequestConverter;
91 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService.BlockingInterface;
92 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoRequest;
93 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse;
94 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
95 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
96 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
97 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
98 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
99 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
100 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
101 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
102 import org.apache.hadoop.hbase.regionserver.HRegion;
103 import org.apache.hadoop.hbase.regionserver.LastSequenceId;
104
105 import org.apache.hadoop.hbase.regionserver.wal.FSHLog;
106 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
107 import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
108 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
109 import org.apache.hadoop.hbase.regionserver.wal.WALEditsReplaySink;
110 import org.apache.hadoop.hbase.util.Bytes;
111 import org.apache.hadoop.hbase.util.CancelableProgressable;
112 import org.apache.hadoop.hbase.util.ClassSize;
113 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
114 import org.apache.hadoop.hbase.util.FSUtils;
115 import org.apache.hadoop.hbase.util.Pair;
116 import org.apache.hadoop.hbase.util.Threads;
117 import org.apache.hadoop.hbase.wal.WAL.Entry;
118 import org.apache.hadoop.hbase.wal.WAL.Reader;
119 import org.apache.hadoop.hbase.wal.WALProvider.Writer;
120 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
121 import org.apache.hadoop.io.MultipleIOException;
122
123 import com.google.common.annotations.VisibleForTesting;
124 import com.google.common.base.Preconditions;
125 import com.google.common.collect.Lists;
126 import com.google.protobuf.ServiceException;
127 import com.google.protobuf.TextFormat;
128
129
130
131
132
133
134 @InterfaceAudience.Private
135 public class WALSplitter {
136 private static final Log LOG = LogFactory.getLog(WALSplitter.class);
137
138
139 public static final boolean SPLIT_SKIP_ERRORS_DEFAULT = false;
140
141
142 protected final Path rootDir;
143 protected final FileSystem fs;
144 protected final Configuration conf;
145
146
147
148 PipelineController controller;
149 OutputSink outputSink;
150 EntryBuffers entryBuffers;
151
152 private Set<TableName> disablingOrDisabledTables =
153 new HashSet<TableName>();
154 private BaseCoordinatedStateManager csm;
155 private final WALFactory walFactory;
156
157 private MonitoredTask status;
158
159
160 protected final LastSequenceId sequenceIdChecker;
161
162 protected boolean distributedLogReplay;
163
164
165 protected Map<String, Long> lastFlushedSequenceIds = new ConcurrentHashMap<String, Long>();
166
167
168 protected Map<String, Map<byte[], Long>> regionMaxSeqIdInStores =
169 new ConcurrentHashMap<String, Map<byte[], Long>>();
170
171
172 protected String failedServerName = "";
173
174
175 private final int numWriterThreads;
176
177
178 private final int minBatchSize;
179
180 WALSplitter(final WALFactory factory, Configuration conf, Path rootDir,
181 FileSystem fs, LastSequenceId idChecker,
182 CoordinatedStateManager csm, RecoveryMode mode) {
183 this.conf = HBaseConfiguration.create(conf);
184 String codecClassName = conf
185 .get(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, WALCellCodec.class.getName());
186 this.conf.set(HConstants.RPC_CODEC_CONF_KEY, codecClassName);
187 this.rootDir = rootDir;
188 this.fs = fs;
189 this.sequenceIdChecker = idChecker;
190 this.csm = (BaseCoordinatedStateManager)csm;
191 this.walFactory = factory;
192 this.controller = new PipelineController();
193
194 entryBuffers = new EntryBuffers(controller,
195 this.conf.getInt("hbase.regionserver.hlog.splitlog.buffersize",
196 128*1024*1024));
197
198
199
200 this.minBatchSize = this.conf.getInt("hbase.regionserver.wal.logreplay.batch.size", 64);
201 this.distributedLogReplay = (RecoveryMode.LOG_REPLAY == mode);
202
203 this.numWriterThreads = this.conf.getInt("hbase.regionserver.hlog.splitlog.writer.threads", 3);
204 if (csm != null && this.distributedLogReplay) {
205 outputSink = new LogReplayOutputSink(controller, entryBuffers, numWriterThreads);
206 } else {
207 if (this.distributedLogReplay) {
208 LOG.info("ZooKeeperWatcher is passed in as NULL so disable distrubitedLogRepaly.");
209 }
210 this.distributedLogReplay = false;
211 outputSink = new LogRecoveredEditsOutputSink(controller, entryBuffers, numWriterThreads);
212 }
213
214 }
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232 public static boolean splitLogFile(Path rootDir, FileStatus logfile, FileSystem fs,
233 Configuration conf, CancelableProgressable reporter, LastSequenceId idChecker,
234 CoordinatedStateManager cp, RecoveryMode mode, final WALFactory factory) throws IOException {
235 WALSplitter s = new WALSplitter(factory, conf, rootDir, fs, idChecker, cp, mode);
236 return s.splitLogFile(logfile, reporter);
237 }
238
239
240
241
242
243 public static List<Path> split(Path rootDir, Path logDir, Path oldLogDir,
244 FileSystem fs, Configuration conf, final WALFactory factory) throws IOException {
245 final FileStatus[] logfiles = SplitLogManager.getFileList(conf,
246 Collections.singletonList(logDir), null);
247 List<Path> splits = new ArrayList<Path>();
248 if (logfiles != null && logfiles.length > 0) {
249 for (FileStatus logfile: logfiles) {
250 WALSplitter s = new WALSplitter(factory, conf, rootDir, fs, null, null,
251 RecoveryMode.LOG_SPLITTING);
252 if (s.splitLogFile(logfile, null)) {
253 finishSplitLogFile(rootDir, oldLogDir, logfile.getPath(), conf);
254 if (s.outputSink.splits != null) {
255 splits.addAll(s.outputSink.splits);
256 }
257 }
258 }
259 }
260 if (!fs.delete(logDir, true)) {
261 throw new IOException("Unable to delete src dir: " + logDir);
262 }
263 return splits;
264 }
265
266
267
268
269
270 boolean splitLogFile(FileStatus logfile, CancelableProgressable reporter) throws IOException {
271 Preconditions.checkState(status == null);
272 Preconditions.checkArgument(logfile.isFile(),
273 "passed in file status is for something other than a regular file.");
274 boolean isCorrupted = false;
275 boolean skipErrors = conf.getBoolean("hbase.hlog.split.skip.errors",
276 SPLIT_SKIP_ERRORS_DEFAULT);
277 int interval = conf.getInt("hbase.splitlog.report.interval.loglines", 1024);
278 Path logPath = logfile.getPath();
279 boolean outputSinkStarted = false;
280 boolean progress_failed = false;
281 int editsCount = 0;
282 int editsSkipped = 0;
283
284 status =
285 TaskMonitor.get().createStatus(
286 "Splitting log file " + logfile.getPath() + "into a temporary staging area.");
287 Reader in = null;
288 try {
289 long logLength = logfile.getLen();
290 LOG.info("Splitting wal: " + logPath + ", length=" + logLength);
291 LOG.info("DistributedLogReplay = " + this.distributedLogReplay);
292 status.setStatus("Opening log file");
293 if (reporter != null && !reporter.progress()) {
294 progress_failed = true;
295 return false;
296 }
297 try {
298 in = getReader(logfile, skipErrors, reporter);
299 } catch (CorruptedLogFileException e) {
300 LOG.warn("Could not get reader, corrupted log file " + logPath, e);
301 ZKSplitLog.markCorrupted(rootDir, logfile.getPath().getName(), fs);
302 isCorrupted = true;
303 }
304 if (in == null) {
305 LOG.warn("Nothing to split in log file " + logPath);
306 return true;
307 }
308 if (csm != null) {
309 try {
310 TableStateManager tsm = csm.getTableStateManager();
311 disablingOrDisabledTables = tsm.getTablesInStates(
312 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING);
313 } catch (CoordinatedStateException e) {
314 throw new IOException("Can't get disabling/disabled tables", e);
315 }
316 }
317 int numOpenedFilesBeforeReporting = conf.getInt("hbase.splitlog.report.openedfiles", 3);
318 int numOpenedFilesLastCheck = 0;
319 outputSink.setReporter(reporter);
320 outputSink.startWriterThreads();
321 outputSinkStarted = true;
322 Entry entry;
323 Long lastFlushedSequenceId = -1L;
324 ServerName serverName = DefaultWALProvider.getServerNameFromWALDirectoryName(logPath);
325 failedServerName = (serverName == null) ? "" : serverName.getServerName();
326 while ((entry = getNextLogLine(in, logPath, skipErrors)) != null) {
327 byte[] region = entry.getKey().getEncodedRegionName();
328 String encodedRegionNameAsStr = Bytes.toString(region);
329 lastFlushedSequenceId = lastFlushedSequenceIds.get(encodedRegionNameAsStr);
330 if (lastFlushedSequenceId == null) {
331 if (this.distributedLogReplay) {
332 RegionStoreSequenceIds ids =
333 csm.getSplitLogWorkerCoordination().getRegionFlushedSequenceId(failedServerName,
334 encodedRegionNameAsStr);
335 if (ids != null) {
336 lastFlushedSequenceId = ids.getLastFlushedSequenceId();
337 if (LOG.isDebugEnabled()) {
338 LOG.debug("DLR Last flushed sequenceid for " + encodedRegionNameAsStr + ": " +
339 TextFormat.shortDebugString(ids));
340 }
341 }
342 } else if (sequenceIdChecker != null) {
343 RegionStoreSequenceIds ids = sequenceIdChecker.getLastSequenceId(region);
344 Map<byte[], Long> maxSeqIdInStores = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
345 for (StoreSequenceId storeSeqId : ids.getStoreSequenceIdList()) {
346 maxSeqIdInStores.put(storeSeqId.getFamilyName().toByteArray(),
347 storeSeqId.getSequenceId());
348 }
349 regionMaxSeqIdInStores.put(encodedRegionNameAsStr, maxSeqIdInStores);
350 lastFlushedSequenceId = ids.getLastFlushedSequenceId();
351 if (LOG.isDebugEnabled()) {
352 LOG.debug("DLS Last flushed sequenceid for " + encodedRegionNameAsStr + ": " +
353 TextFormat.shortDebugString(ids));
354 }
355 }
356 if (lastFlushedSequenceId == null) {
357 lastFlushedSequenceId = -1L;
358 }
359 lastFlushedSequenceIds.put(encodedRegionNameAsStr, lastFlushedSequenceId);
360 }
361 if (lastFlushedSequenceId >= entry.getKey().getLogSeqNum()) {
362 editsSkipped++;
363 continue;
364 }
365 entryBuffers.appendEntry(entry);
366 editsCount++;
367 int moreWritersFromLastCheck = this.getNumOpenWriters() - numOpenedFilesLastCheck;
368
369 if (editsCount % interval == 0
370 || moreWritersFromLastCheck > numOpenedFilesBeforeReporting) {
371 numOpenedFilesLastCheck = this.getNumOpenWriters();
372 String countsStr = (editsCount - (editsSkipped + outputSink.getSkippedEdits()))
373 + " edits, skipped " + editsSkipped + " edits.";
374 status.setStatus("Split " + countsStr);
375 if (reporter != null && !reporter.progress()) {
376 progress_failed = true;
377 return false;
378 }
379 }
380 }
381 } catch (InterruptedException ie) {
382 IOException iie = new InterruptedIOException();
383 iie.initCause(ie);
384 throw iie;
385 } catch (CorruptedLogFileException e) {
386 LOG.warn("Could not parse, corrupted log file " + logPath, e);
387 csm.getSplitLogWorkerCoordination().markCorrupted(rootDir,
388 logfile.getPath().getName(), fs);
389 isCorrupted = true;
390 } catch (IOException e) {
391 e = RemoteExceptionHandler.checkIOException(e);
392 throw e;
393 } finally {
394 LOG.debug("Finishing writing output logs and closing down.");
395 try {
396 if (null != in) {
397 in.close();
398 }
399 } catch (IOException exception) {
400 LOG.warn("Could not close wal reader: " + exception.getMessage());
401 LOG.debug("exception details", exception);
402 }
403 try {
404 if (outputSinkStarted) {
405
406
407 progress_failed = true;
408 progress_failed = outputSink.finishWritingAndClose() == null;
409 }
410 } finally {
411 String msg =
412 "Processed " + editsCount + " edits across " + outputSink.getNumberOfRecoveredRegions()
413 + " regions; edits skipped=" + editsSkipped + "; log file=" + logPath +
414 ", length=" + logfile.getLen() +
415 ", corrupted=" + isCorrupted + ", progress failed=" + progress_failed;
416 LOG.info(msg);
417 status.markComplete(msg);
418 }
419 }
420 return !progress_failed;
421 }
422
423
424
425
426
427
428
429
430
431
432
433
434 public static void finishSplitLogFile(String logfile,
435 Configuration conf) throws IOException {
436 Path rootdir = FSUtils.getRootDir(conf);
437 Path oldLogDir = new Path(rootdir, HConstants.HREGION_OLDLOGDIR_NAME);
438 Path logPath;
439 if (FSUtils.isStartingWithPath(rootdir, logfile)) {
440 logPath = new Path(logfile);
441 } else {
442 logPath = new Path(rootdir, logfile);
443 }
444 finishSplitLogFile(rootdir, oldLogDir, logPath, conf);
445 }
446
447 static void finishSplitLogFile(Path rootdir, Path oldLogDir,
448 Path logPath, Configuration conf) throws IOException {
449 List<Path> processedLogs = new ArrayList<Path>();
450 List<Path> corruptedLogs = new ArrayList<Path>();
451 FileSystem fs;
452 fs = rootdir.getFileSystem(conf);
453 if (ZKSplitLog.isCorrupted(rootdir, logPath.getName(), fs)) {
454 corruptedLogs.add(logPath);
455 } else {
456 processedLogs.add(logPath);
457 }
458 archiveLogs(corruptedLogs, processedLogs, oldLogDir, fs, conf);
459 Path stagingDir = ZKSplitLog.getSplitLogDir(rootdir, logPath.getName());
460 fs.delete(stagingDir, true);
461 }
462
463
464
465
466
467
468
469
470
471
472
473
474
475 private static void archiveLogs(
476 final List<Path> corruptedLogs,
477 final List<Path> processedLogs, final Path oldLogDir,
478 final FileSystem fs, final Configuration conf) throws IOException {
479 final Path corruptDir = new Path(FSUtils.getRootDir(conf), conf.get(
480 "hbase.regionserver.hlog.splitlog.corrupt.dir", HConstants.CORRUPT_DIR_NAME));
481
482 if (!fs.mkdirs(corruptDir)) {
483 LOG.info("Unable to mkdir " + corruptDir);
484 }
485 fs.mkdirs(oldLogDir);
486
487
488
489 for (Path corrupted : corruptedLogs) {
490 Path p = new Path(corruptDir, corrupted.getName());
491 if (fs.exists(corrupted)) {
492 if (!fs.rename(corrupted, p)) {
493 LOG.warn("Unable to move corrupted log " + corrupted + " to " + p);
494 } else {
495 LOG.warn("Moved corrupted log " + corrupted + " to " + p);
496 }
497 }
498 }
499
500 for (Path p : processedLogs) {
501 Path newPath = FSHLog.getWALArchivePath(oldLogDir, p);
502 if (fs.exists(p)) {
503 if (!FSUtils.renameAndSetModifyTime(fs, p, newPath)) {
504 LOG.warn("Unable to move " + p + " to " + newPath);
505 } else {
506 LOG.info("Archived processed log " + p + " to " + newPath);
507 }
508 }
509 }
510 }
511
512
513
514
515
516
517
518
519
520
521
522
523
524 @SuppressWarnings("deprecation")
525 static Path getRegionSplitEditsPath(final FileSystem fs,
526 final Entry logEntry, final Path rootDir, boolean isCreate)
527 throws IOException {
528 Path tableDir = FSUtils.getTableDir(rootDir, logEntry.getKey().getTablename());
529 String encodedRegionName = Bytes.toString(logEntry.getKey().getEncodedRegionName());
530 Path regiondir = HRegion.getRegionDir(tableDir, encodedRegionName);
531 Path dir = getRegionDirRecoveredEditsDir(regiondir);
532
533 if (!fs.exists(regiondir)) {
534 LOG.info("This region's directory doesn't exist: "
535 + regiondir.toString() + ". It is very likely that it was" +
536 " already split so it's safe to discard those edits.");
537 return null;
538 }
539 if (fs.exists(dir) && fs.isFile(dir)) {
540 Path tmp = new Path("/tmp");
541 if (!fs.exists(tmp)) {
542 fs.mkdirs(tmp);
543 }
544 tmp = new Path(tmp,
545 HConstants.RECOVERED_EDITS_DIR + "_" + encodedRegionName);
546 LOG.warn("Found existing old file: " + dir + ". It could be some "
547 + "leftover of an old installation. It should be a folder instead. "
548 + "So moving it to " + tmp);
549 if (!fs.rename(dir, tmp)) {
550 LOG.warn("Failed to sideline old file " + dir);
551 }
552 }
553
554 if (isCreate && !fs.exists(dir)) {
555 if (!fs.mkdirs(dir)) LOG.warn("mkdir failed on " + dir);
556 }
557
558
559 String fileName = formatRecoveredEditsFileName(logEntry.getKey().getLogSeqNum());
560 fileName = getTmpRecoveredEditsFileName(fileName);
561 return new Path(dir, fileName);
562 }
563
564 static String getTmpRecoveredEditsFileName(String fileName) {
565 return fileName + RECOVERED_LOG_TMPFILE_SUFFIX;
566 }
567
568
569
570
571
572
573
574
575
576 static Path getCompletedRecoveredEditsFilePath(Path srcPath,
577 Long maximumEditLogSeqNum) {
578 String fileName = formatRecoveredEditsFileName(maximumEditLogSeqNum);
579 return new Path(srcPath.getParent(), fileName);
580 }
581
582 static String formatRecoveredEditsFileName(final long seqid) {
583 return String.format("%019d", seqid);
584 }
585
586 private static final Pattern EDITFILES_NAME_PATTERN = Pattern.compile("-?[0-9]+");
587 private static final String RECOVERED_LOG_TMPFILE_SUFFIX = ".temp";
588
589
590
591
592
593
594
595 public static Path getRegionDirRecoveredEditsDir(final Path regiondir) {
596 return new Path(regiondir, HConstants.RECOVERED_EDITS_DIR);
597 }
598
599
600
601
602
603
604
605
606
607
608 public static NavigableSet<Path> getSplitEditFilesSorted(final FileSystem fs,
609 final Path regiondir) throws IOException {
610 NavigableSet<Path> filesSorted = new TreeSet<Path>();
611 Path editsdir = getRegionDirRecoveredEditsDir(regiondir);
612 if (!fs.exists(editsdir))
613 return filesSorted;
614 FileStatus[] files = FSUtils.listStatus(fs, editsdir, new PathFilter() {
615 @Override
616 public boolean accept(Path p) {
617 boolean result = false;
618 try {
619
620
621
622
623 Matcher m = EDITFILES_NAME_PATTERN.matcher(p.getName());
624 result = fs.isFile(p) && m.matches();
625
626
627 if (p.getName().endsWith(RECOVERED_LOG_TMPFILE_SUFFIX)) {
628 result = false;
629 }
630
631 if (isSequenceIdFile(p)) {
632 result = false;
633 }
634 } catch (IOException e) {
635 LOG.warn("Failed isFile check on " + p);
636 }
637 return result;
638 }
639 });
640 if (files == null) {
641 return filesSorted;
642 }
643 for (FileStatus status : files) {
644 filesSorted.add(status.getPath());
645 }
646 return filesSorted;
647 }
648
649
650
651
652
653
654
655
656
657
658 public static Path moveAsideBadEditsFile(final FileSystem fs, final Path edits)
659 throws IOException {
660 Path moveAsideName = new Path(edits.getParent(), edits.getName() + "."
661 + System.currentTimeMillis());
662 if (!fs.rename(edits, moveAsideName)) {
663 LOG.warn("Rename failed from " + edits + " to " + moveAsideName);
664 }
665 return moveAsideName;
666 }
667
668 private static final String SEQUENCE_ID_FILE_SUFFIX = ".seqid";
669 private static final String OLD_SEQUENCE_ID_FILE_SUFFIX = "_seqid";
670 private static final int SEQUENCE_ID_FILE_SUFFIX_LENGTH = SEQUENCE_ID_FILE_SUFFIX.length();
671
672
673
674
675 @VisibleForTesting
676 public static boolean isSequenceIdFile(final Path file) {
677 return file.getName().endsWith(SEQUENCE_ID_FILE_SUFFIX)
678 || file.getName().endsWith(OLD_SEQUENCE_ID_FILE_SUFFIX);
679 }
680
681
682
683
684
685
686
687
688
689
690 public static long writeRegionSequenceIdFile(final FileSystem fs, final Path regiondir,
691 long newSeqId, long saftyBumper) throws IOException {
692
693 Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(regiondir);
694 long maxSeqId = 0;
695 FileStatus[] files = null;
696 if (fs.exists(editsdir)) {
697 files = FSUtils.listStatus(fs, editsdir, new PathFilter() {
698 @Override
699 public boolean accept(Path p) {
700 return isSequenceIdFile(p);
701 }
702 });
703 if (files != null) {
704 for (FileStatus status : files) {
705 String fileName = status.getPath().getName();
706 try {
707 Long tmpSeqId = Long.parseLong(fileName.substring(0, fileName.length()
708 - SEQUENCE_ID_FILE_SUFFIX_LENGTH));
709 maxSeqId = Math.max(tmpSeqId, maxSeqId);
710 } catch (NumberFormatException ex) {
711 LOG.warn("Invalid SeqId File Name=" + fileName);
712 }
713 }
714 }
715 }
716 if (maxSeqId > newSeqId) {
717 newSeqId = maxSeqId;
718 }
719 newSeqId += saftyBumper;
720
721
722 Path newSeqIdFile = new Path(editsdir, newSeqId + SEQUENCE_ID_FILE_SUFFIX);
723 if (newSeqId != maxSeqId) {
724 try {
725 if (!fs.createNewFile(newSeqIdFile) && !fs.exists(newSeqIdFile)) {
726 throw new IOException("Failed to create SeqId file:" + newSeqIdFile);
727 }
728 if (LOG.isDebugEnabled()) {
729 LOG.debug("Wrote region seqId=" + newSeqIdFile + " to file, newSeqId=" + newSeqId
730 + ", maxSeqId=" + maxSeqId);
731 }
732 } catch (FileAlreadyExistsException ignored) {
733
734 }
735 }
736
737 if (files != null) {
738 for (FileStatus status : files) {
739 if (newSeqIdFile.equals(status.getPath())) {
740 continue;
741 }
742 fs.delete(status.getPath(), false);
743 }
744 }
745 return newSeqId;
746 }
747
748
749
750
751
752
753
754
755
756 protected Reader getReader(FileStatus file, boolean skipErrors, CancelableProgressable reporter)
757 throws IOException, CorruptedLogFileException {
758 Path path = file.getPath();
759 long length = file.getLen();
760 Reader in;
761
762
763
764
765 if (length <= 0) {
766 LOG.warn("File " + path + " might be still open, length is 0");
767 }
768
769 try {
770 FSUtils.getInstance(fs, conf).recoverFileLease(fs, path, conf, reporter);
771 try {
772 in = getReader(path, reporter);
773 } catch (EOFException e) {
774 if (length <= 0) {
775
776
777
778
779
780 LOG.warn("Could not open " + path + " for reading. File is empty", e);
781 return null;
782 } else {
783
784 return null;
785 }
786 }
787 } catch (IOException e) {
788 if (e instanceof FileNotFoundException) {
789
790 LOG.warn("File " + path + " doesn't exist anymore.", e);
791 return null;
792 }
793 if (!skipErrors || e instanceof InterruptedIOException) {
794 throw e;
795 }
796 CorruptedLogFileException t =
797 new CorruptedLogFileException("skipErrors=true Could not open wal " +
798 path + " ignoring");
799 t.initCause(e);
800 throw t;
801 }
802 return in;
803 }
804
805 static private Entry getNextLogLine(Reader in, Path path, boolean skipErrors)
806 throws CorruptedLogFileException, IOException {
807 try {
808 return in.next();
809 } catch (EOFException eof) {
810
811 LOG.info("EOF from wal " + path + ". continuing");
812 return null;
813 } catch (IOException e) {
814
815
816 if (e.getCause() != null &&
817 (e.getCause() instanceof ParseException ||
818 e.getCause() instanceof org.apache.hadoop.fs.ChecksumException)) {
819 LOG.warn("Parse exception " + e.getCause().toString() + " from wal "
820 + path + ". continuing");
821 return null;
822 }
823 if (!skipErrors) {
824 throw e;
825 }
826 CorruptedLogFileException t =
827 new CorruptedLogFileException("skipErrors=true Ignoring exception" +
828 " while parsing wal " + path + ". Marking as corrupted");
829 t.initCause(e);
830 throw t;
831 }
832 }
833
834
835
836
837
838 protected Writer createWriter(Path logfile)
839 throws IOException {
840 return walFactory.createRecoveredEditsWriter(fs, logfile);
841 }
842
843
844
845
846
847 protected Reader getReader(Path curLogFile, CancelableProgressable reporter) throws IOException {
848 return walFactory.createReader(fs, curLogFile, reporter);
849 }
850
851
852
853
854 private int getNumOpenWriters() {
855 int result = 0;
856 if (this.outputSink != null) {
857 result += this.outputSink.getNumOpenWriters();
858 }
859 return result;
860 }
861
862
863
864
865 public static class PipelineController {
866
867
868 AtomicReference<Throwable> thrown = new AtomicReference<Throwable>();
869
870
871
872 public final Object dataAvailable = new Object();
873
874 void writerThreadError(Throwable t) {
875 thrown.compareAndSet(null, t);
876 }
877
878
879
880
881 void checkForErrors() throws IOException {
882 Throwable thrown = this.thrown.get();
883 if (thrown == null) return;
884 if (thrown instanceof IOException) {
885 throw new IOException(thrown);
886 } else {
887 throw new RuntimeException(thrown);
888 }
889 }
890 }
891
892
893
894
895
896
897
898
899 public static class EntryBuffers {
900 PipelineController controller;
901
902 Map<byte[], RegionEntryBuffer> buffers =
903 new TreeMap<byte[], RegionEntryBuffer>(Bytes.BYTES_COMPARATOR);
904
905
906
907
908 Set<byte[]> currentlyWriting = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
909
910 long totalBuffered = 0;
911 long maxHeapUsage;
912
913 public EntryBuffers(PipelineController controller, long maxHeapUsage) {
914 this.controller = controller;
915 this.maxHeapUsage = maxHeapUsage;
916 }
917
918
919
920
921
922
923
924
925 public void appendEntry(Entry entry) throws InterruptedException, IOException {
926 WALKey key = entry.getKey();
927
928 RegionEntryBuffer buffer;
929 long incrHeap;
930 synchronized (this) {
931 buffer = buffers.get(key.getEncodedRegionName());
932 if (buffer == null) {
933 buffer = new RegionEntryBuffer(key.getTablename(), key.getEncodedRegionName());
934 buffers.put(key.getEncodedRegionName(), buffer);
935 }
936 incrHeap= buffer.appendEntry(entry);
937 }
938
939
940 synchronized (controller.dataAvailable) {
941 totalBuffered += incrHeap;
942 while (totalBuffered > maxHeapUsage && controller.thrown.get() == null) {
943 LOG.debug("Used " + totalBuffered + " bytes of buffered edits, waiting for IO threads...");
944 controller.dataAvailable.wait(2000);
945 }
946 controller.dataAvailable.notifyAll();
947 }
948 controller.checkForErrors();
949 }
950
951
952
953
954 synchronized RegionEntryBuffer getChunkToWrite() {
955 long biggestSize = 0;
956 byte[] biggestBufferKey = null;
957
958 for (Map.Entry<byte[], RegionEntryBuffer> entry : buffers.entrySet()) {
959 long size = entry.getValue().heapSize();
960 if (size > biggestSize && (!currentlyWriting.contains(entry.getKey()))) {
961 biggestSize = size;
962 biggestBufferKey = entry.getKey();
963 }
964 }
965 if (biggestBufferKey == null) {
966 return null;
967 }
968
969 RegionEntryBuffer buffer = buffers.remove(biggestBufferKey);
970 currentlyWriting.add(biggestBufferKey);
971 return buffer;
972 }
973
974 void doneWriting(RegionEntryBuffer buffer) {
975 synchronized (this) {
976 boolean removed = currentlyWriting.remove(buffer.encodedRegionName);
977 assert removed;
978 }
979 long size = buffer.heapSize();
980
981 synchronized (controller.dataAvailable) {
982 totalBuffered -= size;
983
984 controller.dataAvailable.notifyAll();
985 }
986 }
987
988 synchronized boolean isRegionCurrentlyWriting(byte[] region) {
989 return currentlyWriting.contains(region);
990 }
991
992 public void waitUntilDrained() {
993 synchronized (controller.dataAvailable) {
994 while (totalBuffered > 0) {
995 try {
996 controller.dataAvailable.wait(2000);
997 } catch (InterruptedException e) {
998 LOG.warn("Got intrerrupted while waiting for EntryBuffers is drained");
999 Thread.interrupted();
1000 break;
1001 }
1002 }
1003 }
1004 }
1005 }
1006
1007
1008
1009
1010
1011
1012
1013 public static class RegionEntryBuffer implements HeapSize {
1014 long heapInBuffer = 0;
1015 List<Entry> entryBuffer;
1016 TableName tableName;
1017 byte[] encodedRegionName;
1018
1019 RegionEntryBuffer(TableName tableName, byte[] region) {
1020 this.tableName = tableName;
1021 this.encodedRegionName = region;
1022 this.entryBuffer = new LinkedList<Entry>();
1023 }
1024
1025 long appendEntry(Entry entry) {
1026 internify(entry);
1027 entryBuffer.add(entry);
1028 long incrHeap = entry.getEdit().heapSize() +
1029 ClassSize.align(2 * ClassSize.REFERENCE) +
1030 0;
1031 heapInBuffer += incrHeap;
1032 return incrHeap;
1033 }
1034
1035 private void internify(Entry entry) {
1036 WALKey k = entry.getKey();
1037 k.internTableName(this.tableName);
1038 k.internEncodedRegionName(this.encodedRegionName);
1039 }
1040
1041 @Override
1042 public long heapSize() {
1043 return heapInBuffer;
1044 }
1045
1046 public byte[] getEncodedRegionName() {
1047 return encodedRegionName;
1048 }
1049
1050 public List<Entry> getEntryBuffer() {
1051 return entryBuffer;
1052 }
1053
1054 public TableName getTableName() {
1055 return tableName;
1056 }
1057 }
1058
1059 public static class WriterThread extends Thread {
1060 private volatile boolean shouldStop = false;
1061 private PipelineController controller;
1062 private EntryBuffers entryBuffers;
1063 private OutputSink outputSink = null;
1064
1065 WriterThread(PipelineController controller, EntryBuffers entryBuffers, OutputSink sink, int i){
1066 super(Thread.currentThread().getName() + "-Writer-" + i);
1067 this.controller = controller;
1068 this.entryBuffers = entryBuffers;
1069 outputSink = sink;
1070 }
1071
1072 @Override
1073 public void run() {
1074 try {
1075 doRun();
1076 } catch (Throwable t) {
1077 LOG.error("Exiting thread", t);
1078 controller.writerThreadError(t);
1079 }
1080 }
1081
1082 private void doRun() throws IOException {
1083 if (LOG.isTraceEnabled()) LOG.trace("Writer thread starting");
1084 while (true) {
1085 RegionEntryBuffer buffer = entryBuffers.getChunkToWrite();
1086 if (buffer == null) {
1087
1088 synchronized (controller.dataAvailable) {
1089 if (shouldStop && !this.outputSink.flush()) {
1090 return;
1091 }
1092 try {
1093 controller.dataAvailable.wait(500);
1094 } catch (InterruptedException ie) {
1095 if (!shouldStop) {
1096 throw new RuntimeException(ie);
1097 }
1098 }
1099 }
1100 continue;
1101 }
1102
1103 assert buffer != null;
1104 try {
1105 writeBuffer(buffer);
1106 } finally {
1107 entryBuffers.doneWriting(buffer);
1108 }
1109 }
1110 }
1111
1112 private void writeBuffer(RegionEntryBuffer buffer) throws IOException {
1113 outputSink.append(buffer);
1114 }
1115
1116 void finish() {
1117 synchronized (controller.dataAvailable) {
1118 shouldStop = true;
1119 controller.dataAvailable.notifyAll();
1120 }
1121 }
1122 }
1123
1124
1125
1126
1127
1128 public static abstract class OutputSink {
1129
1130 protected PipelineController controller;
1131 protected EntryBuffers entryBuffers;
1132
1133 protected Map<byte[], SinkWriter> writers = Collections
1134 .synchronizedMap(new TreeMap<byte[], SinkWriter>(Bytes.BYTES_COMPARATOR));;
1135
1136 protected final Map<byte[], Long> regionMaximumEditLogSeqNum = Collections
1137 .synchronizedMap(new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR));
1138
1139 protected final List<WriterThread> writerThreads = Lists.newArrayList();
1140
1141
1142 protected final Set<byte[]> blacklistedRegions = Collections
1143 .synchronizedSet(new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR));
1144
1145 protected boolean closeAndCleanCompleted = false;
1146
1147 protected boolean writersClosed = false;
1148
1149 protected final int numThreads;
1150
1151 protected CancelableProgressable reporter = null;
1152
1153 protected AtomicLong skippedEdits = new AtomicLong();
1154
1155 protected List<Path> splits = null;
1156
1157 public OutputSink(PipelineController controller, EntryBuffers entryBuffers, int numWriters) {
1158 numThreads = numWriters;
1159 this.controller = controller;
1160 this.entryBuffers = entryBuffers;
1161 }
1162
1163 void setReporter(CancelableProgressable reporter) {
1164 this.reporter = reporter;
1165 }
1166
1167
1168
1169
1170 public synchronized void startWriterThreads() {
1171 for (int i = 0; i < numThreads; i++) {
1172 WriterThread t = new WriterThread(controller, entryBuffers, this, i);
1173 t.start();
1174 writerThreads.add(t);
1175 }
1176 }
1177
1178
1179
1180
1181
1182 void updateRegionMaximumEditLogSeqNum(Entry entry) {
1183 synchronized (regionMaximumEditLogSeqNum) {
1184 Long currentMaxSeqNum = regionMaximumEditLogSeqNum.get(entry.getKey()
1185 .getEncodedRegionName());
1186 if (currentMaxSeqNum == null || entry.getKey().getLogSeqNum() > currentMaxSeqNum) {
1187 regionMaximumEditLogSeqNum.put(entry.getKey().getEncodedRegionName(), entry.getKey()
1188 .getLogSeqNum());
1189 }
1190 }
1191 }
1192
1193 Long getRegionMaximumEditLogSeqNum(byte[] region) {
1194 return regionMaximumEditLogSeqNum.get(region);
1195 }
1196
1197
1198
1199
1200 int getNumOpenWriters() {
1201 return this.writers.size();
1202 }
1203
1204 long getSkippedEdits() {
1205 return this.skippedEdits.get();
1206 }
1207
1208
1209
1210
1211
1212
1213 protected boolean finishWriting(boolean interrupt) throws IOException {
1214 LOG.debug("Waiting for split writer threads to finish");
1215 boolean progress_failed = false;
1216 for (WriterThread t : writerThreads) {
1217 t.finish();
1218 }
1219 if (interrupt) {
1220 for (WriterThread t : writerThreads) {
1221 t.interrupt();
1222 }
1223 }
1224
1225 for (WriterThread t : writerThreads) {
1226 if (!progress_failed && reporter != null && !reporter.progress()) {
1227 progress_failed = true;
1228 }
1229 try {
1230 t.join();
1231 } catch (InterruptedException ie) {
1232 IOException iie = new InterruptedIOException();
1233 iie.initCause(ie);
1234 throw iie;
1235 }
1236 }
1237 controller.checkForErrors();
1238 LOG.info(this.writerThreads.size() + " split writers finished; closing...");
1239 return (!progress_failed);
1240 }
1241
1242 public abstract List<Path> finishWritingAndClose() throws IOException;
1243
1244
1245
1246
1247 public abstract Map<byte[], Long> getOutputCounts();
1248
1249
1250
1251
1252 public abstract int getNumberOfRecoveredRegions();
1253
1254
1255
1256
1257
1258 public abstract void append(RegionEntryBuffer buffer) throws IOException;
1259
1260
1261
1262
1263
1264 public boolean flush() throws IOException {
1265 return false;
1266 }
1267 }
1268
1269
1270
1271
1272 class LogRecoveredEditsOutputSink extends OutputSink {
1273
1274 public LogRecoveredEditsOutputSink(PipelineController controller, EntryBuffers entryBuffers,
1275 int numWriters) {
1276
1277
1278
1279
1280
1281 super(controller, entryBuffers, numWriters);
1282 }
1283
1284
1285
1286
1287
1288 @Override
1289 public List<Path> finishWritingAndClose() throws IOException {
1290 boolean isSuccessful = false;
1291 List<Path> result = null;
1292 try {
1293 isSuccessful = finishWriting(false);
1294 } finally {
1295 result = close();
1296 List<IOException> thrown = closeLogWriters(null);
1297 if (thrown != null && !thrown.isEmpty()) {
1298 throw MultipleIOException.createIOException(thrown);
1299 }
1300 }
1301 if (isSuccessful) {
1302 splits = result;
1303 }
1304 return splits;
1305 }
1306
1307
1308
1309
1310
1311 private List<Path> close() throws IOException {
1312 Preconditions.checkState(!closeAndCleanCompleted);
1313
1314 final List<Path> paths = new ArrayList<Path>();
1315 final List<IOException> thrown = Lists.newArrayList();
1316 ThreadPoolExecutor closeThreadPool = Threads.getBoundedCachedThreadPool(numThreads, 30L,
1317 TimeUnit.SECONDS, new ThreadFactory() {
1318 private int count = 1;
1319
1320 @Override
1321 public Thread newThread(Runnable r) {
1322 Thread t = new Thread(r, "split-log-closeStream-" + count++);
1323 return t;
1324 }
1325 });
1326 CompletionService<Void> completionService =
1327 new ExecutorCompletionService<Void>(closeThreadPool);
1328 for (final Map.Entry<byte[], SinkWriter> writersEntry : writers.entrySet()) {
1329 if (LOG.isTraceEnabled()) {
1330 LOG.trace("Submitting close of " + ((WriterAndPath)writersEntry.getValue()).p);
1331 }
1332 completionService.submit(new Callable<Void>() {
1333 @Override
1334 public Void call() throws Exception {
1335 WriterAndPath wap = (WriterAndPath) writersEntry.getValue();
1336 if (LOG.isTraceEnabled()) LOG.trace("Closing " + wap.p);
1337 try {
1338 wap.w.close();
1339 } catch (IOException ioe) {
1340 LOG.error("Couldn't close log at " + wap.p, ioe);
1341 thrown.add(ioe);
1342 return null;
1343 }
1344 if (LOG.isDebugEnabled()) {
1345 LOG.debug("Closed wap " + wap.p + " (wrote " + wap.editsWritten
1346 + " edits, skipped " + wap.editsSkipped + " edits in "
1347 + (wap.nanosSpent / 1000 / 1000) + "ms");
1348 }
1349 if (wap.editsWritten == 0) {
1350
1351 if (fs.exists(wap.p) && !fs.delete(wap.p, false)) {
1352 LOG.warn("Failed deleting empty " + wap.p);
1353 throw new IOException("Failed deleting empty " + wap.p);
1354 }
1355 return null;
1356 }
1357
1358 Path dst = getCompletedRecoveredEditsFilePath(wap.p,
1359 regionMaximumEditLogSeqNum.get(writersEntry.getKey()));
1360 try {
1361 if (!dst.equals(wap.p) && fs.exists(dst)) {
1362 LOG.warn("Found existing old edits file. It could be the "
1363 + "result of a previous failed split attempt. Deleting " + dst + ", length="
1364 + fs.getFileStatus(dst).getLen());
1365 if (!fs.delete(dst, false)) {
1366 LOG.warn("Failed deleting of old " + dst);
1367 throw new IOException("Failed deleting of old " + dst);
1368 }
1369 }
1370
1371
1372
1373 if (fs.exists(wap.p)) {
1374 if (!fs.rename(wap.p, dst)) {
1375 throw new IOException("Failed renaming " + wap.p + " to " + dst);
1376 }
1377 LOG.info("Rename " + wap.p + " to " + dst);
1378 }
1379 } catch (IOException ioe) {
1380 LOG.error("Couldn't rename " + wap.p + " to " + dst, ioe);
1381 thrown.add(ioe);
1382 return null;
1383 }
1384 paths.add(dst);
1385 return null;
1386 }
1387 });
1388 }
1389
1390 boolean progress_failed = false;
1391 try {
1392 for (int i = 0, n = this.writers.size(); i < n; i++) {
1393 Future<Void> future = completionService.take();
1394 future.get();
1395 if (!progress_failed && reporter != null && !reporter.progress()) {
1396 progress_failed = true;
1397 }
1398 }
1399 } catch (InterruptedException e) {
1400 IOException iie = new InterruptedIOException();
1401 iie.initCause(e);
1402 throw iie;
1403 } catch (ExecutionException e) {
1404 throw new IOException(e.getCause());
1405 } finally {
1406 closeThreadPool.shutdownNow();
1407 }
1408
1409 if (!thrown.isEmpty()) {
1410 throw MultipleIOException.createIOException(thrown);
1411 }
1412 writersClosed = true;
1413 closeAndCleanCompleted = true;
1414 if (progress_failed) {
1415 return null;
1416 }
1417 return paths;
1418 }
1419
1420 private List<IOException> closeLogWriters(List<IOException> thrown) throws IOException {
1421 if (writersClosed) {
1422 return thrown;
1423 }
1424
1425 if (thrown == null) {
1426 thrown = Lists.newArrayList();
1427 }
1428 try {
1429 for (WriterThread t : writerThreads) {
1430 while (t.isAlive()) {
1431 t.shouldStop = true;
1432 t.interrupt();
1433 try {
1434 t.join(10);
1435 } catch (InterruptedException e) {
1436 IOException iie = new InterruptedIOException();
1437 iie.initCause(e);
1438 throw iie;
1439 }
1440 }
1441 }
1442 } finally {
1443 synchronized (writers) {
1444 WriterAndPath wap = null;
1445 for (SinkWriter tmpWAP : writers.values()) {
1446 try {
1447 wap = (WriterAndPath) tmpWAP;
1448 wap.w.close();
1449 } catch (IOException ioe) {
1450 LOG.error("Couldn't close log at " + wap.p, ioe);
1451 thrown.add(ioe);
1452 continue;
1453 }
1454 LOG.info("Closed log " + wap.p + " (wrote " + wap.editsWritten + " edits in "
1455 + (wap.nanosSpent / 1000 / 1000) + "ms)");
1456 }
1457 }
1458 writersClosed = true;
1459 }
1460
1461 return thrown;
1462 }
1463
1464
1465
1466
1467
1468
1469 private WriterAndPath getWriterAndPath(Entry entry) throws IOException {
1470 byte region[] = entry.getKey().getEncodedRegionName();
1471 WriterAndPath ret = (WriterAndPath) writers.get(region);
1472 if (ret != null) {
1473 return ret;
1474 }
1475
1476
1477 if (blacklistedRegions.contains(region)) {
1478 return null;
1479 }
1480 ret = createWAP(region, entry, rootDir);
1481 if (ret == null) {
1482 blacklistedRegions.add(region);
1483 return null;
1484 }
1485 writers.put(region, ret);
1486 return ret;
1487 }
1488
1489
1490
1491
1492 private WriterAndPath createWAP(byte[] region, Entry entry, Path rootdir) throws IOException {
1493 Path regionedits = getRegionSplitEditsPath(fs, entry, rootdir, true);
1494 if (regionedits == null) {
1495 return null;
1496 }
1497 if (fs.exists(regionedits)) {
1498 LOG.warn("Found old edits file. It could be the "
1499 + "result of a previous failed split attempt. Deleting " + regionedits + ", length="
1500 + fs.getFileStatus(regionedits).getLen());
1501 if (!fs.delete(regionedits, false)) {
1502 LOG.warn("Failed delete of old " + regionedits);
1503 }
1504 }
1505 Writer w = createWriter(regionedits);
1506 LOG.debug("Creating writer path=" + regionedits);
1507 return new WriterAndPath(regionedits, w);
1508 }
1509
1510 private void filterCellByStore(Entry logEntry) {
1511 Map<byte[], Long> maxSeqIdInStores =
1512 regionMaxSeqIdInStores.get(Bytes.toString(logEntry.getKey().getEncodedRegionName()));
1513 if (maxSeqIdInStores == null || maxSeqIdInStores.isEmpty()) {
1514 return;
1515 }
1516
1517
1518 ArrayList<Cell> keptCells = new ArrayList<Cell>(logEntry.getEdit().getCells().size());
1519 for (Cell cell : logEntry.getEdit().getCells()) {
1520 if (CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
1521 keptCells.add(cell);
1522 } else {
1523 byte[] family = CellUtil.cloneFamily(cell);
1524 Long maxSeqId = maxSeqIdInStores.get(family);
1525
1526
1527 if (maxSeqId == null || maxSeqId.longValue() < logEntry.getKey().getLogSeqNum()) {
1528 keptCells.add(cell);
1529 }
1530 }
1531 }
1532
1533
1534
1535
1536 logEntry.getEdit().setCells(keptCells);
1537 }
1538
1539 @Override
1540 public void append(RegionEntryBuffer buffer) throws IOException {
1541 List<Entry> entries = buffer.entryBuffer;
1542 if (entries.isEmpty()) {
1543 LOG.warn("got an empty buffer, skipping");
1544 return;
1545 }
1546
1547 WriterAndPath wap = null;
1548
1549 long startTime = System.nanoTime();
1550 try {
1551 int editsCount = 0;
1552
1553 for (Entry logEntry : entries) {
1554 if (wap == null) {
1555 wap = getWriterAndPath(logEntry);
1556 if (wap == null) {
1557 if (LOG.isDebugEnabled()) {
1558 LOG.debug("getWriterAndPath decided we don't need to write edits for " + logEntry);
1559 }
1560 return;
1561 }
1562 }
1563 filterCellByStore(logEntry);
1564 if (!logEntry.getEdit().isEmpty()) {
1565 wap.w.append(logEntry);
1566 this.updateRegionMaximumEditLogSeqNum(logEntry);
1567 editsCount++;
1568 } else {
1569 wap.incrementSkippedEdits(1);
1570 }
1571 }
1572
1573 wap.incrementEdits(editsCount);
1574 wap.incrementNanoTime(System.nanoTime() - startTime);
1575 } catch (IOException e) {
1576 e = RemoteExceptionHandler.checkIOException(e);
1577 LOG.fatal(" Got while writing log entry to log", e);
1578 throw e;
1579 }
1580 }
1581
1582
1583
1584
1585 @Override
1586 public Map<byte[], Long> getOutputCounts() {
1587 TreeMap<byte[], Long> ret = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
1588 synchronized (writers) {
1589 for (Map.Entry<byte[], SinkWriter> entry : writers.entrySet()) {
1590 ret.put(entry.getKey(), entry.getValue().editsWritten);
1591 }
1592 }
1593 return ret;
1594 }
1595
1596 @Override
1597 public int getNumberOfRecoveredRegions() {
1598 return writers.size();
1599 }
1600 }
1601
1602
1603
1604
1605 public abstract static class SinkWriter {
1606
1607 long editsWritten = 0;
1608
1609 long editsSkipped = 0;
1610
1611 long nanosSpent = 0;
1612
1613 void incrementEdits(int edits) {
1614 editsWritten += edits;
1615 }
1616
1617 void incrementSkippedEdits(int skipped) {
1618 editsSkipped += skipped;
1619 }
1620
1621 void incrementNanoTime(long nanos) {
1622 nanosSpent += nanos;
1623 }
1624 }
1625
1626
1627
1628
1629
1630 private final static class WriterAndPath extends SinkWriter {
1631 final Path p;
1632 final Writer w;
1633
1634 WriterAndPath(final Path p, final Writer w) {
1635 this.p = p;
1636 this.w = w;
1637 }
1638 }
1639
1640
1641
1642
1643 class LogReplayOutputSink extends OutputSink {
1644 private static final double BUFFER_THRESHOLD = 0.35;
1645 private static final String KEY_DELIMITER = "#";
1646
1647 private long waitRegionOnlineTimeOut;
1648 private final Set<String> recoveredRegions = Collections.synchronizedSet(new HashSet<String>());
1649 private final Map<String, RegionServerWriter> writers =
1650 new ConcurrentHashMap<String, RegionServerWriter>();
1651
1652 private final Map<String, HRegionLocation> onlineRegions =
1653 new ConcurrentHashMap<String, HRegionLocation>();
1654
1655 private Map<TableName, HConnection> tableNameToHConnectionMap = Collections
1656 .synchronizedMap(new TreeMap<TableName, HConnection>());
1657
1658
1659
1660
1661 private Map<String, List<Pair<HRegionLocation, Entry>>> serverToBufferQueueMap =
1662 new ConcurrentHashMap<String, List<Pair<HRegionLocation, Entry>>>();
1663 private List<Throwable> thrown = new ArrayList<Throwable>();
1664
1665
1666
1667
1668
1669 private LogRecoveredEditsOutputSink logRecoveredEditsOutputSink;
1670 private boolean hasEditsInDisablingOrDisabledTables = false;
1671
1672 public LogReplayOutputSink(PipelineController controller, EntryBuffers entryBuffers,
1673 int numWriters) {
1674 super(controller, entryBuffers, numWriters);
1675 this.waitRegionOnlineTimeOut = conf.getInt(HConstants.HBASE_SPLITLOG_MANAGER_TIMEOUT,
1676 ZKSplitLogManagerCoordination.DEFAULT_TIMEOUT);
1677 this.logRecoveredEditsOutputSink = new LogRecoveredEditsOutputSink(controller,
1678 entryBuffers, numWriters);
1679 this.logRecoveredEditsOutputSink.setReporter(reporter);
1680 }
1681
1682 @Override
1683 public void append(RegionEntryBuffer buffer) throws IOException {
1684 List<Entry> entries = buffer.entryBuffer;
1685 if (entries.isEmpty()) {
1686 LOG.warn("got an empty buffer, skipping");
1687 return;
1688 }
1689
1690
1691 if (disablingOrDisabledTables.contains(buffer.tableName)) {
1692
1693 logRecoveredEditsOutputSink.append(buffer);
1694 hasEditsInDisablingOrDisabledTables = true;
1695
1696 addToRecoveredRegions(Bytes.toString(buffer.encodedRegionName));
1697 return;
1698 }
1699
1700
1701 groupEditsByServer(entries);
1702
1703
1704 String maxLocKey = null;
1705 int maxSize = 0;
1706 List<Pair<HRegionLocation, Entry>> maxQueue = null;
1707 synchronized (this.serverToBufferQueueMap) {
1708 for (String key : this.serverToBufferQueueMap.keySet()) {
1709 List<Pair<HRegionLocation, Entry>> curQueue = this.serverToBufferQueueMap.get(key);
1710 if (curQueue.size() > maxSize) {
1711 maxSize = curQueue.size();
1712 maxQueue = curQueue;
1713 maxLocKey = key;
1714 }
1715 }
1716 if (maxSize < minBatchSize
1717 && entryBuffers.totalBuffered < BUFFER_THRESHOLD * entryBuffers.maxHeapUsage) {
1718
1719 return;
1720 } else if (maxSize > 0) {
1721 this.serverToBufferQueueMap.remove(maxLocKey);
1722 }
1723 }
1724
1725 if (maxSize > 0) {
1726 processWorkItems(maxLocKey, maxQueue);
1727 }
1728 }
1729
1730 private void addToRecoveredRegions(String encodedRegionName) {
1731 if (!recoveredRegions.contains(encodedRegionName)) {
1732 recoveredRegions.add(encodedRegionName);
1733 }
1734 }
1735
1736
1737
1738
1739
1740 private void groupEditsByServer(List<Entry> entries) throws IOException {
1741 Set<TableName> nonExistentTables = null;
1742 Long cachedLastFlushedSequenceId = -1l;
1743 for (Entry entry : entries) {
1744 WALEdit edit = entry.getEdit();
1745 TableName table = entry.getKey().getTablename();
1746
1747 entry.getKey().setScopes(null);
1748 String encodeRegionNameStr = Bytes.toString(entry.getKey().getEncodedRegionName());
1749
1750 if (nonExistentTables != null && nonExistentTables.contains(table)) {
1751 this.skippedEdits.incrementAndGet();
1752 continue;
1753 }
1754
1755 Map<byte[], Long> maxStoreSequenceIds = null;
1756 boolean needSkip = false;
1757 HRegionLocation loc = null;
1758 String locKey = null;
1759 List<Cell> cells = edit.getCells();
1760 List<Cell> skippedCells = new ArrayList<Cell>();
1761 HConnection hconn = this.getConnectionByTableName(table);
1762
1763 for (Cell cell : cells) {
1764 byte[] row = CellUtil.cloneRow(cell);
1765 byte[] family = CellUtil.cloneFamily(cell);
1766 boolean isCompactionEntry = false;
1767 if (CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
1768 CompactionDescriptor compaction = WALEdit.getCompaction(cell);
1769 if (compaction != null && compaction.hasRegionName()) {
1770 try {
1771 byte[][] regionName = HRegionInfo.parseRegionName(compaction.getRegionName()
1772 .toByteArray());
1773 row = regionName[1];
1774 family = compaction.getFamilyName().toByteArray();
1775 isCompactionEntry = true;
1776 } catch (Exception ex) {
1777 LOG.warn("Unexpected exception received, ignoring " + ex);
1778 skippedCells.add(cell);
1779 continue;
1780 }
1781 } else {
1782 skippedCells.add(cell);
1783 continue;
1784 }
1785 }
1786
1787 try {
1788 loc =
1789 locateRegionAndRefreshLastFlushedSequenceId(hconn, table, row,
1790 encodeRegionNameStr);
1791
1792 if (isCompactionEntry && !encodeRegionNameStr.equalsIgnoreCase(
1793 loc.getRegionInfo().getEncodedName())) {
1794 LOG.info("Not replaying a compaction marker for an older region: "
1795 + encodeRegionNameStr);
1796 needSkip = true;
1797 }
1798 } catch (TableNotFoundException ex) {
1799
1800 LOG.info("Table " + table + " doesn't exist. Skip log replay for region "
1801 + encodeRegionNameStr);
1802 lastFlushedSequenceIds.put(encodeRegionNameStr, Long.MAX_VALUE);
1803 if (nonExistentTables == null) {
1804 nonExistentTables = new TreeSet<TableName>();
1805 }
1806 nonExistentTables.add(table);
1807 this.skippedEdits.incrementAndGet();
1808 needSkip = true;
1809 break;
1810 }
1811
1812 cachedLastFlushedSequenceId =
1813 lastFlushedSequenceIds.get(loc.getRegionInfo().getEncodedName());
1814 if (cachedLastFlushedSequenceId != null
1815 && cachedLastFlushedSequenceId >= entry.getKey().getLogSeqNum()) {
1816
1817 this.skippedEdits.incrementAndGet();
1818 needSkip = true;
1819 break;
1820 } else {
1821 if (maxStoreSequenceIds == null) {
1822 maxStoreSequenceIds =
1823 regionMaxSeqIdInStores.get(loc.getRegionInfo().getEncodedName());
1824 }
1825 if (maxStoreSequenceIds != null) {
1826 Long maxStoreSeqId = maxStoreSequenceIds.get(family);
1827 if (maxStoreSeqId == null || maxStoreSeqId >= entry.getKey().getLogSeqNum()) {
1828
1829 skippedCells.add(cell);
1830 continue;
1831 }
1832 }
1833 }
1834 }
1835
1836
1837 if (loc == null || needSkip) continue;
1838
1839 if (!skippedCells.isEmpty()) {
1840 cells.removeAll(skippedCells);
1841 }
1842
1843 synchronized (serverToBufferQueueMap) {
1844 locKey = loc.getHostnamePort() + KEY_DELIMITER + table;
1845 List<Pair<HRegionLocation, Entry>> queue = serverToBufferQueueMap.get(locKey);
1846 if (queue == null) {
1847 queue =
1848 Collections.synchronizedList(new ArrayList<Pair<HRegionLocation, Entry>>());
1849 serverToBufferQueueMap.put(locKey, queue);
1850 }
1851 queue.add(new Pair<HRegionLocation, Entry>(loc, entry));
1852 }
1853
1854 addToRecoveredRegions(loc.getRegionInfo().getEncodedName());
1855 }
1856 }
1857
1858
1859
1860
1861
1862
1863 private HRegionLocation locateRegionAndRefreshLastFlushedSequenceId(HConnection hconn,
1864 TableName table, byte[] row, String originalEncodedRegionName) throws IOException {
1865
1866 HRegionLocation loc = onlineRegions.get(originalEncodedRegionName);
1867 if(loc != null) return loc;
1868
1869 loc = hconn.getRegionLocation(table, row, true);
1870 if (loc == null) {
1871 throw new IOException("Can't locate location for row:" + Bytes.toString(row)
1872 + " of table:" + table);
1873 }
1874
1875 if (!originalEncodedRegionName.equalsIgnoreCase(loc.getRegionInfo().getEncodedName())) {
1876
1877 lastFlushedSequenceIds.put(originalEncodedRegionName, Long.MAX_VALUE);
1878 HRegionLocation tmpLoc = onlineRegions.get(loc.getRegionInfo().getEncodedName());
1879 if (tmpLoc != null) return tmpLoc;
1880 }
1881
1882 Long lastFlushedSequenceId = -1l;
1883 AtomicBoolean isRecovering = new AtomicBoolean(true);
1884 loc = waitUntilRegionOnline(loc, row, this.waitRegionOnlineTimeOut, isRecovering);
1885 if (!isRecovering.get()) {
1886
1887
1888 lastFlushedSequenceIds.put(loc.getRegionInfo().getEncodedName(), Long.MAX_VALUE);
1889 LOG.info("logReplay skip region: " + loc.getRegionInfo().getEncodedName()
1890 + " because it's not in recovering.");
1891 } else {
1892 Long cachedLastFlushedSequenceId =
1893 lastFlushedSequenceIds.get(loc.getRegionInfo().getEncodedName());
1894
1895
1896
1897 RegionStoreSequenceIds ids =
1898 csm.getSplitLogWorkerCoordination().getRegionFlushedSequenceId(failedServerName,
1899 loc.getRegionInfo().getEncodedName());
1900 if (ids != null) {
1901 lastFlushedSequenceId = ids.getLastFlushedSequenceId();
1902 Map<byte[], Long> storeIds = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
1903 List<StoreSequenceId> maxSeqIdInStores = ids.getStoreSequenceIdList();
1904 for (StoreSequenceId id : maxSeqIdInStores) {
1905 storeIds.put(id.getFamilyName().toByteArray(), id.getSequenceId());
1906 }
1907 regionMaxSeqIdInStores.put(loc.getRegionInfo().getEncodedName(), storeIds);
1908 }
1909
1910 if (cachedLastFlushedSequenceId == null
1911 || lastFlushedSequenceId > cachedLastFlushedSequenceId) {
1912 lastFlushedSequenceIds.put(loc.getRegionInfo().getEncodedName(), lastFlushedSequenceId);
1913 }
1914 }
1915
1916 onlineRegions.put(loc.getRegionInfo().getEncodedName(), loc);
1917 return loc;
1918 }
1919
1920 private void processWorkItems(String key, List<Pair<HRegionLocation, Entry>> actions)
1921 throws IOException {
1922 RegionServerWriter rsw = null;
1923
1924 long startTime = System.nanoTime();
1925 try {
1926 rsw = getRegionServerWriter(key);
1927 rsw.sink.replayEntries(actions);
1928
1929
1930 rsw.incrementEdits(actions.size());
1931 rsw.incrementNanoTime(System.nanoTime() - startTime);
1932 } catch (IOException e) {
1933 e = RemoteExceptionHandler.checkIOException(e);
1934 LOG.fatal(" Got while writing log entry to log", e);
1935 throw e;
1936 }
1937 }
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948 private HRegionLocation waitUntilRegionOnline(HRegionLocation loc, byte[] row,
1949 final long timeout, AtomicBoolean isRecovering)
1950 throws IOException {
1951 final long endTime = EnvironmentEdgeManager.currentTime() + timeout;
1952 final long pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
1953 HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
1954 boolean reloadLocation = false;
1955 TableName tableName = loc.getRegionInfo().getTable();
1956 int tries = 0;
1957 Throwable cause = null;
1958 while (endTime > EnvironmentEdgeManager.currentTime()) {
1959 try {
1960
1961 HConnection hconn = getConnectionByTableName(tableName);
1962 if(reloadLocation) {
1963 loc = hconn.getRegionLocation(tableName, row, true);
1964 }
1965 BlockingInterface remoteSvr = hconn.getAdmin(loc.getServerName());
1966 HRegionInfo region = loc.getRegionInfo();
1967 try {
1968 GetRegionInfoRequest request =
1969 RequestConverter.buildGetRegionInfoRequest(region.getRegionName());
1970 GetRegionInfoResponse response = remoteSvr.getRegionInfo(null, request);
1971 if (HRegionInfo.convert(response.getRegionInfo()) != null) {
1972 isRecovering.set((response.hasIsRecovering()) ? response.getIsRecovering() : true);
1973 return loc;
1974 }
1975 } catch (ServiceException se) {
1976 throw ProtobufUtil.getRemoteException(se);
1977 }
1978 } catch (IOException e) {
1979 cause = e.getCause();
1980 if(!(cause instanceof RegionOpeningException)) {
1981 reloadLocation = true;
1982 }
1983 }
1984 long expectedSleep = ConnectionUtils.getPauseTime(pause, tries);
1985 try {
1986 Thread.sleep(expectedSleep);
1987 } catch (InterruptedException e) {
1988 throw new IOException("Interrupted when waiting region " +
1989 loc.getRegionInfo().getEncodedName() + " online.", e);
1990 }
1991 tries++;
1992 }
1993
1994 throw new IOException("Timeout when waiting region " + loc.getRegionInfo().getEncodedName() +
1995 " online for " + timeout + " milliseconds.", cause);
1996 }
1997
1998 @Override
1999 public boolean flush() throws IOException {
2000 String curLoc = null;
2001 int curSize = 0;
2002 List<Pair<HRegionLocation, Entry>> curQueue = null;
2003 synchronized (this.serverToBufferQueueMap) {
2004 for (String locationKey : this.serverToBufferQueueMap.keySet()) {
2005 curQueue = this.serverToBufferQueueMap.get(locationKey);
2006 if (!curQueue.isEmpty()) {
2007 curSize = curQueue.size();
2008 curLoc = locationKey;
2009 break;
2010 }
2011 }
2012 if (curSize > 0) {
2013 this.serverToBufferQueueMap.remove(curLoc);
2014 }
2015 }
2016
2017 if (curSize > 0) {
2018 this.processWorkItems(curLoc, curQueue);
2019
2020 synchronized(controller.dataAvailable) {
2021 controller.dataAvailable.notifyAll();
2022 }
2023 return true;
2024 }
2025 return false;
2026 }
2027
2028 void addWriterError(Throwable t) {
2029 thrown.add(t);
2030 }
2031
2032 @Override
2033 public List<Path> finishWritingAndClose() throws IOException {
2034 try {
2035 if (!finishWriting(false)) {
2036 return null;
2037 }
2038 if (hasEditsInDisablingOrDisabledTables) {
2039 splits = logRecoveredEditsOutputSink.finishWritingAndClose();
2040 } else {
2041 splits = new ArrayList<Path>();
2042 }
2043
2044 return splits;
2045 } finally {
2046 List<IOException> thrown = closeRegionServerWriters();
2047 if (thrown != null && !thrown.isEmpty()) {
2048 throw MultipleIOException.createIOException(thrown);
2049 }
2050 }
2051 }
2052
2053 @Override
2054 int getNumOpenWriters() {
2055 return this.writers.size() + this.logRecoveredEditsOutputSink.getNumOpenWriters();
2056 }
2057
2058 private List<IOException> closeRegionServerWriters() throws IOException {
2059 List<IOException> result = null;
2060 if (!writersClosed) {
2061 result = Lists.newArrayList();
2062 try {
2063 for (WriterThread t : writerThreads) {
2064 while (t.isAlive()) {
2065 t.shouldStop = true;
2066 t.interrupt();
2067 try {
2068 t.join(10);
2069 } catch (InterruptedException e) {
2070 IOException iie = new InterruptedIOException();
2071 iie.initCause(e);
2072 throw iie;
2073 }
2074 }
2075 }
2076 } finally {
2077 synchronized (writers) {
2078 for (String locationKey : writers.keySet()) {
2079 RegionServerWriter tmpW = writers.get(locationKey);
2080 try {
2081 tmpW.close();
2082 } catch (IOException ioe) {
2083 LOG.error("Couldn't close writer for region server:" + locationKey, ioe);
2084 result.add(ioe);
2085 }
2086 }
2087 }
2088
2089
2090 synchronized (this.tableNameToHConnectionMap) {
2091 for (TableName tableName : this.tableNameToHConnectionMap.keySet()) {
2092 HConnection hconn = this.tableNameToHConnectionMap.get(tableName);
2093 try {
2094 hconn.clearRegionCache();
2095 hconn.close();
2096 } catch (IOException ioe) {
2097 result.add(ioe);
2098 }
2099 }
2100 }
2101 writersClosed = true;
2102 }
2103 }
2104 return result;
2105 }
2106
2107 @Override
2108 public Map<byte[], Long> getOutputCounts() {
2109 TreeMap<byte[], Long> ret = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
2110 synchronized (writers) {
2111 for (Map.Entry<String, RegionServerWriter> entry : writers.entrySet()) {
2112 ret.put(Bytes.toBytes(entry.getKey()), entry.getValue().editsWritten);
2113 }
2114 }
2115 return ret;
2116 }
2117
2118 @Override
2119 public int getNumberOfRecoveredRegions() {
2120 return this.recoveredRegions.size();
2121 }
2122
2123
2124
2125
2126
2127
2128 private RegionServerWriter getRegionServerWriter(String loc) throws IOException {
2129 RegionServerWriter ret = writers.get(loc);
2130 if (ret != null) {
2131 return ret;
2132 }
2133
2134 TableName tableName = getTableFromLocationStr(loc);
2135 if(tableName == null){
2136 throw new IOException("Invalid location string:" + loc + " found. Replay aborted.");
2137 }
2138
2139 HConnection hconn = getConnectionByTableName(tableName);
2140 synchronized (writers) {
2141 ret = writers.get(loc);
2142 if (ret == null) {
2143 ret = new RegionServerWriter(conf, tableName, hconn);
2144 writers.put(loc, ret);
2145 }
2146 }
2147 return ret;
2148 }
2149
2150 private HConnection getConnectionByTableName(final TableName tableName) throws IOException {
2151 HConnection hconn = this.tableNameToHConnectionMap.get(tableName);
2152 if (hconn == null) {
2153 synchronized (this.tableNameToHConnectionMap) {
2154 hconn = this.tableNameToHConnectionMap.get(tableName);
2155 if (hconn == null) {
2156 hconn = HConnectionManager.getConnection(conf);
2157 this.tableNameToHConnectionMap.put(tableName, hconn);
2158 }
2159 }
2160 }
2161 return hconn;
2162 }
2163 private TableName getTableFromLocationStr(String loc) {
2164
2165
2166
2167 String[] splits = loc.split(KEY_DELIMITER);
2168 if (splits.length != 2) {
2169 return null;
2170 }
2171 return TableName.valueOf(splits[1]);
2172 }
2173 }
2174
2175
2176
2177
2178
2179 private final static class RegionServerWriter extends SinkWriter {
2180 final WALEditsReplaySink sink;
2181
2182 RegionServerWriter(final Configuration conf, final TableName tableName, final HConnection conn)
2183 throws IOException {
2184 this.sink = new WALEditsReplaySink(conf, tableName, conn);
2185 }
2186
2187 void close() throws IOException {
2188 }
2189 }
2190
2191 static class CorruptedLogFileException extends Exception {
2192 private static final long serialVersionUID = 1L;
2193
2194 CorruptedLogFileException(String s) {
2195 super(s);
2196 }
2197 }
2198
2199
2200 public static class MutationReplay {
2201 public MutationReplay(MutationType type, Mutation mutation, long nonceGroup, long nonce) {
2202 this.type = type;
2203 this.mutation = mutation;
2204 if(this.mutation.getDurability() != Durability.SKIP_WAL) {
2205
2206 this.mutation.setDurability(Durability.ASYNC_WAL);
2207 }
2208 this.nonceGroup = nonceGroup;
2209 this.nonce = nonce;
2210 }
2211
2212 public final MutationType type;
2213 public final Mutation mutation;
2214 public final long nonceGroup;
2215 public final long nonce;
2216 }
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229 public static List<MutationReplay> getMutationsFromWALEntry(WALEntry entry, CellScanner cells,
2230 Pair<WALKey, WALEdit> logEntry, Durability durability)
2231 throws IOException {
2232 if (entry == null) {
2233
2234 return new ArrayList<MutationReplay>();
2235 }
2236
2237 long replaySeqId = (entry.getKey().hasOrigSequenceNumber()) ?
2238 entry.getKey().getOrigSequenceNumber() : entry.getKey().getLogSequenceNumber();
2239 int count = entry.getAssociatedCellCount();
2240 List<MutationReplay> mutations = new ArrayList<MutationReplay>();
2241 Cell previousCell = null;
2242 Mutation m = null;
2243 WALKey key = null;
2244 WALEdit val = null;
2245 if (logEntry != null) val = new WALEdit();
2246
2247 for (int i = 0; i < count; i++) {
2248
2249 if (!cells.advance()) {
2250 throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i);
2251 }
2252 Cell cell = cells.current();
2253 if (val != null) val.add(cell);
2254
2255 boolean isNewRowOrType =
2256 previousCell == null || previousCell.getTypeByte() != cell.getTypeByte()
2257 || !CellUtil.matchingRow(previousCell, cell);
2258 if (isNewRowOrType) {
2259
2260 if (CellUtil.isDelete(cell)) {
2261 m = new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
2262
2263 mutations.add(new MutationReplay(
2264 MutationType.DELETE, m, HConstants.NO_NONCE, HConstants.NO_NONCE));
2265 } else {
2266 m = new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
2267
2268 long nonceGroup = entry.getKey().hasNonceGroup()
2269 ? entry.getKey().getNonceGroup() : HConstants.NO_NONCE;
2270 long nonce = entry.getKey().hasNonce() ? entry.getKey().getNonce() : HConstants.NO_NONCE;
2271 mutations.add(new MutationReplay(MutationType.PUT, m, nonceGroup, nonce));
2272 }
2273 }
2274 if (CellUtil.isDelete(cell)) {
2275 ((Delete) m).addDeleteMarker(cell);
2276 } else {
2277 ((Put) m).add(cell);
2278 }
2279 m.setDurability(durability);
2280 previousCell = cell;
2281 }
2282
2283
2284 if (logEntry != null) {
2285 org.apache.hadoop.hbase.protobuf.generated.WALProtos.WALKey walKeyProto = entry.getKey();
2286 List<UUID> clusterIds = new ArrayList<UUID>(walKeyProto.getClusterIdsCount());
2287 for (HBaseProtos.UUID uuid : entry.getKey().getClusterIdsList()) {
2288 clusterIds.add(new UUID(uuid.getMostSigBits(), uuid.getLeastSigBits()));
2289 }
2290
2291 key = new HLogKey(walKeyProto.getEncodedRegionName().toByteArray(), TableName.valueOf(
2292 walKeyProto.getTableName().toByteArray()), replaySeqId, walKeyProto.getWriteTime(),
2293 clusterIds, walKeyProto.getNonceGroup(), walKeyProto.getNonce(), null);
2294 logEntry.setFirst(key);
2295 logEntry.setSecond(val);
2296 }
2297
2298 return mutations;
2299 }
2300 }