### Question:
SSTableIndexIndexer { public void index(final Path sstablePath) throws IOException { final FileSystem fileSystem = FileSystem.get(URI.create(sstablePath.toString()), configuration); final FileStatus fileStatus = fileSystem.getFileStatus(sstablePath); if (fileStatus.isDir()) { LOG.info("SSTable Indexing directory {}", sstablePath); final FileStatus[] statuses = fileSystem.listStatus(sstablePath); for (final FileStatus childStatus : statuses) { index(childStatus.getPath()); } } else if (sstablePath.toString().endsWith(SST_EXTENSION)) { final Path sstableIndexPath = new Path(sstablePath.toString() + SSTableIndexIndex.SSTABLE_INDEX_SUFFIX); if (fileSystem.exists(sstableIndexPath)) { LOG.info("Skipping as SSTable index file already exists for {}", sstablePath); } else { final ListenableFuture<IndexRequest> indexFuture = service.submit(new Callable<IndexRequest>() { @Override public IndexRequest call() throws Exception { final long startTime = System.currentTimeMillis(); final long fileSize = fileStatus.getLen(); LOG.info("Indexing SSTABLE Indexing file {}, size {} GB...", sstablePath, decimalFormat.format(fileSize / (1024.0 * 1024.0 * 1024.0))); indexSingleFile(fileSystem, sstablePath); return new IndexRequest(sstableIndexPath, startTime, fileSize); } }); Futures.addCallback(indexFuture, new FutureCallback<IndexRequest>() { public void onSuccess(final IndexRequest indexRequest) { long indexSize = 0; try { indexSize = fileSystem.getFileStatus(indexRequest.getIndexPath()).getLen(); } catch (IOException e) { LOG.error("Error getting file status for index path: {}", indexRequest.getIndexPath()); } final double elapsed = (System.currentTimeMillis() - indexRequest.getStartTime()) / 1000.0; LOG.info("Completed SSTABLE Indexing in {} seconds ({} MB/s). Index size is {} KB.", decimalFormat.format(elapsed), decimalFormat.format(indexRequest.getFileSize() / (1024.0 * 1024.0 * elapsed)), decimalFormat.format(indexSize / 1024.0)); } public void onFailure(Throwable e) { LOG.error("Failed to index.", e); } }); } } } SSTableIndexIndexer(final Configuration configuration); void index(final Path sstablePath); static void main(String[] args); void complete();  }### Answer:
@Test public void testIndex() throws Exception { final File tmpDir = Files.createTempDir(); tmpDir.deleteOnExit(); File sampleFile = new File(getClass().getResource(INDEX_FILE).getFile()); File tmpSampleFile = new File(tmpDir, sampleFile.getName()); Files.copy(sampleFile, tmpSampleFile); fileSystem.getConf().set("hadoop.sstable.split.mb", "1"); SSTableIndexIndexer indexer = new SSTableIndexIndexer(fileSystem.getConf()); indexer.index(new Path(tmpDir.getAbsolutePath(), sampleFile.getName())); indexer.complete(); SSTableIndexIndex index = SSTableIndexIndex.readIndex(fileSystem, new Path(tmpDir.getAbsolutePath(), sampleFile.getName())); assertNotNull(index); assertEquals(1, index.getOffsets().size()); assertEquals(0, index.getOffsets().get(0).getStart()); assertEquals(53, index.getOffsets().get(0).getEnd()); }