@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); } }
@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(); }
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(); } }
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); }
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); }
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);  }