[improve][client] PIP-409: support producer configuration for retry/dead letter topic producer

[thetumbled]

PIP: #24022
Fixes #19463
Fixes #21954

Motivation

Currently, we don't support configure the producer of retry/dead letter topic. But enable the chunk message feature
and disable the batch configuration in hard code, which can't handle many situations. For example, when the throughput
of message of retry topic become considerable, the resource consumed by the un-batched messages is pretty large.
There is no reason that we disable the batch message feature.

For better control for the retry/dead letter topic feature, we can support configuration for the producer of
retry/dead letter topic.

Modifications

Support configuration for the producer of retry/dead letter topic.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

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: thetumbled#71

[thetumbled]

The pip document has been merged, we can continue to review this pr. Thanks for your review. @BewareMyPower @nodece @dao-jun @codelipenghui @shibd @Technoboy-

[lhotari]

Instead of using java.util.function.Function, I'd suggest using a richer interface so that more context parameters can be added besides the topic name. This pattern would also allow adding more context parameters later without breaking the API in backwards incompatible ways.

[lhotari]

Instead of using java.util.function.Function, I'd suggest using a richer interface so that more context parameters can be added besides the topic name.

Suggested change

	/**
	* Function to build the producer for the retry letter topic.
	* The input parameter is the topic name.
	*/
	private Function<String, ProducerBuilder<byte[]>> retryLetterProducerBuilder;
	/**
	* Function to build the producer for the dead letter topic.
	* The input parameter is the topic name.
	*/
	private Function<String, ProducerBuilder<byte[]>> deadLetterProducerBuilder;
	/**
	* Customizer for configuring the producer builder for the retry letter topic.
	*
	* <p>This field holds a function that allows the caller to customize the producer builder
	* settings for the retry letter topic before the producer is created. The customization logic
	* can use the provided context (which includes input topic and subscription details) to adjust
	* configurations such as timeouts, batching, or message routing.
	*/
	private ProducerBuilderCustomizer retryLetterProducerBuilder;
	/**
	* Customizer for configuring the producer builder for the dead letter topic.
	*
	* <p>This field holds a function that allows the caller to customize the producer builder
	* settings for the dead letter topic before the producer is created. Using the provided context,
	* implementations can perform specific adjustments that ensure the dead letter queue operates
	* with the appropriate configurations tailored for handling undeliverable messages.
	*/
	private ProducerBuilderCustomizer deadLetterProducerBuilder;
	/**
	* Functional interface for customizing a producer builder for a specific topic.
	*
	* <p>This interface allows for customizing the producer builder configuration for either the retry letter topic
	* or the dead letter topic. The customization might include setting producer properties such as batching, timeouts,
	* or any other producer-specific configuration.
	*
	* @see ProducerBuilderContext
	*/
	public interface ProducerBuilderCustomizer {
	/**
	* Customize the given producer builder with settings specific to the topic context provided.
	*
	* @param context the context containing information about the input topic and the subscription
	* @param producerBuilder the producer builder instance to be customized
	*/
	void customize(ProducerBuilderContext context, ProducerBuilder<byte[]> producerBuilder);
	}
	/**
	* Provides context information required for customizing a producer builder.
	*
	* <p>This interface supplies relevant details such as the name of the input topic and associated subscription name.
	* This contextual information helps in correctly configuring the producer for the appropriate topic.
	*/
	public interface ProducerBuilderContext {
	/**
	* Returns the default name of topic for the dead letter or retry letter producer. This topic name is used
	* unless the ProducerBuilderCustomizer overrides it.
	*
	* @return a {@code String} representing the input topic name
	*/
	String getDefaultTopicName();
	/**
	* Returns the name of the input topic for which the dead letter or retry letter producer is being configured.
	*
	* @return a {@code String} representing the input topic name
	*/
	String getInputTopicName();
	/**
	* Returns the name of the subscription for which the dead letter or retry letter producer is being configured.
	*
	* @return a {@code String} representing the subscription name
	*/
	String getInputTopicSubscriptionName();
	}

[lhotari]

Great work, @thetumbled. Thanks for making this happen. The PIP-409 solution will help resolve long time issues such as #19463 and #21954.