@Test public void testMultiplePutsSameRow() throws IOException, InitializationException { final String tableName = "nifi"; final String row = "row1"; final String columnFamily = "family1"; final String columnQualifier = "qualifier1"; final String content1 = "content1"; final String content2 = "content2"; final Collection<PutColumn> columns1 = Collections.singletonList(new PutColumn(columnFamily.getBytes(StandardCharsets.UTF_8), columnQualifier.getBytes(StandardCharsets.UTF_8), content1.getBytes(StandardCharsets.UTF_8))); final PutFlowFile putFlowFile1 = new PutFlowFile(tableName, row.getBytes(StandardCharsets.UTF_8), columns1, null); final Collection<PutColumn> columns2 = Collections.singletonList(new PutColumn(columnFamily.getBytes(StandardCharsets.UTF_8), columnQualifier.getBytes(StandardCharsets.UTF_8), content2.getBytes(StandardCharsets.UTF_8))); final PutFlowFile putFlowFile2 = new PutFlowFile(tableName, row.getBytes(StandardCharsets.UTF_8), columns2, 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(putFlowFile1, putFlowFile2)); ArgumentCaptor<List> capture = ArgumentCaptor.forClass(List.class); verify(table, times(1)).put(capture.capture()); final List<Put> puts = capture.getValue(); assertEquals(1, puts.size()); final NavigableMap<byte[], List<Cell>> familyCells = puts.get(0).getFamilyCellMap(); Map.Entry<byte[], List<Cell>> entry = familyCells.firstEntry(); assertEquals(2, entry.getValue().size()); }
@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, Put> rowPuts = new HashMap<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); Put put = rowPuts.get(rowKeyString); if (put == null) { put = new Put(putFlowFile.getRow()); rowPuts.put(rowKeyString, put); } for (final PutColumn column : putFlowFile.getColumns()) { put.addColumn( column.getColumnFamily(), column.getColumnQualifier(), column.getBuffer()); } } table.put(new ArrayList<>(rowPuts.values())); } }
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, Put> rowPuts = new HashMap<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); Put put = rowPuts.get(rowKeyString); if (put == null) { put = new Put(putFlowFile.getRow()); rowPuts.put(rowKeyString, put); } for (final PutColumn column : putFlowFile.getColumns()) { put.addColumn( column.getColumnFamily(), column.getColumnQualifier(), column.getBuffer()); } } table.put(new ArrayList<>(rowPuts.values())); } } }
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, Put> rowPuts = new HashMap<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); Put put = rowPuts.get(rowKeyString); if (put == null) { put = new Put(putFlowFile.getRow()); rowPuts.put(rowKeyString, put); } for (final PutColumn column : putFlowFile.getColumns()) { put.addColumn( column.getColumnFamily(), column.getColumnQualifier(), column.getBuffer()); } } table.put(new ArrayList<>(rowPuts.values())); } }  }
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, Put> rowPuts = new HashMap<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); Put put = rowPuts.get(rowKeyString); if (put == null) { put = new Put(putFlowFile.getRow()); rowPuts.put(rowKeyString, put); } for (final PutColumn column : putFlowFile.getColumns()) { put.addColumn( column.getColumnFamily(), column.getColumnQualifier(), column.getBuffer()); } } table.put(new ArrayList<>(rowPuts.values())); } }  @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 put(final String tableName, final Collection<PutFlowFile> puts) throws IOException { try (final Table table = connection.getTable(TableName.valueOf(tableName))) { final Map<String, Put> rowPuts = new HashMap<>(); for (final PutFlowFile putFlowFile : puts) { final String rowKeyString = new String(putFlowFile.getRow(), StandardCharsets.UTF_8); Put put = rowPuts.get(rowKeyString); if (put == null) { put = new Put(putFlowFile.getRow()); rowPuts.put(rowKeyString, put); } for (final PutColumn column : putFlowFile.getColumns()) { put.addColumn( column.getColumnFamily(), column.getColumnQualifier(), column.getBuffer()); } } table.put(new ArrayList<>(rowPuts.values())); } }  @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);  }