@Test public void testBatchCoprocessorWithoutStartStopKey() throws Throwable { CrossSiteHTable table = new CrossSiteHTable(TEST_UTIL.getConfiguration(), TABLE_NAME); Map<byte[], SumResponse> results = sumInBatch(table, null, CF, QN, HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW); int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 1; i < 11; i++) { expectedResult += i; } expectedResult *= 2; assertEquals("Invalid result", expectedResult, sumResult); table.close(); }
@Test public void testBatchCoprocessorWithPartKeys() throws Throwable { CrossSiteHTable table = new CrossSiteHTable(TEST_UTIL.getConfiguration(), TABLE_NAME); Map<byte[], SumResponse> results = sumInBatch(table, null, CF, QN, Bytes.toBytes("hbase1,00"), Bytes.toBytes("hbase1,05")); int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 1; i <= 6; i++) { expectedResult += i; } assertEquals("Invalid result", expectedResult, sumResult); table.close(); }
@Test public void testBatchCoprocessorWithClusterAndPartKeys() throws Throwable { CrossSiteHTable table = new CrossSiteHTable(TEST_UTIL.getConfiguration(), TABLE_NAME); Map<byte[], SumResponse> results = sumInBatch(table, new String[] { "hbase1", "hbase2" }, CF, QN, Bytes.toBytes("00"), Bytes.toBytes("05")); int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 1; i <= 6; i++) { expectedResult += i; } expectedResult *= 2; assertEquals("Invalid result", expectedResult, sumResult); table.close(); }
private Map<byte[], SumResponse> sumInBatch(final CrossSiteHTable table, final String[] clusterNames, final byte[] family, final byte[] qualifier, final byte[] start, final byte[] end) throws ServiceException, Throwable { ColumnAggregationProtos.SumRequest.Builder builder = ColumnAggregationProtos.SumRequest .newBuilder(); builder.setFamily(HBaseZeroCopyByteString.wrap(family)); if (qualifier != null && qualifier.length > 0) { builder.setQualifier(HBaseZeroCopyByteString.wrap(qualifier)); } final Map<byte[], ColumnAggregationProtos.SumResponse> results = Collections .synchronizedMap(new TreeMap<byte[], ColumnAggregationProtos.SumResponse>( Bytes.BYTES_COMPARATOR)); table.batchCoprocessorService(ColumnAggregationProtos.ColumnAggregationService.getDescriptor() .findMethodByName("sum"), builder.build(), start, end, clusterNames, ColumnAggregationProtos.SumResponse.getDefaultInstance(), new Callback<ColumnAggregationProtos.SumResponse>() { @Override public void update(byte[] region, byte[] row, ColumnAggregationProtos.SumResponse result) { if (region != null) { results.put(region, result); } } }); return results; }
@Test public void testAggregationNullResponse() throws Throwable { Table table = new HTable(util.getConfiguration(), TEST_TABLE); ColumnAggregationWithNullResponseProtos.SumRequest.Builder builder = ColumnAggregationWithNullResponseProtos.SumRequest .newBuilder(); builder.setFamily(ByteStringer.wrap(TEST_FAMILY)); if (TEST_QUALIFIER != null && TEST_QUALIFIER.length > 0) { builder.setQualifier(ByteStringer.wrap(TEST_QUALIFIER)); } Map<byte[], ColumnAggregationWithNullResponseProtos.SumResponse> results = table.batchCoprocessorService( ColumnAggregationServiceNullResponse.getDescriptor().findMethodByName("sum"), builder.build(), ROWS[0], ROWS[ROWS.length - 1], ColumnAggregationWithNullResponseProtos.SumResponse.getDefaultInstance()); int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], ColumnAggregationWithNullResponseProtos.SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 0; i < rowSeperator2; i++) { expectedResult += i; } assertEquals("Invalid result", expectedResult, sumResult); table.close(); }
private Map<byte[], SumResponse> sum(final Table table, final byte[] family, final byte[] qualifier, final byte[] start, final byte[] end) throws ServiceException, Throwable { ColumnAggregationProtos.SumRequest.Builder builder = ColumnAggregationProtos.SumRequest .newBuilder(); builder.setFamily(ByteStringer.wrap(family)); if (qualifier != null && qualifier.length > 0) { builder.setQualifier(ByteStringer.wrap(qualifier)); } return table.batchCoprocessorService( ColumnAggregationProtos.ColumnAggregationService.getDescriptor().findMethodByName("sum"), builder.build(), start, end, ColumnAggregationProtos.SumResponse.getDefaultInstance()); }
@Test public void testAggregationNullResponse() throws Throwable { HTable table = new HTable(util.getConfiguration(), TEST_TABLE); ColumnAggregationWithNullResponseProtos.SumRequest.Builder builder = ColumnAggregationWithNullResponseProtos.SumRequest .newBuilder(); builder.setFamily(HBaseZeroCopyByteString.wrap(TEST_FAMILY)); if (TEST_QUALIFIER != null && TEST_QUALIFIER.length > 0) { builder.setQualifier(HBaseZeroCopyByteString.wrap(TEST_QUALIFIER)); } Map<byte[], ColumnAggregationWithNullResponseProtos.SumResponse> results = table.batchCoprocessorService( ColumnAggregationServiceNullResponse.getDescriptor().findMethodByName("sum"), builder.build(), ROWS[0], ROWS[ROWS.length - 1], ColumnAggregationWithNullResponseProtos.SumResponse.getDefaultInstance()); int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], ColumnAggregationWithNullResponseProtos.SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 0; i < rowSeperator2; i++) { expectedResult += i; } assertEquals("Invalid result", expectedResult, sumResult); table.close(); }
private Map<byte[], SumResponse> sum(final HTable table, final byte[] family, final byte[] qualifier, final byte[] start, final byte[] end) throws ServiceException, Throwable { ColumnAggregationProtos.SumRequest.Builder builder = ColumnAggregationProtos.SumRequest .newBuilder(); builder.setFamily(HBaseZeroCopyByteString.wrap(family)); if (qualifier != null && qualifier.length > 0) { builder.setQualifier(HBaseZeroCopyByteString.wrap(qualifier)); } return table.batchCoprocessorService( ColumnAggregationProtos.ColumnAggregationService.getDescriptor().findMethodByName("sum"), builder.build(), start, end, ColumnAggregationProtos.SumResponse.getDefaultInstance()); }
private Map<byte[], SumResponse> sum(final Table table, final byte[] family, final byte[] qualifier, final byte[] start, final byte[] end) throws ServiceException, Throwable { ColumnAggregationProtos.SumRequest.Builder builder = ColumnAggregationProtos.SumRequest .newBuilder(); builder.setFamily(ByteString.copyFrom(family)); if (qualifier != null && qualifier.length > 0) { builder.setQualifier(ByteString.copyFrom(qualifier)); } return table.batchCoprocessorService( ColumnAggregationProtos.ColumnAggregationService.getDescriptor().findMethodByName("sum"), builder.build(), start, end, ColumnAggregationProtos.SumResponse.getDefaultInstance()); }
@Test public void testAggregationWithErrors() throws Throwable { Table table = new HTable(util.getConfiguration(), TEST_TABLE); final Map<byte[], ColumnAggregationWithErrorsProtos.SumResponse> results = Collections.synchronizedMap( new TreeMap<byte[], ColumnAggregationWithErrorsProtos.SumResponse>( Bytes.BYTES_COMPARATOR )); ColumnAggregationWithErrorsProtos.SumRequest.Builder builder = ColumnAggregationWithErrorsProtos.SumRequest .newBuilder(); builder.setFamily(ByteStringer.wrap(TEST_FAMILY)); if (TEST_QUALIFIER != null && TEST_QUALIFIER.length > 0) { builder.setQualifier(ByteStringer.wrap(TEST_QUALIFIER)); } boolean hasError = false; try { table.batchCoprocessorService( ColumnAggregationWithErrorsProtos.ColumnAggregationServiceWithErrors.getDescriptor() .findMethodByName("sum"), builder.build(), ROWS[0], ROWS[ROWS.length - 1], ColumnAggregationWithErrorsProtos.SumResponse.getDefaultInstance(), new Batch.Callback<ColumnAggregationWithErrorsProtos.SumResponse>() { @Override public void update(byte[] region, byte[] row, ColumnAggregationWithErrorsProtos.SumResponse result) { results.put(region, result); } }); } catch (Throwable t) { LOG.info("Exceptions in coprocessor service", t); hasError = true; } int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], ColumnAggregationWithErrorsProtos.SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 0; i < rowSeperator2; i++) { expectedResult += i; } assertEquals("Invalid result", expectedResult, sumResult); assertTrue(hasError); table.close(); }
@Test public void testAggregationWithErrors() throws Throwable { HTable table = new HTable(util.getConfiguration(), TEST_TABLE); final Map<byte[], ColumnAggregationWithErrorsProtos.SumResponse> results = Collections.synchronizedMap( new TreeMap<byte[], ColumnAggregationWithErrorsProtos.SumResponse>( Bytes.BYTES_COMPARATOR )); ColumnAggregationWithErrorsProtos.SumRequest.Builder builder = ColumnAggregationWithErrorsProtos.SumRequest .newBuilder(); builder.setFamily(HBaseZeroCopyByteString.wrap(TEST_FAMILY)); if (TEST_QUALIFIER != null && TEST_QUALIFIER.length > 0) { builder.setQualifier(HBaseZeroCopyByteString.wrap(TEST_QUALIFIER)); } boolean hasError = false; try { table.batchCoprocessorService( ColumnAggregationWithErrorsProtos.ColumnAggregationServiceWithErrors.getDescriptor() .findMethodByName("sum"), builder.build(), ROWS[0], ROWS[ROWS.length - 1], ColumnAggregationWithErrorsProtos.SumResponse.getDefaultInstance(), new Batch.Callback<ColumnAggregationWithErrorsProtos.SumResponse>() { @Override public void update(byte[] region, byte[] row, ColumnAggregationWithErrorsProtos.SumResponse result) { results.put(region, result); } }); } catch (Throwable t) { LOG.info("Exceptions in coprocessor service", t); hasError = true; } int sumResult = 0; int expectedResult = 0; for (Map.Entry<byte[], ColumnAggregationWithErrorsProtos.SumResponse> e : results.entrySet()) { LOG.info("Got value " + e.getValue().getSum() + " for region " + Bytes.toStringBinary(e.getKey())); sumResult += e.getValue().getSum(); } for (int i = 0; i < rowSeperator2; i++) { expectedResult += i; } assertEquals("Invalid result", expectedResult, sumResult); assertTrue(hasError); table.close(); }