[issue #807] dlq topic producer options

[zzzming]

Fixes #807

Motivation

To customize producer options for DLQ topics.

Modifications

Add DLQProducerOptions to consumer's DLQ policy.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

The existing DLQ test is enhanced to cover customized producer options.

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

A new producer options field, DLQProuducerOptions, is introduced for DLQ policy to govern the producer options.

ConsumerOptions{
		Topics:              topics,
		DLQ: &DLQPolicy{
			MaxDeliveries:   3,
			DeadLetterTopic: dlqTopic,
			DLQProducerOptions: ProducerOptions{
				BatchingMaxPublishDelay: 100 * time.Millisecond,
			},
		},

• Dependencies (does it add or upgrade a dependency): no
• The public API: yes
• The schema: no
• The default values of configurations: yes (the existing lz4 compression type is the default.)
• The wire protocol: no

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (GoDocs)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[pgier]

LGTM, just had one in-line question.

[pgier]

Are there any cases where a user would want to explicitly set to NoCompression?

[zzzming]

The current implementation complies with the default LZ4 compression type in the previous implementation. So user cannot choose NoCompression. If NoCompression is required (I would prefer data has to be sent compressed as the original design), we might have to introduce another option to maintain backward compatibility.

[ne1llee]

DLQ and RLQ are created together, and the same data goes first to RLQ and then to DLQ，so the producer configuration of RLQ also needs to be modified?

pulsar-client-go/pulsar/retry_router.go

Lines 128 to 132 in 6a8e7f3

	producer, err := r.client.CreateProducer(ProducerOptions{
	Topic: r.policy.RetryLetterTopic,
	CompressionType: LZ4,
	BatchingMaxPublishDelay: 100 * time.Millisecond,
	})

[zzzming]

Added code to use the same ProducerOptions for both DLQ and RLQ.

DLQ and RLQ are created together, and the same data goes first to RLQ and then to DLQ，so the producer configuration of RLQ also needs to be modified?

pulsar-client-go/pulsar/retry_router.go

Lines 128 to 132 in 6a8e7f3

	producer, err := r.client.CreateProducer(ProducerOptions{
	Topic: r.policy.RetryLetterTopic,
	CompressionType: LZ4,
	BatchingMaxPublishDelay: 100 * time.Millisecond,
	})

[ne1llee]

May I ask which version this pull will be megred in？

[zzzming]

@wolfstudy @michaeljmarshall Can you help review this PR? @ne1llee is asking when to merge this?

[pgier]

LGTM

[michaeljmarshall]

LGTM