class RetryOnFailureResultReceiver<T> implements ResultReceiver<T> {

    private static final Logger LOGGER = LogManager.getLogger(RetryOnFailureResultReceiver.class);

    private final ClusterService clusterService;
    private final ClusterState initialState;
    private final Predicate<String> hasIndex;
    private final ResultReceiver<T> delegate;
    private final UUID jobId;
    private final BiConsumer<UUID, ResultReceiver<T>> retryAction;
    private int attempt = 1;

    public RetryOnFailureResultReceiver(ClusterService clusterService,
                                        ClusterState initialState,
                                        Predicate<String> hasIndex,
                                        ResultReceiver<T> delegate,
                                        UUID jobId,
                                        BiConsumer<UUID, ResultReceiver<T>> retryAction) {
        this.clusterService = clusterService;
        this.initialState = initialState;
        this.hasIndex = hasIndex;
        this.delegate = delegate;
        this.jobId = jobId;
        this.retryAction = retryAction;
    }

    @Override
    public void setNextRow(Row row) {
        delegate.setNextRow(row);
    }

    @Override
    public void batchFinished() {
        delegate.batchFinished();
    }

    @Override
    public void allFinished() {
        delegate.allFinished();
    }

    @Override
    public void fail(Throwable wrappedError) {<FILL_FUNCTION_BODY>}

    private boolean indexWasTemporaryUnavailable(Throwable t) {
        return t instanceof IndexNotFoundException && hasIndex.test(((IndexNotFoundException) t).getIndex().getName());
    }

    private void retry() {
        UUID newJobId = UUIDs.dirtyUUID();
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Retrying statement due to a shard failure, attempt={}, jobId={}->{}", attempt, jobId, newJobId);
        }
        retryAction.accept(newJobId, this);
    }

    @Override
    public CompletableFuture<T> completionFuture() {
        return delegate.completionFuture();
    }

    @Override
    public String toString() {
        return "RetryOnFailureResultReceiver{" +
               "delegate=" + delegate +
               ", jobId=" + jobId +
               ", attempt=" + attempt +
               '}';
    }
}

        final Throwable error = SQLExceptions.unwrap(wrappedError);
        if (attempt <= Constants.MAX_SHARD_MISSING_RETRIES &&
            (SQLExceptions.isShardFailure(error) || error instanceof ConnectTransportException || indexWasTemporaryUnavailable(error))) {

            if (clusterService.state().blocks().hasGlobalBlockWithStatus(RestStatus.SERVICE_UNAVAILABLE)) {
                delegate.fail(error);
            } else {
                ClusterStateObserver clusterStateObserver = new ClusterStateObserver(initialState, clusterService, null, LOGGER);
                clusterStateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
                    @Override
                    public void onNewClusterState(ClusterState state) {
                        attempt += 1;
                        retry();
                    }

                    @Override
                    public void onClusterServiceClose() {
                        delegate.fail(error);
                    }

                    @Override
                    public void onTimeout(TimeValue timeout) {
                        delegate.fail(error);
                    }
                });
            }
        } else {
            delegate.fail(error);
        }
    
<no_super_class>