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.hamcrest.CoreMatchers.equalTo; 021import static org.hamcrest.CoreMatchers.notNullValue; 022import static org.hamcrest.CoreMatchers.nullValue; 023import static org.hamcrest.MatcherAssert.assertThat; 024import static org.junit.Assert.assertEquals; 025import static org.junit.Assert.assertTrue; 026import static org.junit.Assert.fail; 027import static org.mockito.ArgumentMatchers.any; 028import static org.mockito.Mockito.doAnswer; 029import static org.mockito.Mockito.mock; 030import static org.mockito.Mockito.when; 031 032import java.io.ByteArrayOutputStream; 033import java.io.File; 034import java.io.PrintStream; 035import java.util.ArrayList; 036import java.util.concurrent.ThreadLocalRandom; 037import org.apache.hadoop.conf.Configuration; 038import org.apache.hadoop.fs.FileSystem; 039import org.apache.hadoop.fs.Path; 040import org.apache.hadoop.hbase.Cell; 041import org.apache.hadoop.hbase.CellUtil; 042import org.apache.hadoop.hbase.HBaseClassTestRule; 043import org.apache.hadoop.hbase.HBaseTestingUtil; 044import org.apache.hadoop.hbase.HConstants; 045import org.apache.hadoop.hbase.KeyValue; 046import org.apache.hadoop.hbase.SingleProcessHBaseCluster; 047import org.apache.hadoop.hbase.TableName; 048import org.apache.hadoop.hbase.client.Delete; 049import org.apache.hadoop.hbase.client.Get; 050import org.apache.hadoop.hbase.client.Put; 051import org.apache.hadoop.hbase.client.Result; 052import org.apache.hadoop.hbase.client.Table; 053import org.apache.hadoop.hbase.io.ImmutableBytesWritable; 054import org.apache.hadoop.hbase.mapreduce.WALPlayer.WALKeyValueMapper; 055import org.apache.hadoop.hbase.regionserver.TestRecoveredEdits; 056import org.apache.hadoop.hbase.testclassification.LargeTests; 057import org.apache.hadoop.hbase.testclassification.MapReduceTests; 058import org.apache.hadoop.hbase.tool.BulkLoadHFiles; 059import org.apache.hadoop.hbase.util.Bytes; 060import org.apache.hadoop.hbase.util.CommonFSUtils; 061import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 062import org.apache.hadoop.hbase.util.LauncherSecurityManager; 063import org.apache.hadoop.hbase.util.MapReduceExtendedCell; 064import org.apache.hadoop.hbase.wal.WAL; 065import org.apache.hadoop.hbase.wal.WALEdit; 066import org.apache.hadoop.hbase.wal.WALKey; 067import org.apache.hadoop.io.WritableComparable; 068import org.apache.hadoop.mapreduce.Mapper; 069import org.apache.hadoop.mapreduce.Mapper.Context; 070import org.apache.hadoop.util.ToolRunner; 071import org.junit.AfterClass; 072import org.junit.BeforeClass; 073import org.junit.ClassRule; 074import org.junit.Rule; 075import org.junit.Test; 076import org.junit.experimental.categories.Category; 077import org.junit.rules.TestName; 078import org.mockito.invocation.InvocationOnMock; 079import org.mockito.stubbing.Answer; 080 081/** 082 * Basic test for the WALPlayer M/R tool 083 */ 084@Category({ MapReduceTests.class, LargeTests.class }) 085public class TestWALPlayer { 086 @ClassRule 087 public static final HBaseClassTestRule CLASS_RULE = 088 HBaseClassTestRule.forClass(TestWALPlayer.class); 089 090 private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 091 private static SingleProcessHBaseCluster cluster; 092 private static Path rootDir; 093 private static Path walRootDir; 094 private static FileSystem fs; 095 private static FileSystem logFs; 096 private static Configuration conf; 097 098 @Rule 099 public TestName name = new TestName(); 100 101 @BeforeClass 102 public static void beforeClass() throws Exception { 103 conf = TEST_UTIL.getConfiguration(); 104 rootDir = TEST_UTIL.createRootDir(); 105 walRootDir = TEST_UTIL.createWALRootDir(); 106 fs = CommonFSUtils.getRootDirFileSystem(conf); 107 logFs = CommonFSUtils.getWALFileSystem(conf); 108 cluster = TEST_UTIL.startMiniCluster(); 109 } 110 111 @AfterClass 112 public static void afterClass() throws Exception { 113 TEST_UTIL.shutdownMiniCluster(); 114 fs.delete(rootDir, true); 115 logFs.delete(walRootDir, true); 116 } 117 118 /** 119 * Test that WALPlayer can replay recovered.edits files. 120 */ 121 @Test 122 public void testPlayingRecoveredEdit() throws Exception { 123 TableName tn = TableName.valueOf(TestRecoveredEdits.RECOVEREDEDITS_TABLENAME); 124 TEST_UTIL.createTable(tn, TestRecoveredEdits.RECOVEREDEDITS_COLUMNFAMILY); 125 // Copy testing recovered.edits file that is over under hbase-server test resources 126 // up into a dir in our little hdfs cluster here. 127 runWithDiskBasedSortingDisabledAndEnabled(() -> { 128 String hbaseServerTestResourcesEdits = 129 System.getProperty("test.build.classes") + "/../../../hbase-server/src/test/resources/" 130 + TestRecoveredEdits.RECOVEREDEDITS_PATH.getName(); 131 assertTrue(new File(hbaseServerTestResourcesEdits).exists()); 132 FileSystem dfs = TEST_UTIL.getDFSCluster().getFileSystem(); 133 // Target dir. 134 Path targetDir = new Path("edits").makeQualified(dfs.getUri(), dfs.getHomeDirectory()); 135 assertTrue(dfs.mkdirs(targetDir)); 136 dfs.copyFromLocalFile(new Path(hbaseServerTestResourcesEdits), targetDir); 137 assertEquals(0, 138 ToolRunner.run(new WALPlayer(this.conf), new String[] { targetDir.toString() })); 139 // I don't know how many edits are in this file for this table... so just check more than 1. 140 assertTrue(TEST_UTIL.countRows(tn) > 0); 141 dfs.delete(targetDir, true); 142 }); 143 } 144 145 /** 146 * Tests that when you write multiple cells with the same timestamp they are properly sorted by 147 * their sequenceId in WALPlayer/CellSortReducer so that the correct one wins when querying from 148 * the resulting bulkloaded HFiles. See HBASE-27649 149 */ 150 @Test 151 public void testWALPlayerBulkLoadWithOverriddenTimestamps() throws Exception { 152 final TableName tableName = TableName.valueOf(name.getMethodName() + "1"); 153 final byte[] family = Bytes.toBytes("family"); 154 final byte[] column1 = Bytes.toBytes("c1"); 155 final byte[] column2 = Bytes.toBytes("c2"); 156 final byte[] row = Bytes.toBytes("row"); 157 final Table table = TEST_UTIL.createTable(tableName, family); 158 159 long now = EnvironmentEdgeManager.currentTime(); 160 // put a row into the first table 161 Put p = new Put(row); 162 p.addColumn(family, column1, now, column1); 163 p.addColumn(family, column2, now, column2); 164 165 table.put(p); 166 167 byte[] lastVal = null; 168 169 for (int i = 0; i < 50; i++) { 170 lastVal = Bytes.toBytes(ThreadLocalRandom.current().nextLong()); 171 p = new Put(row); 172 p.addColumn(family, column1, now, lastVal); 173 174 table.put(p); 175 176 // wal rolling is necessary to trigger the bug. otherwise no sorting 177 // needs to occur in the reducer because it's all sorted and coming from a single file. 178 if (i % 10 == 0) { 179 WAL log = cluster.getRegionServer(0).getWAL(null); 180 log.rollWriter(); 181 } 182 } 183 184 WAL log = cluster.getRegionServer(0).getWAL(null); 185 log.rollWriter(); 186 String walInputDir = new Path(cluster.getMaster().getMasterFileSystem().getWALRootDir(), 187 HConstants.HREGION_LOGDIR_NAME).toString(); 188 189 Configuration configuration = new Configuration(TEST_UTIL.getConfiguration()); 190 String outPath = "/tmp/" + name.getMethodName(); 191 configuration.set(WALPlayer.BULK_OUTPUT_CONF_KEY, outPath); 192 configuration.setBoolean(WALPlayer.MULTI_TABLES_SUPPORT, true); 193 194 WALPlayer player = new WALPlayer(configuration); 195 final byte[] finalLastVal = lastVal; 196 197 runWithDiskBasedSortingDisabledAndEnabled(() -> { 198 assertEquals(0, ToolRunner.run(configuration, player, 199 new String[] { walInputDir, tableName.getNameAsString() })); 200 201 Get g = new Get(row); 202 Result result = table.get(g); 203 byte[] value = CellUtil.cloneValue(result.getColumnLatestCell(family, column1)); 204 assertThat(Bytes.toStringBinary(value), equalTo(Bytes.toStringBinary(finalLastVal))); 205 206 TEST_UTIL.truncateTable(tableName); 207 g = new Get(row); 208 result = table.get(g); 209 assertThat(result.listCells(), nullValue()); 210 211 BulkLoadHFiles.create(configuration).bulkLoad(tableName, 212 new Path(outPath, tableName.getNamespaceAsString() + "/" + tableName.getNameAsString())); 213 214 g = new Get(row); 215 result = table.get(g); 216 value = CellUtil.cloneValue(result.getColumnLatestCell(family, column1)); 217 218 assertThat(result.listCells(), notNullValue()); 219 assertThat(Bytes.toStringBinary(value), equalTo(Bytes.toStringBinary(finalLastVal))); 220 221 // cleanup 222 Path out = new Path(outPath); 223 FileSystem fs = out.getFileSystem(configuration); 224 assertTrue(fs.delete(out, true)); 225 }); 226 } 227 228 /** 229 * Simple end-to-end test 230 */ 231 @Test 232 public void testWALPlayer() throws Exception { 233 final TableName tableName1 = TableName.valueOf(name.getMethodName() + "1"); 234 final TableName tableName2 = TableName.valueOf(name.getMethodName() + "2"); 235 final byte[] FAMILY = Bytes.toBytes("family"); 236 final byte[] COLUMN1 = Bytes.toBytes("c1"); 237 final byte[] COLUMN2 = Bytes.toBytes("c2"); 238 final byte[] ROW = Bytes.toBytes("row"); 239 Table t1 = TEST_UTIL.createTable(tableName1, FAMILY); 240 Table t2 = TEST_UTIL.createTable(tableName2, FAMILY); 241 242 // put a row into the first table 243 Put p = new Put(ROW); 244 p.addColumn(FAMILY, COLUMN1, COLUMN1); 245 p.addColumn(FAMILY, COLUMN2, COLUMN2); 246 t1.put(p); 247 // delete one column 248 Delete d = new Delete(ROW); 249 d.addColumns(FAMILY, COLUMN1); 250 t1.delete(d); 251 252 // replay the WAL, map table 1 to table 2 253 WAL log = cluster.getRegionServer(0).getWAL(null); 254 log.rollWriter(); 255 String walInputDir = new Path(cluster.getMaster().getMasterFileSystem().getWALRootDir(), 256 HConstants.HREGION_LOGDIR_NAME).toString(); 257 258 Configuration configuration = TEST_UTIL.getConfiguration(); 259 WALPlayer player = new WALPlayer(configuration); 260 261 runWithDiskBasedSortingDisabledAndEnabled(() -> { 262 String optionName = "_test_.name"; 263 configuration.set(optionName, "1000"); 264 player.setupTime(configuration, optionName); 265 assertEquals(1000, configuration.getLong(optionName, 0)); 266 assertEquals(0, ToolRunner.run(configuration, player, 267 new String[] { walInputDir, tableName1.getNameAsString(), tableName2.getNameAsString() })); 268 269 // verify the WAL was player into table 2 270 Get g = new Get(ROW); 271 Result r = t2.get(g); 272 assertEquals(1, r.size()); 273 assertTrue(CellUtil.matchingQualifier(r.rawCells()[0], COLUMN2)); 274 }); 275 } 276 277 /** 278 * Test WALKeyValueMapper setup and map 279 */ 280 @Test 281 public void testWALKeyValueMapper() throws Exception { 282 testWALKeyValueMapper(WALPlayer.TABLES_KEY); 283 } 284 285 @Test 286 public void testWALKeyValueMapperWithDeprecatedConfig() throws Exception { 287 testWALKeyValueMapper("hlog.input.tables"); 288 } 289 290 private void testWALKeyValueMapper(final String tableConfigKey) throws Exception { 291 Configuration configuration = new Configuration(); 292 configuration.set(tableConfigKey, "table"); 293 WALKeyValueMapper mapper = new WALKeyValueMapper(); 294 WALKey key = mock(WALKey.class); 295 when(key.getTableName()).thenReturn(TableName.valueOf("table")); 296 @SuppressWarnings("unchecked") 297 Mapper<WALKey, WALEdit, WritableComparable<?>, Cell>.Context context = mock(Context.class); 298 when(context.getConfiguration()).thenReturn(configuration); 299 300 WALEdit value = mock(WALEdit.class); 301 ArrayList<Cell> values = new ArrayList<>(); 302 KeyValue kv1 = new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("family"), null); 303 304 values.add(kv1); 305 when(value.getCells()).thenReturn(values); 306 mapper.setup(context); 307 308 doAnswer(new Answer<Void>() { 309 310 @Override 311 public Void answer(InvocationOnMock invocation) throws Throwable { 312 ImmutableBytesWritable writer = (ImmutableBytesWritable) invocation.getArgument(0); 313 MapReduceExtendedCell key = (MapReduceExtendedCell) invocation.getArgument(1); 314 assertEquals("row", Bytes.toString(writer.get())); 315 assertEquals("row", Bytes.toString(CellUtil.cloneRow(key))); 316 return null; 317 } 318 }).when(context).write(any(), any()); 319 320 mapper.map(key, value, context); 321 322 } 323 324 /** 325 * Test main method 326 */ 327 @Test 328 public void testMainMethod() throws Exception { 329 330 PrintStream oldPrintStream = System.err; 331 SecurityManager SECURITY_MANAGER = System.getSecurityManager(); 332 LauncherSecurityManager newSecurityManager = new LauncherSecurityManager(); 333 System.setSecurityManager(newSecurityManager); 334 ByteArrayOutputStream data = new ByteArrayOutputStream(); 335 String[] args = {}; 336 System.setErr(new PrintStream(data)); 337 try { 338 System.setErr(new PrintStream(data)); 339 try { 340 WALPlayer.main(args); 341 fail("should be SecurityException"); 342 } catch (SecurityException e) { 343 assertEquals(-1, newSecurityManager.getExitCode()); 344 assertTrue(data.toString().contains("ERROR: Wrong number of arguments:")); 345 assertTrue(data.toString() 346 .contains("Usage: WALPlayer [options] <WAL inputdir>" + " [<tables> <tableMappings>]")); 347 assertTrue(data.toString().contains("-Dwal.bulk.output=/path/for/output")); 348 } 349 350 } finally { 351 System.setErr(oldPrintStream); 352 System.setSecurityManager(SECURITY_MANAGER); 353 } 354 } 355 356 private static void runWithDiskBasedSortingDisabledAndEnabled(TestMethod method) 357 throws Exception { 358 TEST_UTIL.getConfiguration().setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY, 359 false); 360 try { 361 method.run(); 362 } finally { 363 TEST_UTIL.getConfiguration().unset(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY); 364 } 365 366 TEST_UTIL.getConfiguration().setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY, 367 true); 368 try { 369 method.run(); 370 } finally { 371 TEST_UTIL.getConfiguration().unset(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY); 372 } 373 } 374 375 private interface TestMethod { 376 void run() throws Exception; 377 } 378 379}