[improve][broker] Add http produce backlog quota check

[Dream95]

Motivation

The binary protocol already performs check backlog quota in ServerCnx when creating producers, but the HTTP REST API path was missing this validation.

      .thenAccept(t -> {
            // TODO: Check message backlog and fail if backlog too large.
            if (!t.isPresent()) {
                // Topic not found, and remove from owning partition list.
                publishResult.completeExceptionally(new BrokerServiceException.TopicNotFoundException("Topic not "
                        + "owned by current broker."));
                TopicName topicName = TopicName.get(topic);
                pulsar().getBrokerService().getOwningTopics().get(topicName.getPartitionedTopicName())
                        .remove(topicName.getPartitionIndex());

Modifications

• Added backlog quota checks in publishSingleMessageToPartition method for both destination_storage and message_age quota types

The implementation follows the same pattern used in ServerCnx.handleProducer where backlog quota checks are performed before allowing message production.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as:

• Existing backlog quota tests should cover this functionality
• HTTP produce endpoint tests should verify the behavior

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Changes to REST endpoints:

• The HTTP produce endpoint (/persistent/{tenant}/{namespace}/{topic}/partitions/{partition}) now properly enforces backlog quota limits before publishing messages. When backlog quota is exceeded, the request will fail with an appropriate error, consistent with the binary protocol behavior.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: Dream95#4

[Denovo1998]

Now it does a backlog quota check for each message, maybe it can be optimized?

1. In internalPublishMessages(...), first calculate which partition(s) the current request will write to (actually, you have already calculated in round-robin which partition each message goes to).
2. Perform once for each partition:

• getTopic(partitionTopic)
• checkBacklogQuotaExceeded(...destination_storage + message_age)

3. After all checks pass, then enter the actual publish loop

• If the check for a certain partition fails: mark all the message results corresponding to that partition as failed (no longer attempt to publish), other partitions can continue.

[Denovo1998]

Should we pass in message.getProducerName(); here?

[Denovo1998]

Here you can first use FutureUtil.unwrapCompletionException.

[Denovo1998]

Perhaps it can be replaced with a "poll until quota exceeded (with timeout)" method.