@Test public void testScanWithValidFilter() throws InitializationException, IOException { final String tableName = "nifi"; final TestRunner runner = TestRunners.newTestRunner(TestProcessor.class); final Table table = Mockito.mock(Table.class); when(table.getName()).thenReturn(TableName.valueOf(tableName)); final MockHBaseClientService service = configureHBaseClientService(runner, table); runner.assertValid(service); final CollectingResultHandler handler = new CollectingResultHandler(); final HBaseClientService hBaseClientService = runner.getProcessContext().getProperty(TestProcessor.HBASE_CLIENT_SERVICE) .asControllerService(HBaseClientService.class); final String filter = "PrefixFilter ('Row') AND PageFilter (1) AND FirstKeyOnlyFilter ()"; hBaseClientService.scan(tableName, new ArrayList<Column>(), filter, System.currentTimeMillis(), handler); }
@Override public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler) throws IOException { Filter filter = null; if (!StringUtils.isBlank(filterExpression)) { ParseFilter parseFilter = new ParseFilter(); filter = parseFilter.parseFilterString(filterExpression); } try (final Table table = connection.getTable(TableName.valueOf(tableName)); final ResultScanner scanner = getResults(table, columns, filter, minTime)) { for (final Result result : scanner) { final byte[] rowKey = result.getRow(); final Cell[] cells = result.rawCells(); if (cells == null) { continue; } final ResultCell[] resultCells = new ResultCell[cells.length]; for (int i=0; i < cells.length; i++) { final Cell cell = cells[i]; final ResultCell resultCell = getResultCell(cell); resultCells[i] = resultCell; } handler.handle(rowKey, resultCells); } } }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler) throws IOException { Filter filter = null; if (!StringUtils.isBlank(filterExpression)) { ParseFilter parseFilter = new ParseFilter(); filter = parseFilter.parseFilterString(filterExpression); } try (final Table table = connection.getTable(TableName.valueOf(tableName)); final ResultScanner scanner = getResults(table, columns, filter, minTime)) { for (final Result result : scanner) { final byte[] rowKey = result.getRow(); final Cell[] cells = result.rawCells(); if (cells == null) { continue; } final ResultCell[] resultCells = new ResultCell[cells.length]; for (int i=0; i < cells.length; i++) { final Cell cell = cells[i]; final ResultCell resultCell = getResultCell(cell); resultCells[i] = resultCell; } handler.handle(rowKey, resultCells); } } } }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler) throws IOException { Filter filter = null; if (!StringUtils.isBlank(filterExpression)) { ParseFilter parseFilter = new ParseFilter(); filter = parseFilter.parseFilterString(filterExpression); } try (final Table table = connection.getTable(TableName.valueOf(tableName)); final ResultScanner scanner = getResults(table, columns, filter, minTime)) { for (final Result result : scanner) { final byte[] rowKey = result.getRow(); final Cell[] cells = result.rawCells(); if (cells == null) { continue; } final ResultCell[] resultCells = new ResultCell[cells.length]; for (int i=0; i < cells.length; i++) { final Cell cell = cells[i]; final ResultCell resultCell = getResultCell(cell); resultCells[i] = resultCell; } handler.handle(rowKey, resultCells); } } }  }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler) throws IOException { Filter filter = null; if (!StringUtils.isBlank(filterExpression)) { ParseFilter parseFilter = new ParseFilter(); filter = parseFilter.parseFilterString(filterExpression); } try (final Table table = connection.getTable(TableName.valueOf(tableName)); final ResultScanner scanner = getResults(table, columns, filter, minTime)) { for (final Result result : scanner) { final byte[] rowKey = result.getRow(); final Cell[] cells = result.rawCells(); if (cells == null) { continue; } final ResultCell[] resultCells = new ResultCell[cells.length]; for (int i=0; i < cells.length; i++) { final Cell cell = cells[i]; final ResultCell resultCell = getResultCell(cell); resultCells[i] = resultCell; } handler.handle(rowKey, resultCells); } } }  @OnEnabled void onEnabled(final ConfigurationContext context); @OnDisabled void shutdown(); @Override void put(final String tableName, final Collection<PutFlowFile> puts); @Override void put(final String tableName, final byte[] rowId, final Collection<PutColumn> columns); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s); }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler) throws IOException { Filter filter = null; if (!StringUtils.isBlank(filterExpression)) { ParseFilter parseFilter = new ParseFilter(); filter = parseFilter.parseFilterString(filterExpression); } try (final Table table = connection.getTable(TableName.valueOf(tableName)); final ResultScanner scanner = getResults(table, columns, filter, minTime)) { for (final Result result : scanner) { final byte[] rowKey = result.getRow(); final Cell[] cells = result.rawCells(); if (cells == null) { continue; } final ResultCell[] resultCells = new ResultCell[cells.length]; for (int i=0; i < cells.length; i++) { final Cell cell = cells[i]; final ResultCell resultCell = getResultCell(cell); resultCells[i] = resultCell; } handler.handle(rowKey, resultCells); } } }  @OnEnabled void onEnabled(final ConfigurationContext context); @OnDisabled void shutdown(); @Override void put(final String tableName, final Collection<PutFlowFile> puts); @Override void put(final String tableName, final byte[] rowId, final Collection<PutColumn> columns); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s);  }