Pulsar Admin API client should support retries on operations

[frankjkelly]

Is your enhancement request related to a problem? Please describe.
When autoscaling brokers up and down the namespace bundles move about.
I would like for the Admin Client to handle the retries when a namespace bundle moves and I am trying to either create a new topic or delete a topic.

As best I can tell the Pulsar Client Admin code does not support retries .

Compare Admin Client https://github.com/apache/pulsar/blob/master/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java

to Pulsar Client
https://github.com/apache/pulsar/blob/04aa9e8e51869d1621a7e25402a656084eebfc09/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java

Describe alternatives you've considered
Leave things as they are but this means that the Pulsar Client and the Admin Client are inconsistent and
then I have to write the retry code myself which I am finding complex as certain exceptions are retryable e.g. 503
but others are not e.g. Topic already exists

CC: @sijie

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[frankjkelly]

Bump @sijie ?

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[frankjkelly]

Bump

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[martijngonlag]

@frankjkelly Based on some discussions, the recommended approach here would be to handle the retries on the client side by leveraging the pulsar-admin's response codes.

[frankjkelly]

@martijngonlag Sorry I took so long to get back to you.

The problem then becomes our code is littered with the following since some of the exceptions are not retryable and others are. And then we have to catch this further up and (using Spring) do the retries. It's an awful lot of boilerplate and particularly problematic for customers like us that create and delete a lot of topics. Especially with ZK being such a non-performant beast (and involved on the create/delete topic flow).

    private void createTopic(final URI topic) throws StreamManagerAdminException {
        try {
            val start = System.currentTimeMillis();
            pulsarAdmin.topics().createNonPartitionedTopic(topic.toString());
            val durationMs = System.currentTimeMillis() - start;
            recordDurationMetric("createTopic", durationMs);
        } catch (final PulsarAdminException e) {
            castAndRethrowException(e);
        }
    }


    private void castAndRethrowException(final PulsarAdminException e) throws StreamManagerAdminException {
        //Need to be able to distinguish between exceptions to be able to know which are retryable or not
        // However we do not want to expose Pulsar Internals beyond this layer
        if (e.getStatusCode() == 503) {
            throw new StreamManagerServiceUnavailableException(e);
        } else if (didPulsarAdminExhaustRetries(e)) {
            throw new StreamManagerRetryException(e);
        } else {
            throw new StreamManagerAdminException(e);
        }
    }

    private boolean didPulsarAdminExhaustRetries(final Throwable rootException) {
        for (val t : ExceptionUtils.getThrowableList(rootException)) {
            if (t instanceof AsyncHttpConnector.RetryException) {
                return true;
            }
        }
        return false;
    }

[eolivelli]

I think that it makes sense to retry some (most) of the admin operations.
We do retries while producing messages why can't we do them in the admin API ?