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.client;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertNull;
022
023import java.io.IOException;
024import java.util.ArrayList;
025import java.util.Arrays;
026import java.util.List;
027import java.util.concurrent.ForkJoinPool;
028import org.apache.hadoop.hbase.HBaseClassTestRule;
029import org.apache.hadoop.hbase.HBaseTestingUtil;
030import org.apache.hadoop.hbase.PrivateCellUtil;
031import org.apache.hadoop.hbase.TableName;
032import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
033import org.apache.hadoop.hbase.testclassification.ClientTests;
034import org.apache.hadoop.hbase.testclassification.MediumTests;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.util.Pair;
037import org.junit.AfterClass;
038import org.junit.BeforeClass;
039import org.junit.ClassRule;
040import org.junit.Test;
041import org.junit.experimental.categories.Category;
042import org.junit.runner.RunWith;
043import org.junit.runners.Parameterized;
044import org.junit.runners.Parameterized.Parameter;
045import org.junit.runners.Parameterized.Parameters;
046
047import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
048
049@RunWith(Parameterized.class)
050@Category({ MediumTests.class, ClientTests.class })
051public class TestAsyncTableScanMetrics {
052
053  @ClassRule
054  public static final HBaseClassTestRule CLASS_RULE =
055    HBaseClassTestRule.forClass(TestAsyncTableScanMetrics.class);
056
057  private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
058
059  private static final TableName TABLE_NAME = TableName.valueOf("ScanMetrics");
060
061  private static final byte[] CF = Bytes.toBytes("cf");
062
063  private static final byte[] CQ = Bytes.toBytes("cq");
064
065  private static final byte[] VALUE = Bytes.toBytes("value");
066
067  private static AsyncConnection CONN;
068
069  private static int NUM_REGIONS;
070
071  @FunctionalInterface
072  private interface ScanWithMetrics {
073    Pair<List<Result>, ScanMetrics> scan(Scan scan) throws Exception;
074  }
075
076  @Parameter(0)
077  public String methodName;
078
079  @Parameter(1)
080  public ScanWithMetrics method;
081
082  @Parameters(name = "{index}: scan={0}")
083  public static List<Object[]> params() {
084    ScanWithMetrics doScanWithRawAsyncTable = TestAsyncTableScanMetrics::doScanWithRawAsyncTable;
085    ScanWithMetrics doScanWithAsyncTableScan = TestAsyncTableScanMetrics::doScanWithAsyncTableScan;
086    ScanWithMetrics doScanWithAsyncTableScanner =
087      TestAsyncTableScanMetrics::doScanWithAsyncTableScanner;
088    return Arrays.asList(new Object[] { "doScanWithRawAsyncTable", doScanWithRawAsyncTable },
089      new Object[] { "doScanWithAsyncTableScan", doScanWithAsyncTableScan },
090      new Object[] { "doScanWithAsyncTableScanner", doScanWithAsyncTableScanner });
091  }
092
093  @BeforeClass
094  public static void setUp() throws Exception {
095    UTIL.startMiniCluster(3);
096    // Create 3 rows in the table, with rowkeys starting with "zzz*" so that
097    // scan are forced to hit all the regions.
098    try (Table table = UTIL.createMultiRegionTable(TABLE_NAME, CF)) {
099      table.put(Arrays.asList(new Put(Bytes.toBytes("zzz1")).addColumn(CF, CQ, VALUE),
100        new Put(Bytes.toBytes("zzz2")).addColumn(CF, CQ, VALUE),
101        new Put(Bytes.toBytes("zzz3")).addColumn(CF, CQ, VALUE)));
102    }
103    CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
104    NUM_REGIONS = UTIL.getHBaseCluster().getRegions(TABLE_NAME).size();
105  }
106
107  @AfterClass
108  public static void tearDown() throws Exception {
109    Closeables.close(CONN, true);
110    UTIL.shutdownMiniCluster();
111  }
112
113  private static Pair<List<Result>, ScanMetrics> doScanWithRawAsyncTable(Scan scan)
114    throws IOException, InterruptedException {
115    BufferingScanResultConsumer consumer = new BufferingScanResultConsumer();
116    CONN.getTable(TABLE_NAME).scan(scan, consumer);
117    List<Result> results = new ArrayList<>();
118    for (Result result; (result = consumer.take()) != null;) {
119      results.add(result);
120    }
121    return Pair.newPair(results, consumer.getScanMetrics());
122  }
123
124  private static Pair<List<Result>, ScanMetrics> doScanWithAsyncTableScan(Scan scan)
125    throws Exception {
126    SimpleScanResultConsumerImpl consumer = new SimpleScanResultConsumerImpl();
127    CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool()).scan(scan, consumer);
128    return Pair.newPair(consumer.getAll(), consumer.getScanMetrics());
129  }
130
131  private static Pair<List<Result>, ScanMetrics> doScanWithAsyncTableScanner(Scan scan)
132    throws IOException {
133    try (ResultScanner scanner =
134      CONN.getTable(TABLE_NAME, ForkJoinPool.commonPool()).getScanner(scan)) {
135      List<Result> results = new ArrayList<>();
136      for (Result result; (result = scanner.next()) != null;) {
137        results.add(result);
138      }
139      return Pair.newPair(results, scanner.getScanMetrics());
140    }
141  }
142
143  @Test
144  public void testNoScanMetrics() throws Exception {
145    Pair<List<Result>, ScanMetrics> pair = method.scan(new Scan());
146    assertEquals(3, pair.getFirst().size());
147    assertNull(pair.getSecond());
148  }
149
150  @Test
151  public void testScanMetrics() throws Exception {
152    Pair<List<Result>, ScanMetrics> pair = method.scan(new Scan().setScanMetricsEnabled(true));
153    List<Result> results = pair.getFirst();
154    assertEquals(3, results.size());
155    long bytes = results.stream().flatMap(r -> Arrays.asList(r.rawCells()).stream())
156      .mapToLong(c -> PrivateCellUtil.estimatedSerializedSizeOf(c)).sum();
157    ScanMetrics scanMetrics = pair.getSecond();
158    assertEquals(NUM_REGIONS, scanMetrics.countOfRegions.get());
159    assertEquals(bytes, scanMetrics.countOfBytesInResults.get());
160    assertEquals(NUM_REGIONS, scanMetrics.countOfRPCcalls.get());
161    // also assert a server side metric to ensure that we have published them into the client side
162    // metrics.
163    assertEquals(3, scanMetrics.countOfRowsScanned.get());
164  }
165}