001/**
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019package org.apache.hadoop.hbase.regionserver.wal;
020
021import static org.junit.Assert.assertEquals;
022import static org.junit.Assert.assertTrue;
023
024import java.io.IOException;
025import org.apache.hadoop.conf.Configuration;
026import org.apache.hadoop.fs.FileSystem;
027import org.apache.hadoop.hbase.HBaseTestingUtility;
028import org.apache.hadoop.hbase.HConstants;
029import org.apache.hadoop.hbase.MiniHBaseCluster;
030import org.apache.hadoop.hbase.ServerName;
031import org.apache.hadoop.hbase.StartMiniClusterOption;
032import org.apache.hadoop.hbase.TableName;
033import org.apache.hadoop.hbase.client.Admin;
034import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
035import org.apache.hadoop.hbase.client.Get;
036import org.apache.hadoop.hbase.client.Put;
037import org.apache.hadoop.hbase.client.RegionInfo;
038import org.apache.hadoop.hbase.client.Result;
039import org.apache.hadoop.hbase.client.Table;
040import org.apache.hadoop.hbase.client.TableDescriptor;
041import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
042import org.apache.hadoop.hbase.regionserver.HRegion;
043import org.apache.hadoop.hbase.regionserver.HRegionServer;
044import org.apache.hadoop.hbase.regionserver.Store;
045import org.apache.hadoop.hbase.util.Bytes;
046import org.apache.hadoop.hbase.util.Threads;
047import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
048import org.apache.hadoop.hbase.wal.WAL;
049import org.apache.hadoop.hbase.wal.WALFactory;
050import org.apache.hadoop.hdfs.MiniDFSCluster;
051import org.junit.After;
052import org.junit.Assert;
053import org.junit.Before;
054import org.junit.BeforeClass;
055import org.junit.Rule;
056import org.junit.Test;
057import org.junit.rules.TestName;
058import org.slf4j.Logger;
059import org.slf4j.LoggerFactory;
060
061/**
062 * Test log deletion as logs are rolled.
063 */
064public abstract class AbstractTestLogRolling  {
065  private static final Logger LOG = LoggerFactory.getLogger(AbstractTestLogRolling.class);
066  protected HRegionServer server;
067  protected String tableName;
068  protected byte[] value;
069  protected FileSystem fs;
070  protected MiniDFSCluster dfsCluster;
071  protected Admin admin;
072  protected MiniHBaseCluster cluster;
073  protected static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
074  @Rule public final TestName name = new TestName();
075
076  public AbstractTestLogRolling()  {
077    this.server = null;
078    this.tableName = null;
079
080    String className = this.getClass().getName();
081    StringBuilder v = new StringBuilder(className);
082    while (v.length() < 1000) {
083      v.append(className);
084    }
085    this.value = Bytes.toBytes(v.toString());
086  }
087
088  // Need to override this setup so we can edit the config before it gets sent
089  // to the HDFS & HBase cluster startup.
090  @BeforeClass
091  public static void setUpBeforeClass() throws Exception {
092    /**** configuration for testLogRolling ****/
093    // Force a region split after every 768KB
094    Configuration conf = TEST_UTIL.getConfiguration();
095    conf.setLong(HConstants.HREGION_MAX_FILESIZE, 768L * 1024L);
096
097    // We roll the log after every 32 writes
098    conf.setInt("hbase.regionserver.maxlogentries", 32);
099
100    conf.setInt("hbase.regionserver.logroll.errors.tolerated", 2);
101    conf.setInt("hbase.rpc.timeout", 10 * 1000);
102
103    // For less frequently updated regions flush after every 2 flushes
104    conf.setInt("hbase.hregion.memstore.optionalflushcount", 2);
105
106    // We flush the cache after every 8192 bytes
107    conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 8192);
108
109    // Increase the amount of time between client retries
110    conf.setLong("hbase.client.pause", 10 * 1000);
111
112    // Reduce thread wake frequency so that other threads can get
113    // a chance to run.
114    conf.setInt(HConstants.THREAD_WAKE_FREQUENCY, 2 * 1000);
115
116    // disable low replication check for log roller to get a more stable result
117    // TestWALOpenAfterDNRollingStart will test this option.
118    conf.setLong("hbase.regionserver.hlog.check.lowreplication.interval", 24L * 60 * 60 * 1000);
119  }
120
121  @Before
122  public void setUp() throws Exception {
123    // Use 2 DataNodes and default values for other StartMiniCluster options.
124    TEST_UTIL.startMiniCluster(StartMiniClusterOption.builder().numDataNodes(2).build());
125
126    cluster = TEST_UTIL.getHBaseCluster();
127    dfsCluster = TEST_UTIL.getDFSCluster();
128    fs = TEST_UTIL.getTestFileSystem();
129    admin = TEST_UTIL.getAdmin();
130
131    // disable region rebalancing (interferes with log watching)
132    cluster.getMaster().balanceSwitch(false);
133  }
134
135  @After
136  public void tearDown() throws Exception  {
137    TEST_UTIL.shutdownMiniCluster();
138  }
139
140  protected void startAndWriteData() throws IOException, InterruptedException {
141    // When the hbase:meta table can be opened, the region servers are running
142    TEST_UTIL.getConnection().getTable(TableName.META_TABLE_NAME);
143    this.server = cluster.getRegionServerThreads().get(0).getRegionServer();
144
145    Table table = createTestTable(this.tableName);
146
147    server = TEST_UTIL.getRSForFirstRegionInTable(table.getName());
148    for (int i = 1; i <= 256; i++) {    // 256 writes should cause 8 log rolls
149      doPut(table, i);
150      if (i % 32 == 0) {
151        // After every 32 writes sleep to let the log roller run
152        try {
153          Thread.sleep(2000);
154        } catch (InterruptedException e) {
155          // continue
156        }
157      }
158    }
159  }
160
161  /**
162   * Tests that log rolling doesn't hang when no data is written.
163   */
164  @Test
165  public void testLogRollOnNothingWritten() throws Exception {
166    final Configuration conf = TEST_UTIL.getConfiguration();
167    final WALFactory wals =
168      new WALFactory(conf, ServerName.valueOf("test.com", 8080, 1).toString());
169    final WAL newLog = wals.getWAL(null);
170    try {
171      // Now roll the log before we write anything.
172      newLog.rollWriter(true);
173    } finally {
174      wals.close();
175    }
176  }
177
178  private void assertLogFileSize(WAL log) {
179    if (AbstractFSWALProvider.getNumRolledLogFiles(log) > 0) {
180      assertTrue(AbstractFSWALProvider.getLogFileSize(log) > 0);
181    } else {
182      assertEquals(0, AbstractFSWALProvider.getLogFileSize(log));
183    }
184  }
185
186  /**
187   * Tests that logs are deleted
188   */
189  @Test
190  public void testLogRolling() throws Exception {
191    this.tableName = getName();
192    // TODO: Why does this write data take for ever?
193    startAndWriteData();
194    RegionInfo region = server.getRegions(TableName.valueOf(tableName)).get(0).getRegionInfo();
195    final WAL log = server.getWAL(region);
196    LOG.info("after writing there are " + AbstractFSWALProvider.getNumRolledLogFiles(log) + " log files");
197    assertLogFileSize(log);
198
199    // flush all regions
200    for (HRegion r : server.getOnlineRegionsLocalContext()) {
201      r.flush(true);
202    }
203
204    // Now roll the log
205    log.rollWriter();
206
207    int count = AbstractFSWALProvider.getNumRolledLogFiles(log);
208    LOG.info("after flushing all regions and rolling logs there are " + count + " log files");
209    assertTrue(("actual count: " + count), count <= 2);
210    assertLogFileSize(log);
211  }
212
213  protected String getName() {
214    return "TestLogRolling-" + name.getMethodName();
215  }
216
217  void writeData(Table table, int rownum) throws IOException {
218    doPut(table, rownum);
219
220    // sleep to let the log roller run (if it needs to)
221    try {
222      Thread.sleep(2000);
223    } catch (InterruptedException e) {
224      // continue
225    }
226  }
227
228  void validateData(Table table, int rownum) throws IOException {
229    String row = "row" + String.format("%1$04d", rownum);
230    Get get = new Get(Bytes.toBytes(row));
231    get.addFamily(HConstants.CATALOG_FAMILY);
232    Result result = table.get(get);
233    assertTrue(result.size() == 1);
234    assertTrue(Bytes.equals(value,
235                result.getValue(HConstants.CATALOG_FAMILY, null)));
236    LOG.info("Validated row " + row);
237  }
238
239  /**
240   * Tests that logs are deleted when some region has a compaction
241   * record in WAL and no other records. See HBASE-8597.
242   */
243  @Test
244  public void testCompactionRecordDoesntBlockRolling() throws Exception {
245    Table table = null;
246
247    // When the hbase:meta table can be opened, the region servers are running
248    Table t = TEST_UTIL.getConnection().getTable(TableName.META_TABLE_NAME);
249    try {
250      table = createTestTable(getName());
251
252      server = TEST_UTIL.getRSForFirstRegionInTable(table.getName());
253      HRegion region = server.getRegions(table.getName()).get(0);
254      final WAL log = server.getWAL(region.getRegionInfo());
255      Store s = region.getStore(HConstants.CATALOG_FAMILY);
256
257      //have to flush namespace to ensure it doesn't affect wall tests
258      admin.flush(TableName.NAMESPACE_TABLE_NAME);
259
260      // Put some stuff into table, to make sure we have some files to compact.
261      for (int i = 1; i <= 2; ++i) {
262        doPut(table, i);
263        admin.flush(table.getName());
264      }
265      doPut(table, 3); // don't flush yet, or compaction might trigger before we roll WAL
266      assertEquals("Should have no WAL after initial writes", 0,
267        AbstractFSWALProvider.getNumRolledLogFiles(log));
268      assertEquals(2, s.getStorefilesCount());
269
270      // Roll the log and compact table, to have compaction record in the 2nd WAL.
271      log.rollWriter();
272      assertEquals("Should have WAL; one table is not flushed", 1,
273        AbstractFSWALProvider.getNumRolledLogFiles(log));
274      admin.flush(table.getName());
275      region.compact(false);
276      // Wait for compaction in case if flush triggered it before us.
277      Assert.assertNotNull(s);
278      for (int waitTime = 3000; s.getStorefilesCount() > 1 && waitTime > 0; waitTime -= 200) {
279        Threads.sleepWithoutInterrupt(200);
280      }
281      assertEquals("Compaction didn't happen", 1, s.getStorefilesCount());
282
283      // Write some value to the table so the WAL cannot be deleted until table is flushed.
284      doPut(table, 0); // Now 2nd WAL will have both compaction and put record for table.
285      log.rollWriter(); // 1st WAL deleted, 2nd not deleted yet.
286      assertEquals("Should have WAL; one table is not flushed", 1,
287        AbstractFSWALProvider.getNumRolledLogFiles(log));
288
289      // Flush table to make latest WAL obsolete; write another record, and roll again.
290      admin.flush(table.getName());
291      doPut(table, 1);
292      log.rollWriter(); // Now 2nd WAL is deleted and 3rd is added.
293      assertEquals("Should have 1 WALs at the end", 1,
294        AbstractFSWALProvider.getNumRolledLogFiles(log));
295    } finally {
296      if (t != null) t.close();
297      if (table != null) table.close();
298    }
299  }
300
301  protected void doPut(Table table, int i) throws IOException {
302    Put put = new Put(Bytes.toBytes("row" + String.format("%1$04d", i)));
303    put.addColumn(HConstants.CATALOG_FAMILY, null, value);
304    table.put(put);
305  }
306
307  protected Table createTestTable(String tableName) throws IOException {
308    // Create the test table and open it
309    TableDescriptor desc = TableDescriptorBuilder.newBuilder(TableName.valueOf(getName()))
310        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(HConstants.CATALOG_FAMILY)).build();
311    admin.createTable(desc);
312    return TEST_UTIL.getConnection().getTable(desc.getTableName());
313  }
314}