001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver.wal;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertFalse;
022import static org.junit.jupiter.api.Assertions.assertNotNull;
023import static org.junit.jupiter.api.Assertions.assertTrue;
024
025import java.io.EOFException;
026import java.io.IOException;
027import java.io.InterruptedIOException;
028import java.util.ArrayList;
029import java.util.HashSet;
030import java.util.List;
031import java.util.Set;
032import java.util.concurrent.atomic.AtomicBoolean;
033import org.apache.hadoop.conf.Configuration;
034import org.apache.hadoop.fs.Path;
035import org.apache.hadoop.hbase.Cell;
036import org.apache.hadoop.hbase.HConstants;
037import org.apache.hadoop.hbase.TableName;
038import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
039import org.apache.hadoop.hbase.client.Put;
040import org.apache.hadoop.hbase.client.RegionInfo;
041import org.apache.hadoop.hbase.client.Result;
042import org.apache.hadoop.hbase.client.ResultScanner;
043import org.apache.hadoop.hbase.client.Scan;
044import org.apache.hadoop.hbase.client.Table;
045import org.apache.hadoop.hbase.client.TableDescriptor;
046import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
047import org.apache.hadoop.hbase.fs.HFileSystem;
048import org.apache.hadoop.hbase.regionserver.HRegion;
049import org.apache.hadoop.hbase.testclassification.LargeTests;
050import org.apache.hadoop.hbase.testclassification.VerySlowRegionServerTests;
051import org.apache.hadoop.hbase.util.Bytes;
052import org.apache.hadoop.hbase.util.CommonFSUtils;
053import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
054import org.apache.hadoop.hbase.util.JVMClusterUtil;
055import org.apache.hadoop.hbase.util.RecoverLeaseFSUtils;
056import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
057import org.apache.hadoop.hbase.wal.FSHLogProvider;
058import org.apache.hadoop.hbase.wal.WAL;
059import org.apache.hadoop.hbase.wal.WALFactory;
060import org.apache.hadoop.hbase.wal.WALStreamReader;
061import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
062import org.apache.hadoop.hdfs.server.datanode.DataNode;
063import org.junit.jupiter.api.BeforeAll;
064import org.junit.jupiter.api.Tag;
065import org.junit.jupiter.api.Test;
066import org.slf4j.Logger;
067import org.slf4j.LoggerFactory;
068
069@Tag(VerySlowRegionServerTests.TAG)
070@Tag(LargeTests.TAG)
071public class TestLogRolling extends AbstractTestLogRolling {
072
073  private static final Logger LOG = LoggerFactory.getLogger(TestLogRolling.class);
074
075  @BeforeAll
076  public static void setUpBeforeClass() throws Exception {
077    // TODO: testLogRollOnDatanodeDeath fails if short circuit reads are on under the hadoop2
078    // profile. See HBASE-9337 for related issues.
079    System.setProperty("hbase.tests.use.shortcircuit.reads", "false");
080
081    /**** configuration for testLogRollOnDatanodeDeath ****/
082    // lower the namenode & datanode heartbeat so the namenode
083    // quickly detects datanode failures
084    Configuration conf = TEST_UTIL.getConfiguration();
085    conf.setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
086    conf.setInt("dfs.heartbeat.interval", 1);
087    // the namenode might still try to choose the recently-dead datanode
088    // for a pipeline, so try to a new pipeline multiple times
089    conf.setInt("dfs.client.block.write.retries", 30);
090    conf.setInt("hbase.regionserver.hlog.tolerable.lowreplication", 2);
091    conf.setInt("hbase.regionserver.hlog.lowreplication.rolllimit", 3);
092    conf.set(WALFactory.WAL_PROVIDER, "filesystem");
093  }
094
095  public static class SlowSyncLogWriter extends ProtobufLogWriter {
096    @Override
097    public void sync(boolean forceSync) throws IOException {
098      try {
099        Thread.sleep(syncLatencyMillis);
100      } catch (InterruptedException e) {
101        InterruptedIOException ex = new InterruptedIOException();
102        ex.initCause(e);
103        throw ex;
104      }
105      super.sync(forceSync);
106    }
107  }
108
109  @Override
110  protected void setSlowLogWriter(Configuration conf) {
111    conf.set(FSHLogProvider.WRITER_IMPL, SlowSyncLogWriter.class.getName());
112  }
113
114  @Override
115  protected void setDefaultLogWriter(Configuration conf) {
116    conf.set(FSHLogProvider.WRITER_IMPL, ProtobufLogWriter.class.getName());
117  }
118
119  void batchWriteAndWait(Table table, final FSHLog log, int start, boolean expect, int timeout)
120    throws IOException {
121    for (int i = 0; i < 10; i++) {
122      Put put = new Put(Bytes.toBytes("row" + String.format("%1$04d", (start + i))));
123      put.addColumn(HConstants.CATALOG_FAMILY, null, value);
124      table.put(put);
125    }
126    Put tmpPut = new Put(Bytes.toBytes("tmprow"));
127    tmpPut.addColumn(HConstants.CATALOG_FAMILY, null, value);
128    long startTime = EnvironmentEdgeManager.currentTime();
129    long remaining = timeout;
130    while (remaining > 0) {
131      if (log.isLowReplicationRollEnabled() == expect) {
132        break;
133      } else {
134        // Trigger calling FSHlog#checkLowReplication()
135        table.put(tmpPut);
136        try {
137          Thread.sleep(200);
138        } catch (InterruptedException e) {
139          // continue
140        }
141        remaining = timeout - (EnvironmentEdgeManager.currentTime() - startTime);
142      }
143    }
144  }
145
146  @Test
147  public void testSlowSyncLogRolling() throws Exception {
148    // Create the test table
149    TableDescriptor desc = TableDescriptorBuilder.newBuilder(TableName.valueOf(getName()))
150      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(HConstants.CATALOG_FAMILY)).build();
151    admin.createTable(desc);
152    try (Table table = TEST_UTIL.getConnection().getTable(desc.getTableName())) {
153      server = TEST_UTIL.getRSForFirstRegionInTable(desc.getTableName());
154      RegionInfo region = server.getRegions(desc.getTableName()).get(0).getRegionInfo();
155      final AbstractFSWAL<?> log = getWALAndRegisterSlowSyncHook(region);
156
157      // Set default log writer, no additional latency to any sync on the hlog.
158      checkSlowSync(log, table, -1, 10, false);
159
160      // Adds 200 ms of latency to any sync on the hlog. This should be more than sufficient to
161      // trigger slow sync warnings.
162      // Write some data.
163      // We need to write at least 5 times, but double it. We should only request
164      // a SLOW_SYNC roll once in the current interval.
165      checkSlowSync(log, table, 200, 10, true);
166
167      // Adds 5000 ms of latency to any sync on the hlog. This will trip the other threshold.
168      // Write some data. Should only take one sync.
169      checkSlowSync(log, table, 5000, 1, true);
170
171      // Set default log writer, no additional latency to any sync on the hlog.
172      checkSlowSync(log, table, -1, 10, false);
173    }
174  }
175
176  /**
177   * Tests that logs are rolled upon detecting datanode death Requires an HDFS jar with HDFS-826 &
178   * syncFs() support (HDFS-200)
179   */
180  @Test
181  public void testLogRollOnDatanodeDeath() throws Exception {
182
183    Long oldValue = TEST_UTIL.getConfiguration()
184      .getLong("hbase.regionserver.hlog.check.lowreplication.interval", -1);
185
186    try {
187      /**
188       * When we reuse the code of AsyncFSWAL to FSHLog, the low replication is only checked by
189       * {@link LogRoller#checkLowReplication},so in order to make this test spend less time,we
190       * should minimize following config which is maximized by
191       * {@link AbstractTestLogRolling#setUpBeforeClass}
192       */
193      TEST_UTIL.getConfiguration().setLong("hbase.regionserver.hlog.check.lowreplication.interval",
194        1000);
195      this.tearDown();
196      this.setUp();
197
198      TEST_UTIL.ensureSomeRegionServersAvailable(2);
199      assertTrue(fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) == 2,
200        "This test requires WAL file replication set to 2.");
201      LOG.info("Replication=" + fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
202
203      this.server = cluster.getRegionServer(0);
204
205      // Create the test table and open it
206      TableDescriptor desc = TableDescriptorBuilder.newBuilder(TableName.valueOf(getName()))
207        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(HConstants.CATALOG_FAMILY)).build();
208
209      admin.createTable(desc);
210      Table table = TEST_UTIL.getConnection().getTable(desc.getTableName());
211
212      server = TEST_UTIL.getRSForFirstRegionInTable(desc.getTableName());
213      RegionInfo region = server.getRegions(desc.getTableName()).get(0).getRegionInfo();
214      final FSHLog log = (FSHLog) server.getWAL(region);
215      final AtomicBoolean lowReplicationHookCalled = new AtomicBoolean(false);
216
217      log.registerWALActionsListener(new WALActionsListener() {
218        @Override
219        public void logRollRequested(WALActionsListener.RollRequestReason reason) {
220          switch (reason) {
221            case LOW_REPLICATION:
222              lowReplicationHookCalled.lazySet(true);
223              break;
224            default:
225              break;
226          }
227        }
228      });
229
230      // add up the datanode count, to ensure proper replication when we kill 1
231      // This function is synchronous; when it returns, the dfs cluster is active
232      // We start 3 servers and then stop 2 to avoid a directory naming conflict
233      // when we stop/start a namenode later, as mentioned in HBASE-5163
234      List<DataNode> existingNodes = dfsCluster.getDataNodes();
235      int numDataNodes = 3;
236      TEST_UTIL.getConfiguration().setLong("hbase.regionserver.hlog.check.lowreplication.interval",
237        1000);
238      dfsCluster.startDataNodes(TEST_UTIL.getConfiguration(), numDataNodes, true, null, null);
239      List<DataNode> allNodes = dfsCluster.getDataNodes();
240      for (int i = allNodes.size() - 1; i >= 0; i--) {
241        if (existingNodes.contains(allNodes.get(i))) {
242          dfsCluster.stopDataNode(i);
243        }
244      }
245
246      assertTrue(
247        dfsCluster.getDataNodes().size()
248            >= fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) + 1,
249        "DataNodes " + dfsCluster.getDataNodes().size() + " default replication "
250          + fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
251
252      writeData(table, 2);
253
254      long curTime = EnvironmentEdgeManager.currentTime();
255      LOG.info("log.getCurrentFileName(): " + log.getCurrentFileName());
256      long oldFilenum = AbstractFSWALProvider.extractFileNumFromWAL(log);
257      assertTrue(curTime > oldFilenum && oldFilenum != -1,
258        "Log should have a timestamp older than now");
259
260      assertTrue(oldFilenum == AbstractFSWALProvider.extractFileNumFromWAL(log),
261        "The log shouldn't have rolled yet");
262      final DatanodeInfo[] pipeline = log.getPipeline();
263      assertTrue(pipeline.length == fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
264
265      // kill a datanode in the pipeline to force a log roll on the next sync()
266      // This function is synchronous, when it returns the node is killed.
267      assertTrue(dfsCluster.stopDataNode(pipeline[0].getName()) != null);
268
269      // this write should succeed, but trigger a log roll
270      writeData(table, 2);
271
272      TEST_UTIL.waitFor(10000, 100, () -> {
273        long newFilenum = AbstractFSWALProvider.extractFileNumFromWAL(log);
274        return newFilenum > oldFilenum && newFilenum > curTime && lowReplicationHookCalled.get();
275      });
276
277      long newFilenum = AbstractFSWALProvider.extractFileNumFromWAL(log);
278
279      // write some more log data (this should use a new hdfs_out)
280      writeData(table, 3);
281      assertTrue(AbstractFSWALProvider.extractFileNumFromWAL(log) == newFilenum,
282        "The log should not roll again.");
283      // kill another datanode in the pipeline, so the replicas will be lower than
284      // the configured value 2.
285      assertTrue(dfsCluster.stopDataNode(pipeline[1].getName()) != null);
286
287      batchWriteAndWait(table, log, 3, false, 14000);
288      int replication = log.getLogReplication();
289      assertTrue(!log.isLowReplicationRollEnabled(),
290        "LowReplication Roller should've been disabled, current replication=" + replication);
291
292      dfsCluster.startDataNodes(TEST_UTIL.getConfiguration(), 1, true, null, null);
293
294      // Force roll writer. The new log file will have the default replications,
295      // and the LowReplication Roller will be enabled.
296      log.rollWriter(true);
297      batchWriteAndWait(table, log, 13, true, 10000);
298      replication = log.getLogReplication();
299      assertTrue(replication == fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()),
300        "New log file should have the default replication instead of " + replication);
301      assertTrue(log.isLowReplicationRollEnabled(), "LowReplication Roller should've been enabled");
302    } finally {
303      TEST_UTIL.getConfiguration().setLong("hbase.regionserver.hlog.check.lowreplication.interval",
304        oldValue);
305    }
306  }
307
308  /**
309   * Test that WAL is rolled when all data nodes in the pipeline have been restarted.
310   */
311  @Test
312  public void testLogRollOnPipelineRestart() throws Exception {
313    LOG.info("Starting testLogRollOnPipelineRestart");
314    assertTrue(fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) > 1,
315      "This test requires WAL file replication.");
316    LOG.info("Replication=" + fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
317    // When the hbase:meta table can be opened, the region servers are running
318    Table t = TEST_UTIL.getConnection().getTable(TableName.META_TABLE_NAME);
319    try {
320      this.server = cluster.getRegionServer(0);
321
322      // Create the test table and open it
323      TableDescriptor desc = TableDescriptorBuilder.newBuilder(TableName.valueOf(getName()))
324        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(HConstants.CATALOG_FAMILY)).build();
325
326      admin.createTable(desc);
327      Table table = TEST_UTIL.getConnection().getTable(desc.getTableName());
328
329      server = TEST_UTIL.getRSForFirstRegionInTable(desc.getTableName());
330      RegionInfo region = server.getRegions(desc.getTableName()).get(0).getRegionInfo();
331      final WAL log = server.getWAL(region);
332      final List<Path> paths = new ArrayList<>(1);
333      final List<Integer> preLogRolledCalled = new ArrayList<>();
334
335      paths.add(AbstractFSWALProvider.getCurrentFileName(log));
336      log.registerWALActionsListener(new WALActionsListener() {
337
338        @Override
339        public void preLogRoll(Path oldFile, Path newFile) {
340          LOG.debug("preLogRoll: oldFile=" + oldFile + " newFile=" + newFile);
341          preLogRolledCalled.add(1);
342        }
343
344        @Override
345        public void postLogRoll(Path oldFile, Path newFile) {
346          paths.add(newFile);
347        }
348      });
349
350      writeData(table, 1002);
351
352      long curTime = EnvironmentEdgeManager.currentTime();
353      LOG.info("log.getCurrentFileName()): " + AbstractFSWALProvider.getCurrentFileName(log));
354      long oldFilenum = AbstractFSWALProvider.extractFileNumFromWAL(log);
355      assertTrue(curTime > oldFilenum && oldFilenum != -1,
356        "Log should have a timestamp older than now");
357
358      assertTrue(oldFilenum == AbstractFSWALProvider.extractFileNumFromWAL(log),
359        "The log shouldn't have rolled yet");
360
361      // roll all datanodes in the pipeline
362      dfsCluster.restartDataNodes();
363      Thread.sleep(1000);
364      dfsCluster.waitActive();
365      LOG.info("Data Nodes restarted");
366      validateData(table, 1002);
367
368      // this write should succeed, but trigger a log roll
369      writeData(table, 1003);
370      long newFilenum = AbstractFSWALProvider.extractFileNumFromWAL(log);
371
372      assertTrue(newFilenum > oldFilenum && newFilenum > curTime,
373        "Missing datanode should've triggered a log roll");
374      validateData(table, 1003);
375
376      writeData(table, 1004);
377
378      // roll all datanode again
379      dfsCluster.restartDataNodes();
380      Thread.sleep(1000);
381      dfsCluster.waitActive();
382      LOG.info("Data Nodes restarted");
383      validateData(table, 1004);
384
385      // this write should succeed, but trigger a log roll
386      writeData(table, 1005);
387
388      // force a log roll to read back and verify previously written logs
389      log.rollWriter(true);
390      assertTrue(preLogRolledCalled.size() >= 1,
391        "preLogRolledCalled has size of " + preLogRolledCalled.size());
392
393      // read back the data written
394      Set<String> loggedRows = new HashSet<>();
395      for (Path p : paths) {
396        LOG.debug("recovering lease for " + p);
397        RecoverLeaseFSUtils.recoverFileLease(((HFileSystem) fs).getBackingFs(), p,
398          TEST_UTIL.getConfiguration(), null);
399
400        LOG.debug("Reading WAL " + CommonFSUtils.getPath(p));
401        try (WALStreamReader reader =
402          WALFactory.createStreamReader(fs, p, TEST_UTIL.getConfiguration())) {
403          WAL.Entry entry;
404          while ((entry = reader.next()) != null) {
405            LOG.debug("#" + entry.getKey().getSequenceId() + ": " + entry.getEdit().getCells());
406            for (Cell cell : entry.getEdit().getCells()) {
407              loggedRows.add(
408                Bytes.toStringBinary(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
409            }
410          }
411        } catch (EOFException e) {
412          LOG.debug("EOF reading file " + CommonFSUtils.getPath(p));
413        }
414      }
415
416      // verify the written rows are there
417      assertTrue(loggedRows.contains("row1002"));
418      assertTrue(loggedRows.contains("row1003"));
419      assertTrue(loggedRows.contains("row1004"));
420      assertTrue(loggedRows.contains("row1005"));
421
422      // flush all regions
423      for (HRegion r : server.getOnlineRegionsLocalContext()) {
424        try {
425          r.flush(true);
426        } catch (Exception e) {
427          // This try/catch was added by HBASE-14317. It is needed
428          // because this issue tightened up the semantic such that
429          // a failed append could not be followed by a successful
430          // sync. What is coming out here is a failed sync, a sync
431          // that used to 'pass'.
432          LOG.info(e.toString(), e);
433        }
434      }
435
436      ResultScanner scanner = table.getScanner(new Scan());
437      try {
438        for (int i = 2; i <= 5; i++) {
439          Result r = scanner.next();
440          assertNotNull(r);
441          assertFalse(r.isEmpty());
442          assertEquals("row100" + i, Bytes.toString(r.getRow()));
443        }
444      } finally {
445        scanner.close();
446      }
447
448      // verify that no region servers aborted
449      for (JVMClusterUtil.RegionServerThread rsThread : TEST_UTIL.getHBaseCluster()
450        .getRegionServerThreads()) {
451        assertFalse(rsThread.getRegionServer().isAborted());
452      }
453    } finally {
454      if (t != null) t.close();
455    }
456  }
457
458}