HBaseDataStore implements DataStore { @Override @Timed public List<Document> saveAll(final Table table, List<Document> documents) { if(documents == null || documents.isEmpty()) { throw FoxtrotExceptions.createBadRequestException(table.getName(), "null/empty document list not allowed"); } List<Put> puts = new ArrayList<>(); ImmutableList.Builder<Document> translatedDocuments = ImmutableList.builder(); List<String> errorMessages = new ArrayList<>(); try { for(int i = 0; i < documents.size(); i++) { Document document = documents.get(i); if(!isValidDocument(document, errorMessages, i)) { continue; } Document translatedDocument = translator.translate(table, document); puts.add(getPutForDocument(translatedDocument)); translatedDocuments.add(translatedDocument); } } catch (JsonProcessingException e) { throw FoxtrotExceptions.createBadRequestException(table, e); } if(!errorMessages.isEmpty()) { throw FoxtrotExceptions.createBadRequestException(table.getName(), errorMessages); } try (org.apache.hadoop.hbase.client.Table hTable = tableWrapper.getTable(table)) { hTable.put(puts); } catch (IOException e) { logger.error("Error occurred while ingesting event in HBase : ", e); throw FoxtrotExceptions.createConnectionException(table, e); } return translatedDocuments.build(); } @Inject  HBaseDataStore(HbaseTableConnection tableWrapper, ObjectMapper mapper, DocumentTranslator translator); @Override @Timed void initializeTable(Table table, boolean forceTableCreate); @Override @Timed Document save(final Table table, Document document); @Override @Timed List<Document> saveAll(final Table table, List<Document> documents); @Override @Timed Document get(final Table table, String id); @Override @Timed List<Document> getAll(final Table table, List<String> ids); @VisibleForTesting Put getPutForDocument(Document document); @Override @Timed void updateTable(final Table table);  }
@Test public void testSaveBulk() throws Exception { DocumentTranslator documentTranslator = new DocumentTranslator(TestUtils.createTranslatorConfigWithRawKeyV1()); hbaseDataStore = new HBaseDataStore(hbaseTableConnection, mapper, documentTranslator); List<Document> documents = Lists.newArrayList(); for(int i = 0; i < 10; i++) { documents.add(createDummyDocument()); } hbaseDataStore.saveAll(TEST_APP, documents); for(Document document : documents) { validateSave(v1FormatKey(document.getId()), document); } }
@Test public void testSaveBulkNullDocuments() throws Exception { List<Document> documents = new Vector<Document>(); for(int i = 0; i < 10; i++) { documents.add(null); } try { hbaseDataStore.saveAll(TEST_APP, documents); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkNullIdList() throws Exception { List<Document> documents = null; try { hbaseDataStore.saveAll(TEST_APP, documents); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkNullId() throws Exception { List<Document> documents = new Vector<Document>(); for(int i = 0; i < 10; i++) { documents.add(new Document(null, System.currentTimeMillis(), mapper.valueToTree(Collections.singletonMap("TEST_NAME", "SINGLE_SAVE_TEST")) )); } try { hbaseDataStore.saveAll(TEST_APP, documents); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkNullData() throws Exception { List<Document> documents = new Vector<Document>(); for(int i = 0; i < 10; i++) { documents.add(new Document(UUID.randomUUID() .toString(), System.currentTimeMillis(), null)); } try { hbaseDataStore.saveAll(TEST_APP, documents); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkHBaseWriteException() throws Exception { List<Document> documents = new Vector<Document>(); documents.add(new Document(UUID.randomUUID() .toString(), System.currentTimeMillis(), mapper.valueToTree(Collections.singletonMap("TEST", "TEST")) )); doThrow(new IOException()).when(tableInterface) .put(Matchers.anyListOf(Put.class)); try { hbaseDataStore.saveAll(TEST_APP, documents); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.STORE_CONNECTION_ERROR, ex.getCode()); } }
@Test(expected = StoreConnectionException.class) public void testSaveBulkHBaseCloseException() throws Exception { List<Document> documents = new Vector<Document>(); documents.add(new Document(UUID.randomUUID() .toString(), System.currentTimeMillis(), mapper.valueToTree(Collections.singletonMap("TEST_NAME", "BULK_SAVE_TEST")) )); doThrow(new IOException()).when(tableInterface) .close(); hbaseDataStore.saveAll(TEST_APP, documents); }