@Test public void testSinglePut() throws InitializationException, IOException { final String tableName = "nifi"; final String row = "row1"; final String columnFamily = "family1"; final String columnQualifier = "qualifier1"; final String content = "content1"; final Collection<PutColumn> columns = Collections.singletonList(new PutColumn(columnFamily.getBytes(StandardCharsets.UTF_8), columnQualifier.getBytes(StandardCharsets.UTF_8), content.getBytes(StandardCharsets.UTF_8))); final PutFlowFile putFlowFile = new PutFlowFile(tableName, row.getBytes(StandardCharsets.UTF_8), columns, null); final TestRunner runner = TestRunners.newTestRunner(TestProcessor.class); final Table table = Mockito.mock(Table.class); when(table.getName()).thenReturn(TableName.valueOf(tableName)); final HBaseClientService service = configureHBaseClientService(runner, table); runner.assertValid(service); final HBaseClientService hBaseClientService = runner.getProcessContext().getProperty(TestProcessor.HBASE_CLIENT_SERVICE) .asControllerService(HBaseClientService.class); hBaseClientService.put(tableName, Arrays.asList(putFlowFile)); ArgumentCaptor<List> capture = ArgumentCaptor.forClass(List.class); verify(table, times(1)).put(capture.capture()); final List<Put> puts = capture.getValue(); assertEquals(1, puts.size()); verifyPut(row, columnFamily, columnQualifier, content, puts.get(0)); }
@Override public void put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, List<PutColumn>> sorted = new HashMap<>(); final List<Put> newPuts = new ArrayList<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); List<PutColumn> columns = sorted.get(rowKeyString); if (columns == null) { columns = new ArrayList<>(); sorted.put(rowKeyString, columns); } columns.addAll(putFlowFile.getColumns()); } for (final Map.Entry<String, List<PutColumn>> entry : sorted.entrySet()) { newPuts.addAll(buildPuts(entry.getKey().getBytes(StandardCharsets.UTF_8), entry.getValue())); } table.put(newPuts); } }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, List<PutColumn>> sorted = new HashMap<>(); final List<Put> newPuts = new ArrayList<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); List<PutColumn> columns = sorted.get(rowKeyString); if (columns == null) { columns = new ArrayList<>(); sorted.put(rowKeyString, columns); } columns.addAll(putFlowFile.getColumns()); } for (final Map.Entry<String, List<PutColumn>> entry : sorted.entrySet()) { newPuts.addAll(buildPuts(entry.getKey().getBytes(StandardCharsets.UTF_8), entry.getValue())); } table.put(newPuts); } } }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, List<PutColumn>> sorted = new HashMap<>(); final List<Put> newPuts = new ArrayList<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); List<PutColumn> columns = sorted.get(rowKeyString); if (columns == null) { columns = new ArrayList<>(); sorted.put(rowKeyString, columns); } columns.addAll(putFlowFile.getColumns()); } for (final Map.Entry<String, List<PutColumn>> entry : sorted.entrySet()) { newPuts.addAll(buildPuts(entry.getKey().getBytes(StandardCharsets.UTF_8), entry.getValue())); } table.put(newPuts); } }  }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, List<PutColumn>> sorted = new HashMap<>(); final List<Put> newPuts = new ArrayList<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); List<PutColumn> columns = sorted.get(rowKeyString); if (columns == null) { columns = new ArrayList<>(); sorted.put(rowKeyString, columns); } columns.addAll(putFlowFile.getColumns()); } for (final Map.Entry<String, List<PutColumn>> entry : sorted.entrySet()) { newPuts.addAll(buildPuts(entry.getKey().getBytes(StandardCharsets.UTF_8), entry.getValue())); } table.put(newPuts); } }  @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 boolean checkAndPut(final String tableName, final byte[] rowId, final byte[] family, final byte[] qualifier, final byte[] value, final PutColumn column); @Override void delete(final String tableName, final byte[] rowId); @Override void delete(String tableName, byte[] rowId, String visibilityLabel); @Override void delete(String tableName, List<byte[]> rowIds); @Override void deleteCells(String tableName, List<DeleteRequest> deletes); @Override void delete(String tableName, List<byte[]> rowIds, String visibilityLabel); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(String tableName, Collection<Column> columns, String filterExpression, long minTime, List<String> visibilityLabels, ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, List<String> authorizations, final ResultHandler handler); @Override void scan(final String tableName, final String startRow, final String endRow, String filterExpression,
            final Long timerangeMin, final Long timerangeMax, final Integer limitRows, final Boolean isReversed,
            final Collection<Column> columns, List<String> visibilityLabels, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(float f); @Override byte[] toBytes(int i); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s); @Override String toTransitUri(String tableName, String rowKey); }
HBase_1_1_2_ClientService extends AbstractControllerService implements HBaseClientService { @Override public void put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, List<PutColumn>> sorted = new HashMap<>(); final List<Put> newPuts = new ArrayList<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); List<PutColumn> columns = sorted.get(rowKeyString); if (columns == null) { columns = new ArrayList<>(); sorted.put(rowKeyString, columns); } columns.addAll(putFlowFile.getColumns()); } for (final Map.Entry<String, List<PutColumn>> entry : sorted.entrySet()) { newPuts.addAll(buildPuts(entry.getKey().getBytes(StandardCharsets.UTF_8), entry.getValue())); } table.put(newPuts); } }  @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 boolean checkAndPut(final String tableName, final byte[] rowId, final byte[] family, final byte[] qualifier, final byte[] value, final PutColumn column); @Override void delete(final String tableName, final byte[] rowId); @Override void delete(String tableName, byte[] rowId, String visibilityLabel); @Override void delete(String tableName, List<byte[]> rowIds); @Override void deleteCells(String tableName, List<DeleteRequest> deletes); @Override void delete(String tableName, List<byte[]> rowIds, String visibilityLabel); @Override void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler); @Override void scan(String tableName, Collection<Column> columns, String filterExpression, long minTime, List<String> visibilityLabels, ResultHandler handler); @Override void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, List<String> authorizations, final ResultHandler handler); @Override void scan(final String tableName, final String startRow, final String endRow, String filterExpression,
            final Long timerangeMin, final Long timerangeMax, final Integer limitRows, final Boolean isReversed,
            final Collection<Column> columns, List<String> visibilityLabels, final ResultHandler handler); @Override byte[] toBytes(boolean b); @Override byte[] toBytes(float f); @Override byte[] toBytes(int i); @Override byte[] toBytes(long l); @Override byte[] toBytes(double d); @Override byte[] toBytes(String s); @Override byte[] toBytesBinary(String s); @Override String toTransitUri(String tableName, String rowKey);  }