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.wal;
019
020import static org.junit.Assert.assertEquals;
021
022import org.apache.hadoop.conf.Configuration;
023import org.apache.hadoop.fs.FileStatus;
024import org.apache.hadoop.fs.FileSystem;
025import org.apache.hadoop.fs.Path;
026import org.apache.hadoop.hbase.HBaseClassTestRule;
027import org.apache.hadoop.hbase.HBaseTestingUtil;
028import org.apache.hadoop.hbase.testclassification.MediumTests;
029import org.apache.hadoop.hbase.testclassification.RegionServerTests;
030import org.junit.AfterClass;
031import org.junit.Before;
032import org.junit.BeforeClass;
033import org.junit.ClassRule;
034import org.junit.Rule;
035import org.junit.Test;
036import org.junit.experimental.categories.Category;
037import org.junit.rules.TestName;
038import org.slf4j.Logger;
039import org.slf4j.LoggerFactory;
040
041/**
042 * Tests for TestFSHLogProvider which use WALPerformanceEvaluation for WAL data creation. This class
043 * was created as part of refactoring for hbase-diagnostics module creation in HBASE-28432 to break
044 * cyclic dependency.
045 */
046@Category({ RegionServerTests.class, MediumTests.class })
047public class TestFSHLogProviderWithConcurrentWrites {
048
049  @ClassRule
050  public static final HBaseClassTestRule CLASS_RULE =
051    HBaseClassTestRule.forClass(TestFSHLogProviderWithConcurrentWrites.class);
052
053  private static final Logger LOG =
054    LoggerFactory.getLogger(TestFSHLogProviderWithConcurrentWrites.class);
055
056  private static FileSystem fs;
057  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
058
059  @Rule
060  public final TestName currentTest = new TestName();
061
062  @Before
063  public void setUp() throws Exception {
064    FileStatus[] entries = fs.listStatus(new Path("/"));
065    for (FileStatus dir : entries) {
066      fs.delete(dir.getPath(), true);
067    }
068  }
069
070  @BeforeClass
071  public static void setUpBeforeClass() throws Exception {
072    // Make block sizes small.
073    TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
074    // quicker heartbeat interval for faster DN death notification
075    TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
076    TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
077    TEST_UTIL.getConfiguration().setInt("dfs.client.socket-timeout", 5000);
078
079    // faster failover with cluster.shutdown();fs.close() idiom
080    TEST_UTIL.getConfiguration().setInt("hbase.ipc.client.connect.max.retries", 1);
081    TEST_UTIL.getConfiguration().setInt("dfs.client.block.recovery.retries", 1);
082    TEST_UTIL.getConfiguration().setInt("hbase.ipc.client.connection.maxidletime", 500);
083    TEST_UTIL.startMiniDFSCluster(3);
084
085    // Set up a working space for our tests.
086    TEST_UTIL.createRootDir();
087    fs = TEST_UTIL.getDFSCluster().getFileSystem();
088  }
089
090  @AfterClass
091  public static void tearDownAfterClass() throws Exception {
092    TEST_UTIL.shutdownMiniCluster();
093  }
094
095  /**
096   * Write to a log file with three concurrent threads and verifying all data is written.
097   */
098  @Test
099  public void testConcurrentWrites() throws Exception {
100    // Run the WPE tool with three threads writing 3000 edits each concurrently.
101    // When done, verify that all edits were written.
102    int errCode =
103      WALPerformanceEvaluation.innerMain(new Configuration(TEST_UTIL.getConfiguration()),
104        new String[] { "-threads", "3", "-verify", "-noclosefs", "-iterations", "3000" });
105    assertEquals(0, errCode);
106  }
107}