class 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 {
                // Kick a thread for the index.
                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);
                    }
                });

            }
        }
    }

    public  SSTableIndexIndexer(final Configuration configuration);

}