addCallback.java

    @Override
    public void addCallback(KairosTableRow<C, V, K> tableRow, FutureCallback<ResultSet> callback)
    {
        bufferCounter.incrementAndGet();
        executorService.submit(() -> {
            bufferCounter.decrementAndGet();
            baseKairosTableRepository.addCallback(tableRow, new FutureCallback<ResultSet>()
            {
                @Override
                public void onSuccess(@Nullable ResultSet result)
                {
                    callback.onSuccess(result);
                }

                @Override
                public void onFailure(Throwable t)
                {
                    if (t instanceof InvalidQueryException) {
                        LOG.error("Invalid query, skipping.", t);
                    } else {
                        addCallback(tableRow, callback);
                    }
                    callback.onFailure(t);
                }
            });
        });
    }