[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: