org.apache.hadoop.hbase.client.Result#create ( )源码实例Demo

/**
 * Makes a HBase Result object based on the KeyValue's from the SEP event. Usually, this will only be used in
 * situations where only new data is written (or updates are complete row updates), so we don't expect any
 * delete-type key-values, but just to be sure we filter them out.
 */
@Override
public Result toResult() {

    List<Cell> filteredKeyValues = Lists.newArrayListWithCapacity(sepEvent.getKeyValues().size());

    for (Cell kv : getKeyValues()) {
        if (!CellUtil.isDelete(kv) && !CellUtil.isDeleteFamily(kv)) {
            filteredKeyValues.add(kv);
        }
    }

    // A Result object requires that the KeyValues are sorted (e.g., it does binary search on them)
    Collections.sort(filteredKeyValues, KeyValue.COMPARATOR);
    return Result.create(filteredKeyValues);
}
 

@Test
public void testReportTablesMissingRegionsOneMissing() throws  Exception {
  ResultScanner mockedRS = Mockito.mock(ResultScanner.class);
  when(this.mockedTable.getScanner(any(Scan.class))).thenReturn(mockedRS);
  List<Cell> cells = new ArrayList<>();
  cells.add(createCellForTableState(TableName.valueOf("test-tbl")));
  Result result = Result.create(cells);
  when(mockedRS.next()).thenReturn(result,(Result)null);
  FileStatus status = new FileStatus();
  Path p = new Path(this.testTblDir+ "/182182182121");
  status.setPath(p);
  when(mockedFileSystem.listStatus(new Path(this.testTblDir)))
    .thenReturn(new FileStatus[]{status});
  Map<TableName, List<Path>> report = fixer.reportTablesMissingRegions(null);
  assertEquals("Should had returned 1 missing region",
    1,report.size());
}
 

@Test
public void testMap() throws Exception {
    MorphlineResultToSolrMapper resultMapper = new MorphlineResultToSolrMapper();
    resultMapper.configure(ImmutableMap.of(
        MorphlineResultToSolrMapper.MORPHLINE_FILE_PARAM, "src/test/resources/test-morphlines/extractHBaseCells.conf")
        );

    Cell kvA = new KeyValue(ROW, COLUMN_FAMILY_A, QUALIFIER_A, Bytes.toBytes(42));
    Cell kvB = new KeyValue(ROW, COLUMN_FAMILY_B, QUALIFIER_B, "dummy value".getBytes("UTF-8"));
    Result result = Result.create(Lists.newArrayList(kvA, kvB));

    Multimap expectedMap = ImmutableMultimap.of("fieldA", 42, "fieldB", "dummy value");

    resultMapper.map(result, updateWriter);
    verify(updateWriter).add(solrInputDocCaptor.capture());
    
    SolrInputDocument solrDocument = solrInputDocCaptor.getValue();
    assertEquals(expectedMap, toRecord(solrDocument).getFields());
}
 

@Test
public void testMapWithMultipleOutputFields() throws Exception {
    MorphlineResultToSolrMapper resultMapper = new MorphlineResultToSolrMapper();
    resultMapper.configure(ImmutableMap.of(
            MorphlineResultToSolrMapper.MORPHLINE_FILE_PARAM, "src/test/resources/test-morphlines/extractHBaseCellsWithMultipleOutputFields.conf"));

    KeyValue kvA = new KeyValue(ROW, COLUMN_FAMILY_A, QUALIFIER_A, Bytes.toBytes(42));
    KeyValue kvX = new KeyValue(ROW, COLUMN_FAMILY_B, QUALIFIER_A, "Basti".getBytes("UTF-8"));
    KeyValue kvB = new KeyValue(ROW, COLUMN_FAMILY_B, QUALIFIER_B, "dummy value".getBytes("UTF-8"));
    KeyValue kvC = new KeyValue(ROW, COLUMN_FAMILY_B, QUALIFIER_C, "Nadja".getBytes("UTF-8"));
    Result result = Result.create(new Cell[] {kvA, kvX, kvB, kvC});

    Multimap expectedMap = ImmutableMultimap.of("fieldA", 42, "fieldB", "Basti", "fieldC", "Nadja");

    resultMapper.map(result, updateWriter);
    verify(updateWriter).add(solrInputDocCaptor.capture());
    
    SolrInputDocument solrDocument = solrInputDocCaptor.getValue();
    assertEquals(expectedMap, toRecord(solrDocument).getFields());
}
 

private RegionInfo createRegionInMeta(ResultScanner mockedRS) throws Exception {
  when(this.mockedTable.getScanner(any(Scan.class))).thenReturn(mockedRS);
  RegionInfo info = createRegionInfo("test-tbl");
  List<Cell> cells = new ArrayList<>();
  cells.add(createCellForRegionInfo(info));
  Result result = Result.create(cells);
  when(mockedRS.next()).thenReturn(result,(Result)null);
  return info;
}
 

@Override
public Result next() throws IOException {
  values.clear();
  scanner.nextRaw(values);
  statisticsCollector.collectStatistics(values);
  if (values.isEmpty()) {
    //we are done
    return null;
  }

  return Result.create(values);
}
 

private void ensureNoUpdatesWhenCoveredByDelete(RegionCoprocessorEnvironment env, IndexCodec codec, List<Cell> currentState,
    Delete d) throws IOException {
  LocalHBaseState table = new SimpleTableState(Result.create(currentState));
  LocalTableState state = new LocalTableState(table, d);
  state.setCurrentTimestamp(d.getTimeStamp());
  // now we shouldn't see anything when getting the index update
  state.addPendingUpdates(d.getFamilyCellMap().get(FAMILY));
  Iterable<IndexUpdate> updates = codec.getIndexUpserts(state, IndexMetaData.NULL_INDEX_META_DATA, null, null);
  for (IndexUpdate update : updates) {
    assertFalse("Had some index updates, though it should have been covered by the delete",
      update.isValid());
  }
}
 

@Test
public void testGetApproxSubmitTime()
    throws IOException, MissingColumnInResultException {
  JobHistoryRawService rawService = new JobHistoryRawService(hbaseConnection);
  Cell[] cells = new Cell[1];
  long modts = 1396550668000L;
  cells[0] = CellUtil.createCell(Bytes.toBytes("someRowKey"),
      Constants.INFO_FAM_BYTES, Constants.JOBHISTORY_LAST_MODIFIED_COL_BYTES,
      HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put.getCode(),
      Bytes.toBytes(modts));
  Result result = Result.create(cells);
  long st = rawService.getApproxSubmitTime(result);
  long expts = modts - Constants.AVERGAE_JOB_DURATION;
  assertEquals(expts, st);
}
 

/**
 * Convert a protocol buffer Result to a client Result
 *
 * @param proto the protocol buffer Result to convert
 * @param scanner Optional cell scanner.
 * @return the converted client Result
 * @throws IOException
 */
public static Result toResult(final ClientProtos.Result proto, final CellScanner scanner)
throws IOException {
  List<CellProtos.Cell> values = proto.getCellList();

  if (proto.hasExists()) {
    if ((values != null && !values.isEmpty()) ||
        (proto.hasAssociatedCellCount() && proto.getAssociatedCellCount() > 0)) {
      throw new IllegalArgumentException("bad proto: exists with cells is no allowed " + proto);
    }
    if (proto.getStale()) {
      return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE_STALE :EMPTY_RESULT_EXISTS_FALSE_STALE;
    }
    return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE : EMPTY_RESULT_EXISTS_FALSE;
  }

  // TODO: Unit test that has some Cells in scanner and some in the proto.
  List<Cell> cells = null;
  if (proto.hasAssociatedCellCount()) {
    int count = proto.getAssociatedCellCount();
    cells = new ArrayList<>(count + values.size());
    for (int i = 0; i < count; i++) {
      if (!scanner.advance()) throw new IOException("Failed get " + i + " of " + count);
      cells.add(scanner.current());
    }
  }

  if (!values.isEmpty()){
    if (cells == null) cells = new ArrayList<>(values.size());
    ExtendedCellBuilder builder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
    for (CellProtos.Cell c: values) {
      cells.add(toCell(builder, c));
    }
  }

  return (cells == null || cells.isEmpty())
      ? (proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT)
      : Result.create(cells, null, proto.getStale());
}
 

@Test
public void testExtract_IncludeAll() {
    Result result = Result.create(new Cell[]{
        new KeyValue(ROW, COLFAM, Bytes.toBytes("ABB"), Bytes.toBytes("value ABB")),
        new KeyValue(ROW, COLFAM, Bytes.toBytes("ABC"), Bytes.toBytes("value ABC"))
    });
    
    assertExtractEquals(Lists.newArrayList("ABB:value ABB", "ABC:value ABC"), extractor.extract(result));
}
 

@Before
public void setUp() {
    KeyValue kvA1 = new KeyValue(ROW, COLFAM_A, QUALIFIER_A1, VALUE_A1);
    KeyValue kvA2 = new KeyValue(ROW, COLFAM_A, QUALIFIER_A2, VALUE_A2);
    KeyValue kvB1 = new KeyValue(ROW, COLFAM_B, QUALIFIER_B1, VALUE_B1);

    result = Result.create(Lists.<Cell>newArrayList(kvA1, kvA2, kvB1));
}
 

public static Result toResult(ImmutableBytesWritable bytes) {
    byte [] buf = bytes.get();
    int offset = bytes.getOffset();
    int finalOffset = bytes.getSize() + offset;
    List<Cell> kvs = new ArrayList<Cell>();
    while(offset < finalOffset) {
      int keyLength = Bytes.toInt(buf, offset);
      offset += Bytes.SIZEOF_INT;
      kvs.add(new KeyValue(buf, offset, keyLength));
      offset += keyLength;
    }
    return Result.create(kvs);
}
 

@Test
public void mapperTest() throws Exception {
    // Given
    List<TestAgentStat> givenAgentStats = new ArrayList<>();
    List<Put> puts = new ArrayList<>();
    long initialTimestamp = System.currentTimeMillis();
    int numBatch = RandomUtils.nextInt(1, MAX_NUM_TEST_VALUES);
    for (int i = 0; i < numBatch; i++) {
        int batchSize = RandomUtils.nextInt(1, MAX_NUM_TEST_VALUES);
        List<TestAgentStat> agentStatBatch = createAgentStats(initialTimestamp, COLLECT_INVERVAL, batchSize);
        givenAgentStats.addAll(agentStatBatch);
        puts.addAll(this.hbaseOperationFactory.createPuts(AGENT_ID, AGENT_STAT_TYPE, agentStatBatch, this.serializer));
        initialTimestamp += batchSize * COLLECT_INVERVAL;
    }
    List<Cell> cellsToPut = new ArrayList<>();
    for (Put put : puts) {
        List<Cell> cells = put.getFamilyCellMap().get(HbaseColumnFamily.AGENT_STAT_STATISTICS.getName());
        cellsToPut.addAll(cells);
    }
    Result result = Result.create(cellsToPut);

    // When
    AgentStatMapperV2<TestAgentStat> mapper = new AgentStatMapperV2<>(this.hbaseOperationFactory, this.decoder, TEST_FILTER);
    List<TestAgentStat> mappedAgentStats = mapper.mapRow(result, 0);

    // Then
    givenAgentStats.sort(AgentStatMapperV2.REVERSE_TIMESTAMP_COMPARATOR);
    Assert.assertEquals(givenAgentStats, mappedAgentStats);
}
 

@Override
public Result next() throws IOException {
    List<Cell> cells = Lists.newArrayList();
    scanner.next(cells);
    if (cells.isEmpty())
        return null;
    else
        return Result.create(cells);
}
 

@Before
public void setUp() {
    KeyValue kvA1 = new KeyValue(ROW, COLFAM_A, QUALIFIER_A1, VALUE_A1);
    KeyValue kvA2 = new KeyValue(ROW, COLFAM_A, QUALIFIER_A2, VALUE_A2);
    KeyValue kvB1 = new KeyValue(ROW, COLFAM_B, QUALIFIER_B1, VALUE_B1);

    result = Result.create(Lists.<Cell>newArrayList(kvA1, kvA2, kvB1));
}
 

@Test
public void testRecovery() throws IOException {
  int countPerRound = 100;
  for (int round = 0; round < 5; round++) {
    for (int i = 0; i < countPerRound; i++) {
      int row = round * countPerRound + i;
      Put put = new Put(Bytes.toBytes(row)).addColumn(CF, CQ, Bytes.toBytes(row));
      region.update(r -> r.put(put));
    }
    region.close(true);
    region = createMasterRegion(
      ServerName.valueOf("localhost", 12345, System.currentTimeMillis() + round + 1));
    try (RegionScanner scanner = region.getScanner(new Scan())) {
      List<Cell> cells = new ArrayList<>();
      boolean moreValues = true;
      for (int i = 0; i < (round + 1) * countPerRound; i++) {
        assertTrue(moreValues);
        moreValues = scanner.next(cells);
        assertEquals(1, cells.size());
        Result result = Result.create(cells);
        cells.clear();
        assertEquals(i, Bytes.toInt(result.getRow()));
        assertEquals(i, Bytes.toInt(result.getValue(CF, CQ)));
      }
      assertFalse(moreValues);
    }
  }
}
 

@Override
protected void calculateIndexUpdates(List<RowData> rowDataList, SolrUpdateCollector updateCollector) throws IOException {
    Map<String, KeyValue> idToKeyValue = calculateUniqueEvents(rowDataList);
    for (Entry<String, KeyValue> idToKvEntry : idToKeyValue.entrySet()) {
        String documentId = idToKvEntry.getKey();

        KeyValue keyValue = idToKvEntry.getValue();
        if (CellUtil.isDelete(keyValue)) {
            handleDelete(documentId, keyValue, updateCollector, uniqueKeyFormatter);
        } else {
            Result result = Result.create(Collections.<Cell>singletonList(keyValue));
            SolrUpdateWriter updateWriter = new RowAndFamilyAddingSolrUpdateWriter(
                    conf.getRowField(),
                    conf.getColumnFamilyField(),
                    uniqueKeyFormatter,
                    keyValue,
                    new IdAddingSolrUpdateWriter(
                            conf.getUniqueKeyField(),
                            documentId,
                            conf.getTableNameField(),
                            tableName,
                            updateCollector));

            mapper.map(result, updateWriter);

        }
    }
}
 

private void verifyCountAndOrder(InternalScanner scanner,
    int expectedKVCount, int expectedRowCount, boolean forward)
    throws IOException {
  List<Cell> kvList = new ArrayList<>();
  Result lastResult = null;
  int rowCount = 0;
  int kvCount = 0;
  try {
    while (scanner.next(kvList)) {
      if (kvList.isEmpty()) continue;
      rowCount++;
      kvCount += kvList.size();
      if (lastResult != null) {
        Result curResult = Result.create(kvList);
        assertEquals("LastResult:" + lastResult + "CurResult:" + curResult,
            forward,
            Bytes.compareTo(curResult.getRow(), lastResult.getRow()) > 0);
      }
      lastResult = Result.create(kvList);
      kvList.clear();
    }
  } finally {
    scanner.close();
  }
  if (!kvList.isEmpty()) {
    rowCount++;
    kvCount += kvList.size();
    kvList.clear();
  }
  assertEquals(expectedKVCount, kvCount);
  assertEquals(expectedRowCount, rowCount);
}
 

@Test
public void testContainsTarget_True() {
    Result result = Result.create(Lists.<Cell>newArrayList(new KeyValue(ROW, COLUMN_FAMILY, COLUMN_QUALIFIER,
            Bytes.toBytes("value"))));
    assertTrue(extractor.containsTarget(result));
}
 

/**
 * Create Results from the cells using the cells meta data.
 * @param cellScanner
 * @param response
 * @return results
 */
public static Result[] getResults(CellScanner cellScanner, ScanResponse response)
    throws IOException {
  if (response == null) return null;
  // If cellscanner, then the number of Results to return is the count of elements in the
  // cellsPerResult list.  Otherwise, it is how many results are embedded inside the response.
  int noOfResults = cellScanner != null?
    response.getCellsPerResultCount(): response.getResultsCount();
  Result[] results = new Result[noOfResults];
  for (int i = 0; i < noOfResults; i++) {
    if (cellScanner != null) {
      // Cells are out in cellblocks.  Group them up again as Results.  How many to read at a
      // time will be found in getCellsLength -- length here is how many Cells in the i'th Result
      int noOfCells = response.getCellsPerResult(i);
      boolean isPartial =
          response.getPartialFlagPerResultCount() > i ?
              response.getPartialFlagPerResult(i) : false;
      List<Cell> cells = new ArrayList<>(noOfCells);
      for (int j = 0; j < noOfCells; j++) {
        try {
          if (cellScanner.advance() == false) {
            // We are not able to retrieve the exact number of cells which ResultCellMeta says us.
            // We have to scan for the same results again. Throwing DNRIOE as a client retry on the
            // same scanner will result in OutOfOrderScannerNextException
            String msg = "Results sent from server=" + noOfResults + ". But only got " + i
              + " results completely at client. Resetting the scanner to scan again.";
            LOG.error(msg);
            throw new DoNotRetryIOException(msg);
          }
        } catch (IOException ioe) {
          // We are getting IOE while retrieving the cells for Results.
          // We have to scan for the same results again. Throwing DNRIOE as a client retry on the
          // same scanner will result in OutOfOrderScannerNextException
          LOG.error("Exception while reading cells from result."
            + "Resetting the scanner to scan again.", ioe);
          throw new DoNotRetryIOException("Resetting the scanner.", ioe);
        }
        cells.add(cellScanner.current());
      }
      results[i] = Result.create(cells, null, response.getStale(), isPartial);
    } else {
      // Result is pure pb.
      results[i] = ProtobufUtil.toResult(response.getResults(i));
    }
  }
  return results;
}