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.mapreduce; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertTrue; 022import static org.junit.Assert.fail; 023import static org.mockito.ArgumentMatchers.any; 024import static org.mockito.Mockito.doAnswer; 025import static org.mockito.Mockito.mock; 026import static org.mockito.Mockito.when; 027import java.io.ByteArrayOutputStream; 028import java.io.File; 029import java.io.PrintStream; 030import java.util.ArrayList; 031import org.apache.hadoop.conf.Configuration; 032import org.apache.hadoop.fs.FileSystem; 033import org.apache.hadoop.fs.Path; 034import org.apache.hadoop.hbase.Cell; 035import org.apache.hadoop.hbase.CellUtil; 036import org.apache.hadoop.hbase.HBaseClassTestRule; 037import org.apache.hadoop.hbase.HBaseTestingUtility; 038import org.apache.hadoop.hbase.HConstants; 039import org.apache.hadoop.hbase.KeyValue; 040import org.apache.hadoop.hbase.MiniHBaseCluster; 041import org.apache.hadoop.hbase.TableName; 042import org.apache.hadoop.hbase.client.Delete; 043import org.apache.hadoop.hbase.client.Get; 044import org.apache.hadoop.hbase.client.Put; 045import org.apache.hadoop.hbase.client.Result; 046import org.apache.hadoop.hbase.client.Table; 047import org.apache.hadoop.hbase.io.ImmutableBytesWritable; 048import org.apache.hadoop.hbase.mapreduce.WALPlayer.WALKeyValueMapper; 049import org.apache.hadoop.hbase.regionserver.TestRecoveredEdits; 050import org.apache.hadoop.hbase.testclassification.LargeTests; 051import org.apache.hadoop.hbase.testclassification.MapReduceTests; 052import org.apache.hadoop.hbase.util.Bytes; 053import org.apache.hadoop.hbase.util.CommonFSUtils; 054import org.apache.hadoop.hbase.util.LauncherSecurityManager; 055import org.apache.hadoop.hbase.wal.WAL; 056import org.apache.hadoop.hbase.wal.WALEdit; 057import org.apache.hadoop.hbase.wal.WALKey; 058import org.apache.hadoop.mapreduce.Mapper; 059import org.apache.hadoop.mapreduce.Mapper.Context; 060import org.apache.hadoop.util.ToolRunner; 061import org.junit.AfterClass; 062import org.junit.BeforeClass; 063import org.junit.ClassRule; 064import org.junit.Rule; 065import org.junit.Test; 066import org.junit.experimental.categories.Category; 067import org.junit.rules.TestName; 068import org.mockito.invocation.InvocationOnMock; 069import org.mockito.stubbing.Answer; 070 071/** 072 * Basic test for the WALPlayer M/R tool 073 */ 074@Category({MapReduceTests.class, LargeTests.class}) 075//TODO : Remove this in 3.0 076public class TestWALPlayer { 077 @ClassRule 078 public static final HBaseClassTestRule CLASS_RULE = 079 HBaseClassTestRule.forClass(TestWALPlayer.class); 080 081 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); 082 private static MiniHBaseCluster cluster; 083 private static Path rootDir; 084 private static Path walRootDir; 085 private static FileSystem fs; 086 private static FileSystem logFs; 087 private static Configuration conf; 088 089 @Rule 090 public TestName name = new TestName(); 091 092 @BeforeClass 093 public static void beforeClass() throws Exception { 094 conf = TEST_UTIL.getConfiguration(); 095 rootDir = TEST_UTIL.createRootDir(); 096 walRootDir = TEST_UTIL.createWALRootDir(); 097 fs = CommonFSUtils.getRootDirFileSystem(conf); 098 logFs = CommonFSUtils.getWALFileSystem(conf); 099 cluster = TEST_UTIL.startMiniCluster(); 100 } 101 102 @AfterClass 103 public static void afterClass() throws Exception { 104 TEST_UTIL.shutdownMiniCluster(); 105 fs.delete(rootDir, true); 106 logFs.delete(walRootDir, true); 107 } 108 109 /** 110 * Test that WALPlayer can replay recovered.edits files. 111 */ 112 @Test 113 public void testPlayingRecoveredEdit() throws Exception { 114 TableName tn = TableName.valueOf(TestRecoveredEdits.RECOVEREDEDITS_TABLENAME); 115 TEST_UTIL.createTable(tn, TestRecoveredEdits.RECOVEREDEDITS_COLUMNFAMILY); 116 // Copy testing recovered.edits file that is over under hbase-server test resources 117 // up into a dir in our little hdfs cluster here. 118 String hbaseServerTestResourcesEdits = System.getProperty("test.build.classes") + 119 "/../../../hbase-server/src/test/resources/" + 120 TestRecoveredEdits.RECOVEREDEDITS_PATH.getName(); 121 assertTrue(new File(hbaseServerTestResourcesEdits).exists()); 122 FileSystem dfs = TEST_UTIL.getDFSCluster().getFileSystem(); 123 // Target dir. 124 Path targetDir = new Path("edits").makeQualified(dfs.getUri(), dfs.getHomeDirectory()); 125 assertTrue(dfs.mkdirs(targetDir)); 126 dfs.copyFromLocalFile(new Path(hbaseServerTestResourcesEdits), targetDir); 127 assertEquals(0, 128 ToolRunner.run(new WALPlayer(this.conf), new String [] {targetDir.toString()})); 129 // I don't know how many edits are in this file for this table... so just check more than 1. 130 assertTrue(TEST_UTIL.countRows(tn) > 0); 131 } 132 133 /** 134 * Simple end-to-end test 135 */ 136 @Test 137 public void testWALPlayer() throws Exception { 138 final TableName tableName1 = TableName.valueOf(name.getMethodName() + "1"); 139 final TableName tableName2 = TableName.valueOf(name.getMethodName() + "2"); 140 final byte[] FAMILY = Bytes.toBytes("family"); 141 final byte[] COLUMN1 = Bytes.toBytes("c1"); 142 final byte[] COLUMN2 = Bytes.toBytes("c2"); 143 final byte[] ROW = Bytes.toBytes("row"); 144 Table t1 = TEST_UTIL.createTable(tableName1, FAMILY); 145 Table t2 = TEST_UTIL.createTable(tableName2, FAMILY); 146 147 // put a row into the first table 148 Put p = new Put(ROW); 149 p.addColumn(FAMILY, COLUMN1, COLUMN1); 150 p.addColumn(FAMILY, COLUMN2, COLUMN2); 151 t1.put(p); 152 // delete one column 153 Delete d = new Delete(ROW); 154 d.addColumns(FAMILY, COLUMN1); 155 t1.delete(d); 156 157 // replay the WAL, map table 1 to table 2 158 WAL log = cluster.getRegionServer(0).getWAL(null); 159 log.rollWriter(); 160 String walInputDir = new Path(cluster.getMaster().getMasterFileSystem() 161 .getWALRootDir(), HConstants.HREGION_LOGDIR_NAME).toString(); 162 163 Configuration configuration= TEST_UTIL.getConfiguration(); 164 WALPlayer player = new WALPlayer(configuration); 165 String optionName="_test_.name"; 166 configuration.set(optionName, "1000"); 167 player.setupTime(configuration, optionName); 168 assertEquals(1000,configuration.getLong(optionName,0)); 169 assertEquals(0, ToolRunner.run(configuration, player, 170 new String[] {walInputDir, tableName1.getNameAsString(), 171 tableName2.getNameAsString() })); 172 173 174 // verify the WAL was player into table 2 175 Get g = new Get(ROW); 176 Result r = t2.get(g); 177 assertEquals(1, r.size()); 178 assertTrue(CellUtil.matchingQualifier(r.rawCells()[0], COLUMN2)); 179 } 180 181 /** 182 * Test WALKeyValueMapper setup and map 183 */ 184 @Test 185 public void testWALKeyValueMapper() throws Exception { 186 testWALKeyValueMapper(WALPlayer.TABLES_KEY); 187 } 188 189 @Test 190 public void testWALKeyValueMapperWithDeprecatedConfig() throws Exception { 191 testWALKeyValueMapper("hlog.input.tables"); 192 } 193 194 private void testWALKeyValueMapper(final String tableConfigKey) throws Exception { 195 Configuration configuration = new Configuration(); 196 configuration.set(tableConfigKey, "table"); 197 WALKeyValueMapper mapper = new WALKeyValueMapper(); 198 WALKey key = mock(WALKey.class); 199 when(key.getTableName()).thenReturn(TableName.valueOf("table")); 200 @SuppressWarnings("unchecked") 201 Mapper<WALKey, WALEdit, ImmutableBytesWritable, KeyValue>.Context context = mock(Context.class); 202 when(context.getConfiguration()).thenReturn(configuration); 203 204 WALEdit value = mock(WALEdit.class); 205 ArrayList<Cell> values = new ArrayList<>(); 206 KeyValue kv1 = new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("family"), null); 207 208 values.add(kv1); 209 when(value.getCells()).thenReturn(values); 210 mapper.setup(context); 211 212 doAnswer(new Answer<Void>() { 213 214 @Override 215 public Void answer(InvocationOnMock invocation) throws Throwable { 216 ImmutableBytesWritable writer = invocation.getArgument(0); 217 KeyValue key = invocation.getArgument(1); 218 assertEquals("row", Bytes.toString(writer.get())); 219 assertEquals("row", Bytes.toString(CellUtil.cloneRow(key))); 220 return null; 221 } 222 }).when(context).write(any(), any()); 223 224 mapper.map(key, value, context); 225 226 } 227 228 /** 229 * Test main method 230 */ 231 @Test 232 public void testMainMethod() throws Exception { 233 234 PrintStream oldPrintStream = System.err; 235 SecurityManager SECURITY_MANAGER = System.getSecurityManager(); 236 LauncherSecurityManager newSecurityManager= new LauncherSecurityManager(); 237 System.setSecurityManager(newSecurityManager); 238 ByteArrayOutputStream data = new ByteArrayOutputStream(); 239 String[] args = {}; 240 System.setErr(new PrintStream(data)); 241 try { 242 System.setErr(new PrintStream(data)); 243 try { 244 WALPlayer.main(args); 245 fail("should be SecurityException"); 246 } catch (SecurityException e) { 247 assertEquals(-1, newSecurityManager.getExitCode()); 248 assertTrue(data.toString().contains("ERROR: Wrong number of arguments:")); 249 assertTrue(data.toString().contains("Usage: WALPlayer [options] <WAL inputdir>" + 250 " [<tables> <tableMappings>]")); 251 assertTrue(data.toString().contains("-Dwal.bulk.output=/path/for/output")); 252 } 253 254 } finally { 255 System.setErr(oldPrintStream); 256 System.setSecurityManager(SECURITY_MANAGER); 257 } 258 259 } 260 261}