[improve] [broker] Support create RawReader based on configuration

[hangc0276]

Fixes #xyz

Main Issue: #xyz

PIP: #xyz

Motivation

When I use one Pulsar client to create multiple RawReaders to consume multiple topics, I need fine-grained control on the RawReader, such as changing the receiver queue size and enabling the AutoScaledReceiverQueueSize to control the Pulsar client memory usage to avoid direct memory OOM.

However, the current RawReader configuration is hard-coded and can't change anything.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java

Lines 56 to 63 in 999e39b

	consumerConfiguration = new ConsumerConfigurationData<>();
	consumerConfiguration.getTopicNames().add(topic);
	consumerConfiguration.setSubscriptionName(subscription);
	consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
	consumerConfiguration.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
	consumerConfiguration.setReadCompacted(true);
	consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
	consumerConfiguration.setAckReceiptEnabled(true);

We can provide more flexible control over RawReader creation.

Modifications

Provide a more flexible API to allow user to pass ConsumerConfigurationData to create RawReader.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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

Documentation

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

Matching PR in forked repository

PR in forked repository:

[poorbarcode]

We can add a test to confirm the config has been used.

[hangc0276]

We can add a test to confirm the config has been used.

@poorbarcode Added a unit test.

[horizonzy]

LGTM