1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver.wal;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertNotNull;
23 import static org.junit.Assert.assertTrue;
24 import static org.junit.Assert.fail;
25 import static org.mockito.Matchers.any;
26 import static org.mockito.Matchers.eq;
27 import static org.mockito.Mockito.doAnswer;
28 import static org.mockito.Mockito.spy;
29 import static org.mockito.Mockito.when;
30
31 import java.io.FilterInputStream;
32 import java.io.IOException;
33 import java.lang.reflect.Field;
34 import java.security.PrivilegedExceptionAction;
35 import java.util.ArrayList;
36 import java.util.Arrays;
37 import java.util.Collection;
38 import java.util.HashSet;
39 import java.util.List;
40 import java.util.Set;
41 import java.util.concurrent.atomic.AtomicBoolean;
42 import java.util.concurrent.atomic.AtomicInteger;
43
44 import org.apache.commons.logging.Log;
45 import org.apache.commons.logging.LogFactory;
46 import org.apache.hadoop.conf.Configuration;
47 import org.apache.hadoop.fs.FSDataInputStream;
48 import org.apache.hadoop.fs.FileStatus;
49 import org.apache.hadoop.fs.FileSystem;
50 import org.apache.hadoop.fs.Path;
51 import org.apache.hadoop.fs.PathFilter;
52 import org.apache.hadoop.hbase.Cell;
53 import org.apache.hadoop.hbase.HBaseConfiguration;
54 import org.apache.hadoop.hbase.HBaseTestingUtility;
55 import org.apache.hadoop.hbase.HColumnDescriptor;
56 import org.apache.hadoop.hbase.HConstants;
57 import org.apache.hadoop.hbase.HRegionInfo;
58 import org.apache.hadoop.hbase.HTableDescriptor;
59 import org.apache.hadoop.hbase.KeyValue;
60 import org.apache.hadoop.hbase.MasterNotRunningException;
61 import org.apache.hadoop.hbase.MiniHBaseCluster;
62 import org.apache.hadoop.hbase.ServerName;
63 import org.apache.hadoop.hbase.TableName;
64 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
65 import org.apache.hadoop.hbase.client.Delete;
66 import org.apache.hadoop.hbase.client.Get;
67 import org.apache.hadoop.hbase.client.HTable;
68 import org.apache.hadoop.hbase.client.Put;
69 import org.apache.hadoop.hbase.client.Result;
70 import org.apache.hadoop.hbase.client.ResultScanner;
71 import org.apache.hadoop.hbase.client.Scan;
72 import org.apache.hadoop.hbase.client.Table;
73 import org.apache.hadoop.hbase.fs.HFileSystem;
74 import org.apache.hadoop.hbase.master.HMaster;
75 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
76 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
77 import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
78 import org.apache.hadoop.hbase.regionserver.DefaultStoreFlusher;
79 import org.apache.hadoop.hbase.regionserver.FlushRequestListener;
80 import org.apache.hadoop.hbase.regionserver.FlushRequester;
81 import org.apache.hadoop.hbase.regionserver.HRegion;
82 import org.apache.hadoop.hbase.regionserver.HRegionServer;
83 import org.apache.hadoop.hbase.regionserver.MemStoreSnapshot;
84 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
85 import org.apache.hadoop.hbase.regionserver.Region;
86 import org.apache.hadoop.hbase.regionserver.RegionScanner;
87 import org.apache.hadoop.hbase.regionserver.RegionServerServices;
88 import org.apache.hadoop.hbase.regionserver.Store;
89 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
90 import org.apache.hadoop.hbase.security.User;
91 import org.apache.hadoop.hbase.testclassification.MediumTests;
92 import org.apache.hadoop.hbase.util.Bytes;
93 import org.apache.hadoop.hbase.util.EnvironmentEdge;
94 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
95 import org.apache.hadoop.hbase.util.FSUtils;
96 import org.apache.hadoop.hbase.util.HFileTestUtil;
97 import org.apache.hadoop.hbase.util.Pair;
98 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
99 import org.apache.hadoop.hbase.wal.WAL;
100 import org.apache.hadoop.hbase.wal.WALFactory;
101 import org.apache.hadoop.hbase.wal.WALKey;
102 import org.apache.hadoop.hbase.wal.WALSplitter;
103 import org.apache.hadoop.hdfs.DFSInputStream;
104 import org.junit.After;
105 import org.junit.AfterClass;
106 import org.junit.Before;
107 import org.junit.BeforeClass;
108 import org.junit.Rule;
109 import org.junit.Test;
110 import org.junit.experimental.categories.Category;
111 import org.junit.rules.TestName;
112 import org.mockito.Mockito;
113 import org.mockito.invocation.InvocationOnMock;
114 import org.mockito.stubbing.Answer;
115
116
117
118
119 @Category(MediumTests.class)
120 public class TestWALReplay {
121 private static final Log LOG = LogFactory.getLog(TestWALReplay.class);
122 static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
123 private final EnvironmentEdge ee = EnvironmentEdgeManager.getDelegate();
124 private Path hbaseRootDir = null;
125 private Path hbaseWALRootDir = null;
126 private String logName;
127 private Path oldLogDir;
128 private Path logDir;
129 private FileSystem fs;
130 private Configuration conf;
131 private RecoveryMode mode;
132 private WALFactory wals;
133
134 @Rule
135 public final TestName currentTest = new TestName();
136
137
138 @BeforeClass
139 public static void setUpBeforeClass() throws Exception {
140 Configuration conf = TEST_UTIL.getConfiguration();
141 conf.setBoolean("dfs.support.append", true);
142
143 conf.setInt("dfs.client.block.recovery.retries", 2);
144 TEST_UTIL.startMiniCluster(3);
145 Path hbaseRootDir =
146 TEST_UTIL.getDFSCluster().getFileSystem().makeQualified(new Path("/hbase"));
147 Path hbaseWALRootDir =
148 TEST_UTIL.getDFSCluster().getFileSystem().makeQualified(new Path("/hbaselog"));
149 LOG.info(HConstants.HBASE_DIR + "=" + hbaseRootDir);
150 LOG.info(HFileSystem.HBASE_WAL_DIR + "=" + hbaseWALRootDir);
151 FSUtils.setRootDir(conf, hbaseRootDir);
152 FSUtils.setWALRootDir(conf, hbaseWALRootDir);
153 }
154
155 @AfterClass
156 public static void tearDownAfterClass() throws Exception {
157 TEST_UTIL.shutdownMiniCluster();
158 }
159
160 @Before
161 public void setUp() throws Exception {
162 this.conf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
163 this.fs = TEST_UTIL.getDFSCluster().getFileSystem();
164 this.hbaseRootDir = FSUtils.getRootDir(this.conf);
165 this.hbaseWALRootDir = FSUtils.getWALRootDir(this.conf);
166 this.oldLogDir = new Path(this.hbaseWALRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
167 this.logName = DefaultWALProvider.getWALDirectoryName(currentTest.getMethodName() + "-manual");
168 this.logDir = new Path(this.hbaseWALRootDir, logName);
169 if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseRootDir)) {
170 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
171 }
172 if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseWALRootDir)) {
173 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseWALRootDir, true);
174 }
175 this.mode = (conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false) ?
176 RecoveryMode.LOG_REPLAY : RecoveryMode.LOG_SPLITTING);
177 this.wals = new WALFactory(conf, null, currentTest.getMethodName());
178 }
179
180 @After
181 public void tearDown() throws Exception {
182 this.wals.close();
183 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
184 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseWALRootDir, true);
185 }
186
187
188
189
190 private void deleteDir(final Path p) throws IOException {
191 if (this.fs.exists(p)) {
192 if (!this.fs.delete(p, true)) {
193 throw new IOException("Failed remove of " + p);
194 }
195 }
196 }
197
198
199
200
201
202 @Test
203 public void testReplayEditsAfterRegionMovedWithMultiCF() throws Exception {
204 final TableName tableName =
205 TableName.valueOf("testReplayEditsAfterRegionMovedWithMultiCF");
206 byte[] family1 = Bytes.toBytes("cf1");
207 byte[] family2 = Bytes.toBytes("cf2");
208 byte[] qualifier = Bytes.toBytes("q");
209 byte[] value = Bytes.toBytes("testV");
210 byte[][] familys = { family1, family2 };
211 TEST_UTIL.createTable(tableName, familys);
212 Table htable = new HTable(TEST_UTIL.getConfiguration(), tableName);
213 Put put = new Put(Bytes.toBytes("r1"));
214 put.add(family1, qualifier, value);
215 htable.put(put);
216 ResultScanner resultScanner = htable.getScanner(new Scan());
217 int count = 0;
218 while (resultScanner.next() != null) {
219 count++;
220 }
221 resultScanner.close();
222 assertEquals(1, count);
223
224 MiniHBaseCluster hbaseCluster = TEST_UTIL.getMiniHBaseCluster();
225 List<HRegion> regions = hbaseCluster.getRegions(tableName);
226 assertEquals(1, regions.size());
227
228
229 Region destRegion = regions.get(0);
230 int originServerNum = hbaseCluster
231 .getServerWith(destRegion.getRegionInfo().getRegionName());
232 assertTrue("Please start more than 1 regionserver", hbaseCluster
233 .getRegionServerThreads().size() > 1);
234 int destServerNum = 0;
235 while (destServerNum == originServerNum) {
236 destServerNum++;
237 }
238 HRegionServer originServer = hbaseCluster.getRegionServer(originServerNum);
239 HRegionServer destServer = hbaseCluster.getRegionServer(destServerNum);
240
241 moveRegionAndWait(destRegion, destServer);
242
243
244 Delete del = new Delete(Bytes.toBytes("r1"));
245 htable.delete(del);
246 resultScanner = htable.getScanner(new Scan());
247 count = 0;
248 while (resultScanner.next() != null) {
249 count++;
250 }
251 resultScanner.close();
252 assertEquals(0, count);
253
254
255 Region region = destServer.getOnlineRegion(destRegion.getRegionInfo().getRegionName());
256 region.flush(true);
257
258 for (Store store : region.getStores()) {
259 store.triggerMajorCompaction();
260 }
261 region.compact(true);
262
263
264 moveRegionAndWait(destRegion, originServer);
265
266 originServer.abort("testing");
267
268
269 Result result = htable.get(new Get(Bytes.toBytes("r1")));
270 if (result != null) {
271 assertTrue("Row is deleted, but we get" + result.toString(),
272 (result == null) || result.isEmpty());
273 }
274 resultScanner.close();
275 }
276
277 private void moveRegionAndWait(Region destRegion, HRegionServer destServer)
278 throws InterruptedException, MasterNotRunningException,
279 ZooKeeperConnectionException, IOException {
280 HMaster master = TEST_UTIL.getMiniHBaseCluster().getMaster();
281 TEST_UTIL.getHBaseAdmin().move(
282 destRegion.getRegionInfo().getEncodedNameAsBytes(),
283 Bytes.toBytes(destServer.getServerName().getServerName()));
284 while (true) {
285 ServerName serverName = master.getAssignmentManager()
286 .getRegionStates().getRegionServerOfRegion(destRegion.getRegionInfo());
287 if (serverName != null && serverName.equals(destServer.getServerName())) {
288 TEST_UTIL.assertRegionOnServer(
289 destRegion.getRegionInfo(), serverName, 200);
290 break;
291 }
292 Thread.sleep(10);
293 }
294 }
295
296
297
298
299
300
301 @Test
302 public void test2727() throws Exception {
303
304
305 final TableName tableName =
306 TableName.valueOf("test2727");
307
308 MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
309 HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
310 Path basedir = FSUtils.getWALTableDir(conf, tableName);
311 deleteDir(basedir);
312
313 HTableDescriptor htd = createBasic3FamilyHTD(tableName);
314 HRegion region2 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
315 HRegion.closeHRegion(region2);
316 final byte [] rowName = tableName.getName();
317
318 WAL wal1 = createWAL(this.conf);
319
320 final int countPerFamily = 1000;
321
322 for (HColumnDescriptor hcd: htd.getFamilies()) {
323 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily, ee,
324 wal1, htd, mvcc);
325 }
326 wal1.shutdown();
327 runWALSplit(this.conf);
328
329 WAL wal2 = createWAL(this.conf);
330
331 for (HColumnDescriptor hcd: htd.getFamilies()) {
332 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily,
333 ee, wal2, htd, mvcc);
334 }
335 wal2.shutdown();
336 runWALSplit(this.conf);
337
338 WAL wal3 = createWAL(this.conf);
339 try {
340 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal3);
341 long seqid = region.getOpenSeqNum();
342
343
344 assertTrue(seqid > mvcc.getWritePoint());
345 assertEquals(seqid - 1, mvcc.getWritePoint());
346 LOG.debug("region.getOpenSeqNum(): " + region.getOpenSeqNum() + ", wal3.id: "
347 + mvcc.getReadPoint());
348
349
350 region.close();
351 } finally {
352 wal3.close();
353 }
354 }
355
356
357
358
359
360
361
362
363
364
365 @Test
366 public void testRegionMadeOfBulkLoadedFilesOnly()
367 throws IOException, SecurityException, IllegalArgumentException,
368 NoSuchFieldException, IllegalAccessException, InterruptedException {
369 final TableName tableName =
370 TableName.valueOf("testRegionMadeOfBulkLoadedFilesOnly");
371 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
372 final Path basedir = new Path(this.hbaseRootDir, tableName.getNameAsString());
373 deleteDir(basedir);
374 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
375 HRegion region2 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
376 HRegion.closeHRegion(region2);
377 WAL wal = createWAL(this.conf);
378 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
379
380 byte [] family = htd.getFamilies().iterator().next().getName();
381 Path f = new Path(basedir, "hfile");
382 HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(""),
383 Bytes.toBytes("z"), 10);
384 List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
385 hfs.add(Pair.newPair(family, f.toString()));
386 region.bulkLoadHFiles(hfs, true, null);
387
388
389 byte [] row = tableName.getName();
390 region.put((new Put(row)).add(family, family, family));
391 wal.sync();
392 final int rowsInsertedCount = 11;
393
394 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
395
396
397 final Configuration newConf = HBaseConfiguration.create(this.conf);
398 User user = HBaseTestingUtility.getDifferentUser(newConf,
399 tableName.getNameAsString());
400 user.runAs(new PrivilegedExceptionAction() {
401 @Override
402 public Object run() throws Exception {
403 runWALSplit(newConf);
404 WAL wal2 = createWAL(newConf);
405
406 HRegion region2 = HRegion.openHRegion(newConf, FileSystem.get(newConf),
407 hbaseRootDir, hri, htd, wal2);
408 long seqid2 = region2.getOpenSeqNum();
409 assertTrue(seqid2 > -1);
410 assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
411
412
413 region2.close();
414 wal2.close();
415 return null;
416 }
417 });
418 }
419
420
421
422
423
424
425
426
427
428
429
430
431 @Test
432 public void testCompactedBulkLoadedFiles()
433 throws IOException, SecurityException, IllegalArgumentException,
434 NoSuchFieldException, IllegalAccessException, InterruptedException {
435 final TableName tableName =
436 TableName.valueOf("testCompactedBulkLoadedFiles");
437 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
438 final Path basedir = new Path(this.hbaseRootDir, tableName.getNameAsString());
439 deleteDir(basedir);
440 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
441 HRegion region2 = HRegion.createHRegion(hri,
442 hbaseRootDir, this.conf, htd);
443 HRegion.closeHRegion(region2);
444 WAL wal = createWAL(this.conf);
445 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
446
447
448 byte [] row = tableName.getName();
449 byte [] family = htd.getFamilies().iterator().next().getName();
450 region.put((new Put(row)).add(family, family, family));
451 wal.sync();
452
453 List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
454 for (int i = 0; i < 3; i++) {
455 Path f = new Path(basedir, "hfile"+i);
456 HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(i + "00"),
457 Bytes.toBytes(i + "50"), 10);
458 hfs.add(Pair.newPair(family, f.toString()));
459 }
460 region.bulkLoadHFiles(hfs, true, null);
461 final int rowsInsertedCount = 31;
462 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
463
464
465 region.compact(true);
466 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
467
468
469 final Configuration newConf = HBaseConfiguration.create(this.conf);
470 User user = HBaseTestingUtility.getDifferentUser(newConf,
471 tableName.getNameAsString());
472 user.runAs(new PrivilegedExceptionAction() {
473 @Override
474 public Object run() throws Exception {
475 runWALSplit(newConf);
476 WAL wal2 = createWAL(newConf);
477
478 HRegion region2 = HRegion.openHRegion(newConf, FileSystem.get(newConf),
479 hbaseRootDir, hri, htd, wal2);
480 long seqid2 = region2.getOpenSeqNum();
481 assertTrue(seqid2 > -1);
482 assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
483
484
485 region2.close();
486 wal2.close();
487 return null;
488 }
489 });
490 }
491
492
493
494
495
496
497
498
499
500
501
502 @Test
503 public void testReplayEditsWrittenViaHRegion()
504 throws IOException, SecurityException, IllegalArgumentException,
505 NoSuchFieldException, IllegalAccessException, InterruptedException {
506 final TableName tableName =
507 TableName.valueOf("testReplayEditsWrittenViaHRegion");
508 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
509 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
510 deleteDir(basedir);
511 final byte[] rowName = tableName.getName();
512 final int countPerFamily = 10;
513 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
514 HRegion region3 = HRegion.createHRegion(hri,
515 hbaseRootDir, this.conf, htd);
516 HRegion.closeHRegion(region3);
517
518
519
520 WAL wal = createWAL(this.conf);
521 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
522 long seqid = region.getOpenSeqNum();
523 boolean first = true;
524 for (HColumnDescriptor hcd: htd.getFamilies()) {
525 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
526 if (first) {
527
528 region.flush(true);
529 first = false;
530 }
531 }
532
533 final Get g = new Get(rowName);
534 Result result = region.get(g);
535 assertEquals(countPerFamily * htd.getFamilies().size(),
536 result.size());
537
538
539
540 region.close(true);
541 wal.shutdown();
542 runWALSplit(this.conf);
543 WAL wal2 = createWAL(this.conf);
544 HRegion region2 = HRegion.openHRegion(conf, this.fs, hbaseRootDir, hri, htd, wal2);
545 long seqid2 = region2.getOpenSeqNum();
546 assertTrue(seqid + result.size() < seqid2);
547 final Result result1b = region2.get(g);
548 assertEquals(result.size(), result1b.size());
549
550
551
552
553 for (HColumnDescriptor hcd: htd.getFamilies()) {
554 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region2, "y");
555 }
556
557 final Result result2 = region2.get(g);
558 assertEquals(2 * result.size(), result2.size());
559 wal2.sync();
560 final Configuration newConf = HBaseConfiguration.create(this.conf);
561 User user = HBaseTestingUtility.getDifferentUser(newConf,
562 tableName.getNameAsString());
563 user.runAs(new PrivilegedExceptionAction() {
564 @Override
565 public Object run() throws Exception {
566 runWALSplit(newConf);
567 FileSystem newFS = FileSystem.get(newConf);
568
569 WAL wal3 = createWAL(newConf);
570 final AtomicInteger countOfRestoredEdits = new AtomicInteger(0);
571 HRegion region3 = new HRegion(basedir, wal3, newFS, newConf, hri, htd, null) {
572 @Override
573 protected boolean restoreEdit(Store s, Cell cell) {
574 boolean b = super.restoreEdit(s, cell);
575 countOfRestoredEdits.incrementAndGet();
576 return b;
577 }
578 };
579 long seqid3 = region3.initialize();
580 Result result3 = region3.get(g);
581
582 assertEquals(result2.size(), result3.size());
583 assertEquals(htd.getFamilies().size() * countPerFamily,
584 countOfRestoredEdits.get());
585
586
587 region3.close();
588 wal3.close();
589 return null;
590 }
591 });
592 }
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612 @Test
613 public void testReplayEditsAfterPartialFlush()
614 throws IOException, SecurityException, IllegalArgumentException,
615 NoSuchFieldException, IllegalAccessException, InterruptedException {
616 final TableName tableName =
617 TableName.valueOf("testReplayEditsWrittenViaHRegion");
618 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
619 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
620 deleteDir(basedir);
621 final byte[] rowName = tableName.getName();
622 final int countPerFamily = 10;
623 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
624 HRegion region3 = HRegion.createHRegion(hri,
625 hbaseRootDir, this.conf, htd);
626 HRegion.closeHRegion(region3);
627
628
629
630 WAL wal = createWAL(this.conf);
631 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
632 long seqid = region.getOpenSeqNum();
633 for (HColumnDescriptor hcd: htd.getFamilies()) {
634 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
635 }
636
637
638 final Get g = new Get(rowName);
639 Result result = region.get(g);
640 assertEquals(countPerFamily * htd.getFamilies().size(),
641 result.size());
642
643
644 region.flush(true);
645 region.close(true);
646 wal.shutdown();
647
648
649
650
651
652 int cf_count = 0;
653 for (HColumnDescriptor hcd: htd.getFamilies()) {
654 cf_count++;
655 if (cf_count == 2) {
656 region.getRegionFileSystem().deleteFamily(hcd.getNameAsString());
657 }
658 }
659
660
661
662 runWALSplit(this.conf);
663 WAL wal2 = createWAL(this.conf);
664 HRegion region2 = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal2);
665 long seqid2 = region2.getOpenSeqNum();
666 assertTrue(seqid + result.size() < seqid2);
667
668 final Result result1b = region2.get(g);
669 assertEquals(result.size(), result1b.size());
670 }
671
672
673
674
675 public static class CustomStoreFlusher extends DefaultStoreFlusher {
676
677 static final AtomicBoolean throwExceptionWhenFlushing = new AtomicBoolean(false);
678
679 public CustomStoreFlusher(Configuration conf, Store store) {
680 super(conf, store);
681 }
682 @Override
683 public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
684 MonitoredTask status, ThroughputController throughputController) throws IOException {
685 if (throwExceptionWhenFlushing.get()) {
686 throw new IOException("Simulated exception by tests");
687 }
688 return super.flushSnapshot(snapshot, cacheFlushId, status, throughputController);
689 }
690
691 };
692
693
694
695
696
697
698
699 @Test
700 public void testReplayEditsAfterAbortingFlush() throws IOException {
701 final TableName tableName =
702 TableName.valueOf("testReplayEditsAfterAbortingFlush");
703 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
704 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
705 deleteDir(basedir);
706 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
707 HRegion region3 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
708 region3.close();
709 region3.getWAL().close();
710
711
712
713 WAL wal = createWAL(this.conf);
714 RegionServerServices rsServices = Mockito.mock(RegionServerServices.class);
715 Mockito.doReturn(false).when(rsServices).isAborted();
716 when(rsServices.getServerName()).thenReturn(ServerName.valueOf("foo", 10, 10));
717 Configuration customConf = new Configuration(this.conf);
718 customConf.set(DefaultStoreEngine.DEFAULT_STORE_FLUSHER_CLASS_KEY,
719 CustomStoreFlusher.class.getName());
720 HRegion region =
721 HRegion.openHRegion(this.hbaseRootDir, hri, htd, wal, customConf, rsServices, null);
722 int writtenRowCount = 10;
723 List<HColumnDescriptor> families = new ArrayList<HColumnDescriptor>(
724 htd.getFamilies());
725 for (int i = 0; i < writtenRowCount; i++) {
726 Put put = new Put(Bytes.toBytes(tableName + Integer.toString(i)));
727 put.add(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
728 Bytes.toBytes("val"));
729 region.put(put);
730 }
731
732
733 RegionScanner scanner = region.getScanner(new Scan());
734 assertEquals(writtenRowCount, getScannedCount(scanner));
735
736
737 CustomStoreFlusher.throwExceptionWhenFlushing.set(true);
738 try {
739 region.flush(true);
740 fail("Injected exception hasn't been thrown");
741 } catch (Throwable t) {
742 LOG.info("Expected simulated exception when flushing region,"
743 + t.getMessage());
744
745 Mockito.doReturn(true).when(rsServices).isAborted();
746 region.setClosing(false);
747
748 }
749
750 int moreRow = 10;
751 for (int i = writtenRowCount; i < writtenRowCount + moreRow; i++) {
752 Put put = new Put(Bytes.toBytes(tableName + Integer.toString(i)));
753 put.add(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
754 Bytes.toBytes("val"));
755 region.put(put);
756 }
757 writtenRowCount += moreRow;
758
759 CustomStoreFlusher.throwExceptionWhenFlushing.set(false);
760 try {
761 region.flush(true);
762 } catch (IOException t) {
763 LOG.info("Expected exception when flushing region because server is stopped,"
764 + t.getMessage());
765 }
766
767 region.close(true);
768 wal.shutdown();
769
770
771 runWALSplit(this.conf);
772 WAL wal2 = createWAL(this.conf);
773 Mockito.doReturn(false).when(rsServices).isAborted();
774 HRegion region2 =
775 HRegion.openHRegion(this.hbaseRootDir, hri, htd, wal2, this.conf, rsServices, null);
776 scanner = region2.getScanner(new Scan());
777 assertEquals(writtenRowCount, getScannedCount(scanner));
778 }
779
780 private int getScannedCount(RegionScanner scanner) throws IOException {
781 int scannedCount = 0;
782 List<Cell> results = new ArrayList<Cell>();
783 while (true) {
784 boolean existMore = scanner.next(results);
785 if (!results.isEmpty())
786 scannedCount++;
787 if (!existMore)
788 break;
789 results.clear();
790 }
791 return scannedCount;
792 }
793
794
795
796
797
798
799 @Test
800 public void testReplayEditsWrittenIntoWAL() throws Exception {
801 final TableName tableName =
802 TableName.valueOf("testReplayEditsWrittenIntoWAL");
803 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
804 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
805 final Path basedir = FSUtils.getTableDir(hbaseRootDir, tableName);
806 deleteDir(basedir);
807
808 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
809 HRegion region2 = HRegion.createHRegion(hri,
810 hbaseRootDir, this.conf, htd);
811 HRegion.closeHRegion(region2);
812 final WAL wal = createWAL(this.conf);
813 final byte[] rowName = tableName.getName();
814 final byte[] regionName = hri.getEncodedNameAsBytes();
815
816
817 final int countPerFamily = 1000;
818 Set<byte[]> familyNames = new HashSet<byte[]>();
819 for (HColumnDescriptor hcd: htd.getFamilies()) {
820 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily,
821 ee, wal, htd, mvcc);
822 familyNames.add(hcd.getName());
823 }
824
825
826 wal.startCacheFlush(regionName, familyNames);
827 wal.completeCacheFlush(regionName);
828
829
830 WALEdit edit = new WALEdit();
831 long now = ee.currentTime();
832 edit.add(new KeyValue(rowName, Bytes.toBytes("another family"), rowName,
833 now, rowName));
834 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName, now, mvcc), edit, true);
835
836
837 edit = new WALEdit();
838 now = ee.currentTime();
839 edit.add(new KeyValue(rowName, Bytes.toBytes("c"), null, now, KeyValue.Type.DeleteFamily));
840 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName, now, mvcc), edit, true);
841
842
843 wal.sync();
844
845
846 final Configuration newConf = HBaseConfiguration.create(this.conf);
847 User user = HBaseTestingUtility.getDifferentUser(newConf,
848 ".replay.wal.secondtime");
849 user.runAs(new PrivilegedExceptionAction<Void>() {
850 @Override
851 public Void run() throws Exception {
852 runWALSplit(newConf);
853 FileSystem newFS = FileSystem.get(newConf);
854
855 newConf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 1024 * 100);
856
857 WAL newWal = createWAL(newConf);
858 final AtomicInteger flushcount = new AtomicInteger(0);
859 try {
860 final HRegion region =
861 new HRegion(basedir, newWal, newFS, newConf, hri, htd, null) {
862 @Override
863 protected FlushResult internalFlushcache(final WAL wal, final long myseqid,
864 final Collection<Store> storesToFlush, MonitoredTask status,
865 boolean writeFlushWalMarker)
866 throws IOException {
867 LOG.info("InternalFlushCache Invoked");
868 FlushResult fs = super.internalFlushcache(wal, myseqid, storesToFlush,
869 Mockito.mock(MonitoredTask.class), writeFlushWalMarker);
870 flushcount.incrementAndGet();
871 return fs;
872 }
873 };
874
875 long seqid = region.initialize();
876
877
878 long writePoint = mvcc.getWritePoint();
879
880
881 assertTrue("Flushcount=" + flushcount.get(), flushcount.get() > 0);
882 assertTrue((seqid - 1) == writePoint);
883
884 Get get = new Get(rowName);
885 Result result = region.get(get);
886
887 assertEquals(countPerFamily * (htd.getFamilies().size() - 1),
888 result.size());
889 region.close();
890 } finally {
891 newWal.close();
892 }
893 return null;
894 }
895 });
896 }
897
898 @Test
899
900 public void testSequentialEditLogSeqNum() throws IOException {
901 final TableName tableName = TableName.valueOf(currentTest.getMethodName());
902 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
903 final Path basedir =
904 FSUtils.getTableDir(this.hbaseRootDir, tableName);
905 deleteDir(basedir);
906 final byte[] rowName = tableName.getName();
907 final int countPerFamily = 10;
908 final HTableDescriptor htd = createBasic1FamilyHTD(tableName);
909
910
911 MockWAL wal = createMockWAL();
912
913 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
914 for (HColumnDescriptor hcd : htd.getFamilies()) {
915 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
916 }
917
918
919
920 region.flush(true);
921 for (HColumnDescriptor hcd : htd.getFamilies()) {
922 addRegionEdits(rowName, hcd.getName(), 5, this.ee, region, "x");
923 }
924 long lastestSeqNumber = region.getSequenceId();
925
926 wal.doCompleteCacheFlush = true;
927
928
929 wal.completeCacheFlush(hri.getEncodedNameAsBytes());
930 wal.shutdown();
931 FileStatus[] listStatus = wal.getFiles();
932 assertNotNull(listStatus);
933 assertTrue(listStatus.length > 0);
934 WALSplitter.splitLogFile(hbaseWALRootDir, listStatus[0],
935 this.fs, this.conf, null, null, null, mode, wals);
936 FileStatus[] listStatus1 = this.fs.listStatus(
937 new Path(FSUtils.getWALTableDir(this.conf, tableName), new Path(hri.getEncodedName(),
938 "recovered.edits")), new PathFilter() {
939 @Override
940 public boolean accept(Path p) {
941 if (WALSplitter.isSequenceIdFile(p)) {
942 return false;
943 }
944 return true;
945 }
946 });
947 int editCount = 0;
948 for (FileStatus fileStatus : listStatus1) {
949 editCount = Integer.parseInt(fileStatus.getPath().getName());
950 }
951
952 assertEquals(
953 "The sequence number of the recoverd.edits and the current edit seq should be same",
954 lastestSeqNumber, editCount);
955 }
956
957
958
959
960 @Test
961 public void testDatalossWhenInputError() throws IOException, InstantiationException,
962 IllegalAccessException {
963 final TableName tableName = TableName.valueOf("testDatalossWhenInputError");
964 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
965 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
966 deleteDir(basedir);
967 final byte[] rowName = tableName.getName();
968 final int countPerFamily = 10;
969 final HTableDescriptor htd = createBasic1FamilyHTD(tableName);
970 HRegion region1 = HBaseTestingUtility.createRegionAndWAL(hri, hbaseRootDir, this.hbaseWALRootDir, this.conf, htd);
971 Path regionDir = region1.getWALRegionDir();
972 HBaseTestingUtility.closeRegionAndWAL(region1);
973
974 WAL wal = createWAL(this.conf);
975 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
976 for (HColumnDescriptor hcd : htd.getFamilies()) {
977 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
978 }
979
980 final Get g = new Get(rowName);
981 Result result = region.get(g);
982 assertEquals(countPerFamily * htd.getFamilies().size(), result.size());
983
984
985 region.close(true);
986 wal.shutdown();
987
988 runWALSplit(this.conf);
989
990
991 Path editFile = WALSplitter.getSplitEditFilesSorted(this.fs, regionDir).first();
992 FSDataInputStream stream = fs.open(editFile);
993 stream.seek(ProtobufLogReader.PB_WAL_MAGIC.length);
994 Class<? extends DefaultWALProvider.Reader> logReaderClass =
995 conf.getClass("hbase.regionserver.hlog.reader.impl", ProtobufLogReader.class,
996 DefaultWALProvider.Reader.class);
997 DefaultWALProvider.Reader reader = logReaderClass.newInstance();
998 reader.init(this.fs, editFile, conf, stream);
999 final long headerLength = stream.getPos();
1000 reader.close();
1001 FileSystem spyFs = spy(this.fs);
1002 doAnswer(new Answer<FSDataInputStream>() {
1003
1004 @Override
1005 public FSDataInputStream answer(InvocationOnMock invocation) throws Throwable {
1006 FSDataInputStream stream = (FSDataInputStream) invocation.callRealMethod();
1007 Field field = FilterInputStream.class.getDeclaredField("in");
1008 field.setAccessible(true);
1009 final DFSInputStream in = (DFSInputStream) field.get(stream);
1010 DFSInputStream spyIn = spy(in);
1011 doAnswer(new Answer<Integer>() {
1012
1013 private long pos;
1014
1015 @Override
1016 public Integer answer(InvocationOnMock invocation) throws Throwable {
1017 if (pos >= headerLength) {
1018 throw new IOException("read over limit");
1019 }
1020 int b = (Integer) invocation.callRealMethod();
1021 if (b > 0) {
1022 pos += b;
1023 }
1024 return b;
1025 }
1026 }).when(spyIn).read(any(byte[].class), any(int.class), any(int.class));
1027 doAnswer(new Answer<Void>() {
1028
1029 @Override
1030 public Void answer(InvocationOnMock invocation) throws Throwable {
1031 invocation.callRealMethod();
1032 in.close();
1033 return null;
1034 }
1035 }).when(spyIn).close();
1036 field.set(stream, spyIn);
1037 return stream;
1038 }
1039 }).when(spyFs).open(eq(editFile));
1040
1041 WAL wal2 = createWAL(this.conf);
1042 HRegion region2;
1043 try {
1044
1045 region2 = HRegion.openHRegion(conf, spyFs, hbaseRootDir, hri, htd, wal2);
1046 assertEquals(result.size(), region2.get(g).size());
1047 } catch (IOException e) {
1048 assertEquals("read over limit", e.getMessage());
1049 }
1050 region2 = HRegion.openHRegion(conf, fs, hbaseRootDir, hri, htd, wal2);
1051 assertEquals(result.size(), region2.get(g).size());
1052 }
1053
1054
1055
1056
1057 private void testNameConflictWhenSplit(boolean largeFirst) throws IOException {
1058 final TableName tableName = TableName.valueOf("testReplayEditsWrittenIntoWAL");
1059 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
1060 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
1061 final Path basedir = FSUtils.getTableDir(hbaseRootDir, tableName);
1062 deleteDir(basedir);
1063
1064 final HTableDescriptor htd = createBasic1FamilyHTD(tableName);
1065 HRegion region = HBaseTestingUtility.createRegionAndWAL(hri, hbaseRootDir, this.hbaseWALRootDir, this.conf, htd);
1066 HBaseTestingUtility.closeRegionAndWAL(region);
1067 final byte[] family = htd.getColumnFamilies()[0].getName();
1068 final byte[] rowName = tableName.getName();
1069 FSWALEntry entry1 = createFSWALEntry(htd, hri, 1L, rowName, family, ee, mvcc, 1);
1070 FSWALEntry entry2 = createFSWALEntry(htd, hri, 2L, rowName, family, ee, mvcc, 2);
1071
1072 Path largeFile = new Path(logDir, "wal-1");
1073 Path smallFile = new Path(logDir, "wal-2");
1074 writerWALFile(largeFile, Arrays.asList(entry1, entry2));
1075 writerWALFile(smallFile, Arrays.asList(entry2));
1076 FileStatus first, second;
1077 if (largeFirst) {
1078 first = fs.getFileStatus(largeFile);
1079 second = fs.getFileStatus(smallFile);
1080 } else {
1081 first = fs.getFileStatus(smallFile);
1082 second = fs.getFileStatus(largeFile);
1083 }
1084 WALSplitter.splitLogFile(hbaseWALRootDir, first, fs, conf, null, null, null,
1085 RecoveryMode.LOG_SPLITTING, wals);
1086 WALSplitter.splitLogFile(hbaseWALRootDir, second, fs, conf, null, null, null,
1087 RecoveryMode.LOG_SPLITTING, wals);
1088 WAL wal = createWAL(this.conf);
1089 region = HRegion.openHRegion(conf, this.fs, hbaseRootDir, hri, htd, wal);
1090 assertTrue(region.getOpenSeqNum() > mvcc.getWritePoint());
1091 assertEquals(2, region.get(new Get(rowName)).size());
1092 }
1093
1094 @Test
1095 public void testNameConflictWhenSplit0() throws IOException {
1096 testNameConflictWhenSplit(true);
1097 }
1098
1099 @Test
1100 public void testNameConflictWhenSplit1() throws IOException {
1101 testNameConflictWhenSplit(false);
1102 }
1103
1104 static class MockWAL extends FSHLog {
1105 boolean doCompleteCacheFlush = false;
1106
1107 public MockWAL(FileSystem fs, Path walRootDir, String logName, Configuration conf)
1108 throws IOException {
1109 super(fs, walRootDir, logName, HConstants.HREGION_OLDLOGDIR_NAME, conf, null, true, null, null);
1110 }
1111
1112 @Override
1113 public void completeCacheFlush(byte[] encodedRegionName) {
1114 if (!doCompleteCacheFlush) {
1115 return;
1116 }
1117 super.completeCacheFlush(encodedRegionName);
1118 }
1119 }
1120
1121 private HTableDescriptor createBasic1FamilyHTD(final TableName tableName) {
1122 HTableDescriptor htd = new HTableDescriptor(tableName);
1123 HColumnDescriptor a = new HColumnDescriptor(Bytes.toBytes("a"));
1124 htd.addFamily(a);
1125 return htd;
1126 }
1127
1128 private MockWAL createMockWAL() throws IOException {
1129 MockWAL wal = new MockWAL(fs, hbaseWALRootDir, logName, conf);
1130
1131
1132 HBaseTestingUtility.setMaxRecoveryErrorCount(wal.getOutputStream(), 1);
1133 return wal;
1134 }
1135
1136
1137
1138 class TestFlusher implements FlushRequester {
1139
1140 @Override
1141 public boolean requestFlush(Region region, boolean force) {
1142 try {
1143 region.flush(force);
1144 return true;
1145 } catch (IOException e) {
1146 throw new RuntimeException("Exception flushing", e);
1147 }
1148 }
1149
1150 @Override
1151 public boolean requestDelayedFlush(Region region, long when, boolean forceFlushAllStores) {
1152 return true;
1153 }
1154
1155 @Override
1156 public void registerFlushRequestListener(FlushRequestListener listener) {
1157
1158 }
1159
1160 @Override
1161 public boolean unregisterFlushRequestListener(FlushRequestListener listener) {
1162 return false;
1163 }
1164
1165 @Override
1166 public void setGlobalMemstoreLimit(long globalMemStoreSize) {
1167
1168 }
1169 }
1170
1171 private WALKey createWALKey(final TableName tableName, final HRegionInfo hri,
1172 final MultiVersionConcurrencyControl mvcc) {
1173 return new WALKey(hri.getEncodedNameAsBytes(), tableName, 999, mvcc);
1174 }
1175
1176 private WALEdit createWALEdit(final byte[] rowName, final byte[] family, EnvironmentEdge ee,
1177 int index) {
1178 byte[] qualifierBytes = Bytes.toBytes(Integer.toString(index));
1179 byte[] columnBytes = Bytes.toBytes(Bytes.toString(family) + ":" + Integer.toString(index));
1180 WALEdit edit = new WALEdit();
1181 edit.add(new KeyValue(rowName, family, qualifierBytes, ee.currentTime(), columnBytes));
1182 return edit;
1183 }
1184
1185 private FSWALEntry createFSWALEntry(HTableDescriptor htd, HRegionInfo hri, long sequence,
1186 byte[] rowName, byte[] family, EnvironmentEdge ee, MultiVersionConcurrencyControl mvcc,
1187 int index) throws IOException {
1188 FSWALEntry entry =
1189 new FSWALEntry(sequence, createWALKey(htd.getTableName(), hri, mvcc), createWALEdit(
1190 rowName, family, ee, index), htd, hri, true);
1191 entry.stampRegionSequenceId();
1192 return entry;
1193 }
1194
1195 private void addWALEdits(final TableName tableName, final HRegionInfo hri, final byte[] rowName,
1196 final byte[] family, final int count, EnvironmentEdge ee, final WAL wal,
1197 final HTableDescriptor htd, final MultiVersionConcurrencyControl mvcc) throws IOException {
1198 for (int j = 0; j < count; j++) {
1199 wal.append(htd, hri, createWALKey(tableName, hri, mvcc),
1200 createWALEdit(rowName, family, ee, j), true);
1201 }
1202 wal.sync();
1203 }
1204
1205 static List<Put> addRegionEdits(final byte[] rowName, final byte[] family, final int count,
1206 EnvironmentEdge ee, final Region r, final String qualifierPrefix) throws IOException {
1207 List<Put> puts = new ArrayList<Put>();
1208 for (int j = 0; j < count; j++) {
1209 byte[] qualifier = Bytes.toBytes(qualifierPrefix + Integer.toString(j));
1210 Put p = new Put(rowName);
1211 p.add(family, qualifier, ee.currentTime(), rowName);
1212 r.put(p);
1213 puts.add(p);
1214 }
1215 return puts;
1216 }
1217
1218
1219
1220
1221
1222
1223 private HRegionInfo createBasic3FamilyHRegionInfo(final TableName tableName) {
1224 return new HRegionInfo(tableName, null, null, false);
1225 }
1226
1227
1228
1229
1230
1231
1232
1233 private Path runWALSplit(final Configuration c) throws IOException {
1234 List<Path> splits = WALSplitter.split(
1235 hbaseWALRootDir, logDir, oldLogDir, FSUtils.getWALFileSystem(c), c, wals);
1236
1237 assertEquals("splits=" + splits, 1, splits.size());
1238
1239 assertTrue(fs.exists(splits.get(0)));
1240 LOG.info("Split file=" + splits.get(0));
1241 return splits.get(0);
1242 }
1243
1244
1245
1246
1247
1248
1249 private WAL createWAL(final Configuration c) throws IOException {
1250 FSHLog wal = new FSHLog(FSUtils.getWALFileSystem(c), hbaseWALRootDir, logName, c);
1251
1252
1253 HBaseTestingUtility.setMaxRecoveryErrorCount(wal.getOutputStream(), 1);
1254 return wal;
1255 }
1256
1257 private HTableDescriptor createBasic3FamilyHTD(final TableName tableName) {
1258 HTableDescriptor htd = new HTableDescriptor(tableName);
1259 HColumnDescriptor a = new HColumnDescriptor(Bytes.toBytes("a"));
1260 htd.addFamily(a);
1261 HColumnDescriptor b = new HColumnDescriptor(Bytes.toBytes("b"));
1262 htd.addFamily(b);
1263 HColumnDescriptor c = new HColumnDescriptor(Bytes.toBytes("c"));
1264 htd.addFamily(c);
1265 return htd;
1266 }
1267
1268 private void writerWALFile(Path file, List<FSWALEntry> entries) throws IOException {
1269 fs.mkdirs(file.getParent());
1270 ProtobufLogWriter writer = new ProtobufLogWriter();
1271 writer.init(fs, file, conf, true);
1272 for (FSWALEntry entry : entries) {
1273 writer.append(entry);
1274 }
1275 writer.sync();
1276 writer.close();
1277 }
1278 }