class HBaseDataStore implements DataStore {

    @Override
    @Timed
    public List<Document> getAll(final Table table, List<String> ids) {
        if(CollectionUtils.isEmpty(ids)) {
            throw FoxtrotExceptions.createBadRequestException(table.getName(), "Empty ID List");
        }
        try (org.apache.hadoop.hbase.client.Table hTable = tableWrapper.getTable(table)) {
            List<Get> gets = new ArrayList<>(ids.size());
            for(String id : ids) {
                Get get = new Get(Bytes.toBytes(translator.rawStorageIdFromDocumentId(table, id))).addColumn(COLUMN_FAMILY,
                                                                                                             DOCUMENT_FIELD_NAME
                                                                                                            )
                        .addColumn(COLUMN_FAMILY, DOCUMENT_META_FIELD_NAME)
                        .addColumn(COLUMN_FAMILY, TIMESTAMP_FIELD_NAME)
                        .setMaxVersions(1);
                gets.add(get);
            }
            Result[] getResults = hTable.get(gets);
            List<String> missingIds = new ArrayList<>();
            List<Document> results = new ArrayList<>(ids.size());
            for(int index = 0; index < getResults.length; index++) {
                Result getResult = getResults[index];
                if(!getResult.isEmpty()) {
                    byte[] data = getResult.getValue(COLUMN_FAMILY, DOCUMENT_FIELD_NAME);
                    byte[] metadata = getResult.getValue(COLUMN_FAMILY, DOCUMENT_META_FIELD_NAME);
                    byte[] timestamp = getResult.getValue(COLUMN_FAMILY, TIMESTAMP_FIELD_NAME);
                    long time = Bytes.toLong(timestamp);
                    DocumentMetadata documentMetadata = (null != metadata) ? mapper.readValue(metadata, DocumentMetadata.class) : null;
                    final String docId = (null == metadata) ? Bytes.toString(getResult.getRow())
                            .split(":")[0] : documentMetadata.getRawStorageId();
                    results.add(translator.translateBack(new Document(docId, time, documentMetadata, mapper.readTree(data))));
                } else {
                    missingIds.add(ids.get(index));
                }
            }
            if (!missingIds.isEmpty()) {
                logger.error("ID's missing in HBase - {}", missingIds);
                throw FoxtrotExceptions.createMissingDocumentsException(table, missingIds);
            }
            return results;
        } catch (JsonProcessingException e) {
            throw FoxtrotExceptions.createBadRequestException(table, e);
        } catch (IOException e) {
            throw FoxtrotExceptions.createConnectionException(table, e);
        }
    }

    @Inject public  HBaseDataStore(HbaseTableConnection tableWrapper, ObjectMapper mapper, DocumentTranslator translator);

}