View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.wal;
20  
21  import java.io.IOException;
22  import java.util.HashSet;
23  import java.util.Random;
24  import java.util.Set;
25  
26  import static org.junit.Assert.assertEquals;
27  import static org.apache.hadoop.hbase.wal.BoundedRegionGroupingProvider.NUM_REGION_GROUPS;
28  import static org.apache.hadoop.hbase.wal.BoundedRegionGroupingProvider.DEFAULT_NUM_REGION_GROUPS;
29  import static org.apache.hadoop.hbase.wal.WALFactory.WAL_PROVIDER;
30  
31  import org.apache.commons.logging.Log;
32  import org.apache.commons.logging.LogFactory;
33  import org.apache.hadoop.conf.Configuration;
34  import org.apache.hadoop.fs.FileStatus;
35  import org.apache.hadoop.fs.FileSystem;
36  import org.apache.hadoop.fs.Path;
37  import org.apache.hadoop.hbase.HBaseTestingUtility;
38  import org.apache.hadoop.hbase.testclassification.LargeTests;
39  import org.apache.hadoop.hbase.util.Bytes;
40  import org.apache.hadoop.hbase.util.FSUtils;
41  import org.junit.After;
42  import org.junit.AfterClass;
43  import org.junit.Before;
44  import org.junit.BeforeClass;
45  import org.junit.Rule;
46  import org.junit.Test;
47  import org.junit.experimental.categories.Category;
48  import org.junit.rules.TestName;
49  
50  @Category(LargeTests.class)
51  public class TestBoundedRegionGroupingProvider {
52    protected static final Log LOG = LogFactory.getLog(TestBoundedRegionGroupingProvider.class);
53  
54    @Rule
55    public TestName currentTest = new TestName();
56    protected static Configuration conf;
57    protected static FileSystem fs;
58    protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
59  
60    @Before
61    public void setUp() throws Exception {
62      FileStatus[] entries = fs.listStatus(new Path("/"));
63      for (FileStatus dir : entries) {
64        fs.delete(dir.getPath(), true);
65      }
66    }
67  
68    @After
69    public void tearDown() throws Exception {
70    }
71  
72    @BeforeClass
73    public static void setUpBeforeClass() throws Exception {
74      conf = TEST_UTIL.getConfiguration();
75      // Make block sizes small.
76      conf.setInt("dfs.blocksize", 1024 * 1024);
77      // quicker heartbeat interval for faster DN death notification
78      conf.setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
79      conf.setInt("dfs.heartbeat.interval", 1);
80      conf.setInt("dfs.client.socket-timeout", 5000);
81  
82      // faster failover with cluster.shutdown();fs.close() idiom
83      conf.setInt("hbase.ipc.client.connect.max.retries", 1);
84      conf.setInt("dfs.client.block.recovery.retries", 1);
85      conf.setInt("hbase.ipc.client.connection.maxidletime", 500);
86  
87      conf.setClass(WAL_PROVIDER, BoundedRegionGroupingProvider.class, WALProvider.class);
88  
89      TEST_UTIL.startMiniDFSCluster(3);
90  
91      fs = TEST_UTIL.getDFSCluster().getFileSystem();
92    }
93  
94    @AfterClass
95    public static void tearDownAfterClass() throws Exception {
96      TEST_UTIL.shutdownMiniCluster();
97    }
98  
99    /**
100    * Write to a log file with three concurrent threads and verifying all data is written.
101    */
102   @Test
103   public void testConcurrentWrites() throws Exception {
104     // Run the WPE tool with three threads writing 3000 edits each concurrently.
105     // When done, verify that all edits were written.
106     int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
107         new String [] {"-threads", "3", "-verify", "-noclosefs", "-iterations", "3000"});
108     assertEquals(0, errCode);
109   }
110 
111   /**
112    * Make sure we can successfully run with more regions then our bound.
113    */
114   @Test
115   public void testMoreRegionsThanBound() throws Exception {
116     final String parallelism = Integer.toString(DEFAULT_NUM_REGION_GROUPS * 2);
117     int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
118         new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
119             "-regions", parallelism});
120     assertEquals(0, errCode);
121   }
122 
123   @Test
124   public void testBoundsGreaterThanDefault() throws Exception {
125     final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
126     try {
127       conf.setInt(NUM_REGION_GROUPS, temp*4);
128       final String parallelism = Integer.toString(temp*4);
129       int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
130           new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
131               "-regions", parallelism});
132       assertEquals(0, errCode);
133     } finally {
134       conf.setInt(NUM_REGION_GROUPS, temp);
135     }
136   }
137 
138   @Test
139   public void testMoreRegionsThanBoundWithBoundsGreaterThanDefault() throws Exception {
140     final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
141     try {
142       conf.setInt(NUM_REGION_GROUPS, temp*4);
143       final String parallelism = Integer.toString(temp*4*2);
144       int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
145           new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
146               "-regions", parallelism});
147       assertEquals(0, errCode);
148     } finally {
149       conf.setInt(NUM_REGION_GROUPS, temp);
150     }
151   }
152 
153   /**
154    * Ensure that we can use Set.add to deduplicate WALs
155    */
156   @Test
157   public void setMembershipDedups() throws IOException {
158     final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
159     WALFactory wals = null;
160     try {
161       conf.setInt(NUM_REGION_GROUPS, temp*4);
162       // Set HDFS root directory for storing WAL
163       FSUtils.setRootDir(conf, TEST_UTIL.getDataTestDirOnTestFS());
164 
165       wals = new WALFactory(conf, null, currentTest.getMethodName());
166       final Set<WAL> seen = new HashSet<WAL>(temp*4);
167       final Random random = new Random();
168       int count = 0;
169       // we know that this should see one of the wals more than once
170       for (int i = 0; i < temp*8; i++) {
171         final WAL maybeNewWAL = wals.getWAL(Bytes.toBytes(random.nextInt()));
172         LOG.info("Iteration " + i + ", checking wal " + maybeNewWAL);
173         if (seen.add(maybeNewWAL)) {
174           count++;
175         }
176       }
177       assertEquals("received back a different number of WALs that are not equal() to each other " +
178           "than the bound we placed.", temp*4, count);
179     } finally {
180       if (wals != null) {
181         wals.close();
182       }
183       conf.setInt(NUM_REGION_GROUPS, temp);
184     }
185   }
186 }