HBase application: Unit testing by mocking the HBase

Question

I have a method in my application which is accessing data from HBase. It uses scan method to query hbase. I want to write unit test cases for testing this function. So I want to mock the hbase calls. How to do it? I am using Mockit for mocking.

Answer 1

If you're using Mockito, you can stub your classes to make them return what you want.

Let's say you had a class called HBaseHelper and a method called getData() within the class that used a scanner to retrieve data from hbase. Now let's say you have another method called useData() in another class as such:

public String useData() {
  String data = hbaseHelper.getData();

  // ... Do things with data
  return data;
}

If you are using Mockito, you can effectively do something like this inside your test to return dummy 'data' and test the method that uses this data:

import org.mockito.Mock;
import org.mockito.Mockito.when;

@Mock
HBaseHelper hbaseHelper;

@Test
public void testFoo() {
  when(hbaseHelper.getData()).thenReturn("hello world");

  assertThat(useData()).equals("hello world");
}

Answer 2

In Java you can use HBaseTestingUtility like this:

private static final HBaseTestingUtility TEST_UTIL =
  new HBaseTestingUtility();

@BeforeClass
public static void setUpBeforeClass() throws Exception {
TEST_UTIL.getConfiguration().setBoolean("hbase.table.sanity.checks", false);
TEST_UTIL.startMiniCluster();
}

@AfterClass
public static void tearDownAfterClass() throws Exception {
TEST_UTIL.shutdownMiniCluster();
EnvironmentEdgeManager.reset();
}

Additionally you may need Thrift server to use some client libs:

private static final HBaseTestingUtility TEST_UTIL =
  new HBaseTestingUtility();

ThriftServer thriftServer;
Thread thriftServerThread;

@BeforeClass
public static void setUpBeforeClass() throws Exception {
TEST_UTIL.getConfiguration().setBoolean("hbase.table.sanity.checks", false);
TEST_UTIL.startMiniCluster();

List<String> args = new ArrayList<>();
port = HBaseTestingUtility.randomFreePort();
args.add("-" + ThriftServer.PORT_OPTION);
args.add(String.valueOf(port));
args.add("-infoport");
int infoPort = HBaseTestingUtility.randomFreePort();
args.add(String.valueOf(infoPort));
args.add("start");

thriftServer = new ThriftServer(TEST_UTIL.getConfiguration());

thriftServerThread = new Thread(new Runnable() {
  @Override
  public void run() {
    thriftServer.doMain(args.toArray(new String[args.size()]));
  }
});
thriftServerThread.setDaemon(true)
thriftServerThread.start();
}

@AfterClass
public static void tearDownAfterClass() throws Exception {
TEST_UTIL.shutdownMiniCluster();
EnvironmentEdgeManager.reset();
}

With pyspark through py4j this way:

def setUp(self):
    super(StreamingTest, self).setUp()

    # --- hbase configuration ---
    hbase_testing_utility_clz = self.sparkStreamingContext._jvm.java.lang.Thread.currentThread().getContextClassLoader() \
        .loadClass('org.apache.hadoop.hbase.HBaseTestingUtility')
    self._hbaseTestingUtility = hbase_testing_utility_clz.newInstance()

    self._hbaseTestingUtility.startMiniCluster()


def tearDown(self):
    if self._hbaseTestingUtility is not None:
        self._hbaseTestingUtility.shutdownMiniCluster()

and if Thrift server is needed (for example to use happybase client lib):

def setUp(self):
    super(StreamingTest, self).setUp()

    # --- hbase configuration ---
    hbase_testing_utility_clz = self.sparkStreamingContext._jvm.java.lang.Thread.currentThread().getContextClassLoader() \
        .loadClass('org.apache.hadoop.hbase.HBaseTestingUtility')
    self._hbaseTestingUtility = hbase_testing_utility_clz.newInstance()
    self._hbaseTestingUtility.getConfiguration().setBoolean("hbase.table.sanity.checks", False)  # for thrift
    self._hbaseTestingUtility.startMiniCluster()

    # --- thrift server configuration ---
    thrift_server_clz = self.sparkStreamingContext._jvm.java.lang.Thread.currentThread().getContextClassLoader() \
        .loadClass('org.apache.hadoop.hbase.thrift.ThriftServer')

    # make thrift server instance
    cArgs = self.sparkStreamingContext.sparkContext._gateway.new_array(self.sparkStreamingContext._jvm.java.lang.Class, 1)
    cArgs[0] = self._hbaseTestingUtility.getConfiguration().getClass()
    iArgs = self.sparkStreamingContext.sparkContext._gateway.new_array(self.sparkStreamingContext._jvm.java.lang.Object, 1)
    iArgs[0] = self._hbaseTestingUtility.getConfiguration()

    self._thriftServer = thrift_server_clz\
        .getDeclaredConstructor(cArgs)\
        .newInstance(iArgs)

    # prepare server start arguments
    tArgs = self.sparkStreamingContext.sparkContext._gateway.new_array(self.sparkStreamingContext._jvm.java.lang.String, 5)
    port = self._hbaseTestingUtility.randomFreePort()
    self.thrift_port = port
    tArgs[0] = "-port"
    tArgs[1] = str(port)
    tArgs[2] = "-infoport"
    info_port = self._hbaseTestingUtility.randomFreePort()
    tArgs[3] = str(info_port)
    tArgs[4] = "start"

    mArgs = self.sparkStreamingContext.sparkContext._gateway.new_array(self.sparkStreamingContext._jvm.java.lang.Class, 1)
    mArgs[0] = tArgs.getClass()
    method = thrift_server_clz.getDeclaredMethod('doMain', mArgs)
    method.setAccessible(True)

    args = self.sparkStreamingContext.sparkContext._gateway.new_array(self.sparkStreamingContext._jvm.java.lang.Object, 1)

    # start server in separate thread
    args[0] = tArgs
    self.thrift_server_thread = threading.Thread(target=method.invoke, args=[self._thriftServer, args])
    self.thrift_server_thread.setDaemon(True)
    self.thrift_server_thread.start()

of course hbase and thrift jars should be passed trough spark submit:

--jars path/to/jar1.jar,path/to/jar2.jar, --conf spark.driver.userClassPathFirst=true