This is an automated email from the ASF dual-hosted git repository.
RobertIndie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new ef2a6317 [improve][doc] Add chunking documentation for DLQ/RLQ and add
integration tests (#1511)
ef2a6317 is described below
commit ef2a63176b874e175e95ae6fc22e5f27ecc093a5
Author: zhou zhuohan <[email protected]>
AuthorDate: Tue Jun 30 00:04:12 2026 +0800
[improve][doc] Add chunking documentation for DLQ/RLQ and add integration
tests (#1511)
Master Issue: https://github.com/apache/pulsar/pull/21048
Related Issue: https://github.com/apache/pulsar-client-go/pull/805
### Motivation
In the Java SDK (PR https://github.com/apache/pulsar/pull/21048), the
DLQ/RLQ producers are hardcoded with
`enableBatching(false).enableChunking(true)` to ensure large chunked
messages can be forwarded to DLQ/RLQ
topics successfully.
However, in the Go SDK, the DLQ/RLQ producers are created using
`DLQPolicy.ProducerOptions`, which defaults
to a zero-value `ProducerOptions{}` (i.e., `EnableChunking: false`,
`DisableBatching: false`). This means
that if a producer sends a chunked message larger than the broker's
`maxMessageSize`, and the consumer is
configured with DLQ/RLQ, the DLQ/RLQ producer will fail to forward the
message because it cannot chunk it.
We considered automatically enabling chunking in the DLQ/RLQ producers when
the user has not explicitly
configured `ProducerOptions`. However, since `EnableChunking` and
`DisableBatching` are both `bool` fields
with a zero value of `false`, there is no way to distinguish between:
- The user intentionally setting them to `false` (meaning "I don't want
chunking")
- The user not configuring them at all (meaning "I didn't think about it")
Automatically overriding these values would break backward compatibility
for users who explicitly set
`EnableChunking: false`. Therefore, instead of changing the default
behavior, we add documentation to
guide users to enable chunking in `DLQPolicy.ProducerOptions` when using
chunked messages with DLQ/RLQ.
### Modifications
- Added a note to the `EnableChunking` field documentation in
`ProducerOptions` (`producer.go`) explaining
that when chunking is enabled and the consumer uses DLQ/RLQ, it is
recommended to also enable chunking
in `DLQPolicy.ProducerOptions`.
- Added a note to the `DLQPolicy.ProducerOptions` field documentation
(`consumer.go`) explaining that if
the messages being consumed were produced with chunking enabled, it is
recommended to also enable chunking
here. This helps in decoupled producer/consumer scenarios where the
consumer developer may not be aware
of the producer's chunking configuration.
- Added `TestChunkReconsumeLater` integration test to verify chunked
messages (5MB, exceeding broker
maxMessageSize) can be sent to the retry topic via `ReconsumeLater` and
routed to DLQ after exceeding
max retries, when `DLQPolicy.ProducerOptions` has chunking enabled.
- Added `TestChunkDLQWithNack` integration test to verify chunked messages
(5MB) are routed to DLQ
after exceeding max redelivery count via `Nack`, when
`DLQPolicy.ProducerOptions` has chunking enabled.
### Verifying this change
This change added tests and can be verified as follows:
- Added `TestChunkReconsumeLater`: sends a 5MB chunked message, verifies it
flows through retry topic
via `ReconsumeLater` and eventually lands in DLQ after exceeding
`MaxDeliveries`.
- Added `TestChunkDLQWithNack`: sends a 5MB chunked message, verifies it is
routed to DLQ after
exceeding max redelivery count via `Nack`, with original properties
preserved.
Both tests use payloads larger than broker `maxMessageSize` (~1MB) to
ensure that DLQ/RLQ producers
**must** have chunking enabled to successfully forward messages. Without
chunking enabled in
`DLQPolicy.ProducerOptions`, these tests would timeout.
---
pulsar/consumer.go | 9 +-
pulsar/message_chunking_test.go | 223 ++++++++++++++++++++++++++++++++++++++++
pulsar/producer.go | 12 +++
3 files changed, 243 insertions(+), 1 deletion(-)
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 9aeabb03..5cff870f 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -83,7 +83,14 @@ type DLQPolicy struct {
// a topic.
DeadLetterTopicProducerName string
- // ProducerOptions is the producer options to produce messages to the
DLQ and RLQ topic
+ // ProducerOptions is the producer options to produce messages to the
DLQ and RLQ topic.
+ // If the messages being consumed were produced with chunking enabled,
it is recommended to also enable
+ // chunking here so that large chunked messages can be forwarded to the
DLQ/RLQ topic successfully.
+ // Example:
+ // ProducerOptions: pulsar.ProducerOptions{
+ // EnableChunking: true,
+ // DisableBatching: true,
+ // }
ProducerOptions ProducerOptions
// RetryLetterTopic specifies the name of the topic where the retry
messages will be sent.
diff --git a/pulsar/message_chunking_test.go b/pulsar/message_chunking_test.go
index 2d6462e7..546412f9 100644
--- a/pulsar/message_chunking_test.go
+++ b/pulsar/message_chunking_test.go
@@ -745,3 +745,226 @@ func TestResendChunkWithAckHoleMessages(t *testing.T) {
cancel()
assert.ErrorIs(t, err, context.DeadlineExceeded)
}
+
+// TestChunkReconsumeLater tests that chunked messages can be sent to the
retry topic via ReconsumeLater,
+// and are routed to the DLQ topic after exceeding the maximum number of
retries.
+// Payload exceeds broker maxMessageSize, so RLQ/DLQ producers must enable
chunking.
+func TestChunkReconsumeLater(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topic := newTopicName()
+ subName := fmt.Sprintf("chunk-rlq-sub-%d", time.Now().Unix())
+ maxRedeliveries := 2
+
+ // Create producer with chunking enabled
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: true,
+ EnableChunking: true,
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, producer)
+ defer producer.Close()
+
+ // Create consumer with retry and DLQ enabled
+ // DLQPolicy.ProducerOptions enables chunking to ensure RLQ/DLQ
producers can send large messages
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: subName,
+ Type: Shared,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ DLQ: &DLQPolicy{
+ MaxDeliveries: uint32(maxRedeliveries),
+ ProducerOptions: ProducerOptions{
+ DisableBatching: true,
+ EnableChunking: true,
+ },
+ },
+ RetryEnable: true,
+ NackRedeliveryDelay: 1 * time.Second,
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, consumer)
+ defer consumer.Close()
+
+ // Send a chunked message larger than broker maxMessageSize (5MB, far
exceeding the 1MB broker limit)
+ // If RLQ/DLQ producer does not enable chunking, sending will fail
+ content := createTestMessagePayload(5 * _brokerMaxMessageSize)
+ msgID, err := producer.Send(context.Background(), &ProducerMessage{
+ Payload: content,
+ Key: "chunk-key",
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, msgID)
+
+ // First receive and ReconsumeLater
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ msg, err := consumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.Equal(t, content, msg.Payload())
+ assert.Equal(t, "chunk-key", msg.Key())
+ consumer.ReconsumeLater(msg, 1*time.Second)
+
+ // Second receive (from retry topic), ReconsumeLater again
+ // If RLQ producer does not enable chunking, the message won't be
received here (send failure)
+ ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
+ msg, err = consumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.Equal(t, content, msg.Payload())
+ consumer.ReconsumeLater(msg, 1*time.Second)
+
+ // Third receive (from retry topic), ReconsumeLater again, now exceeds
maxRedeliveries, should route to DLQ
+ ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
+ msg, err = consumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.Equal(t, content, msg.Payload())
+ consumer.ReconsumeLater(msg, 1*time.Second)
+
+ // Confirm no more messages on the original consumer
+ ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+ msg, err = consumer.Receive(ctx)
+ cancel()
+ assert.ErrorIs(t, err, context.DeadlineExceeded)
+ assert.Nil(t, msg)
+
+ // Consume from DLQ topic, verify the large chunked message is
correctly routed to DLQ
+ // If DLQ producer does not enable chunking, the message won't be
received here
+ dlqTopic := "persistent://public/default/" + topic + "-" + subName +
"-DLQ"
+ dlqConsumer, err := client.Subscribe(ConsumerOptions{
+ Topic: dlqTopic,
+ SubscriptionName: "dlq-sub",
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.NoError(t, err)
+ defer dlqConsumer.Close()
+
+ ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
+ dlqMsg, err := dlqConsumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.Equal(t, content, dlqMsg.Payload())
+ assert.Equal(t, "chunk-key", dlqMsg.Key())
+
+ // Verify DLQ message properties
+ assert.NotEmpty(t, dlqMsg.Properties()[SysPropertyRealTopic])
+ assert.NotEmpty(t, dlqMsg.Properties()[SysPropertyOriginMessageID])
+
+ assert.NoError(t, dlqConsumer.Ack(dlqMsg))
+
+ // No more messages on the DLQ topic
+ ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second)
+ dlqMsg, err = dlqConsumer.Receive(ctx)
+ cancel()
+ assert.ErrorIs(t, err, context.DeadlineExceeded)
+ assert.Nil(t, dlqMsg)
+}
+
+// TestChunkDLQWithNack tests that chunked messages trigger redelivery via
Nack,
+// and are routed to the DLQ topic after exceeding the maximum redelivery
count.
+// Uses a payload larger than broker maxMessageSize to ensure DLQ producer
must enable chunking to send successfully.
+func TestChunkDLQWithNack(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topic := newTopicName()
+ dlqTopic := newTopicName()
+ subName := "chunk-dlq-sub"
+ maxRedeliveries := uint32(2)
+
+ // Create DLQ consumer
+ dlqConsumer, err := client.Subscribe(ConsumerOptions{
+ Topic: dlqTopic,
+ SubscriptionName: "dlq-verify",
+ })
+ assert.NoError(t, err)
+ defer dlqConsumer.Close()
+
+ // Create producer with chunking enabled
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topic,
+ DisableBatching: true,
+ EnableChunking: true,
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, producer)
+ defer producer.Close()
+
+ // Create consumer with DLQ configured
+ // DLQPolicy.ProducerOptions enables chunking to ensure DLQ producer
can send large messages
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: subName,
+ Type: Shared,
+ DLQ: &DLQPolicy{
+ MaxDeliveries: maxRedeliveries,
+ DeadLetterTopic: dlqTopic,
+ ProducerOptions: ProducerOptions{
+ DisableBatching: true,
+ EnableChunking: true,
+ },
+ },
+ NackRedeliveryDelay: 1 * time.Second,
+ })
+ assert.NoError(t, err)
+ defer consumer.Close()
+
+ // Send a chunked message larger than broker maxMessageSize (5MB, far
exceeding the 1MB broker limit)
+ // If DLQ producer does not enable chunking, sending will fail
+ content := createTestMessagePayload(5 * _brokerMaxMessageSize)
+ _, err = producer.Send(context.Background(), &ProducerMessage{
+ Payload: content,
+ Key: "chunk-dlq-key",
+ Properties: map[string]string{
+ "custom-prop": "custom-value",
+ },
+ })
+ assert.NoError(t, err)
+
+ // Receive and Nack the message maxRedeliveries times
+ for i := 0; i < int(maxRedeliveries); i++ {
+ ctx, cancel := context.WithTimeout(context.Background(),
15*time.Second)
+ msg, err := consumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.Equal(t, content, msg.Payload())
+ assert.Equal(t, "chunk-dlq-key", msg.Key())
+ consumer.Nack(msg)
+ }
+
+ // Message should be routed to DLQ
+ // If DLQ producer does not enable chunking, the large message cannot
be sent and this will timeout
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ dlqMsg, err := dlqConsumer.Receive(ctx)
+ cancel()
+ assert.NoError(t, err)
+ assert.NotNil(t, dlqMsg)
+ assert.Equal(t, content, dlqMsg.Payload())
+ assert.Equal(t, "chunk-dlq-key", dlqMsg.Key())
+
+ // Verify original properties are preserved
+ assert.Equal(t, "custom-value", dlqMsg.Properties()["custom-prop"])
+
+ // Verify DLQ metadata properties
+ assert.NotEmpty(t, dlqMsg.Properties()[SysPropertyRealTopic])
+ assert.Contains(t, dlqMsg.Properties()[SysPropertyRealTopic], topic)
+ assert.NotEmpty(t, dlqMsg.Properties()[SysPropertyOriginMessageID])
+
+ assert.NoError(t, dlqConsumer.Ack(dlqMsg))
+
+ // No more messages on the original consumer
+ ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second)
+ msg, err := consumer.Receive(ctx)
+ cancel()
+ assert.ErrorIs(t, err, context.DeadlineExceeded)
+ assert.Nil(t, msg)
+}
diff --git a/pulsar/producer.go b/pulsar/producer.go
index c8b6b09c..e3ea7c13 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -198,6 +198,18 @@ type ProducerOptions struct {
// EnableChunking controls whether automatic chunking of messages is
enabled for the producer. By default, chunking
// is disabled.
// Chunking can not be enabled when batching is enabled.
+ //
+ // Note: If chunking is enabled and the consumer is configured with
DLQ/RLQ (Dead Letter Queue / Retry Letter
+ // Queue), it is recommended to also enable chunking in the DLQ
policy's ProducerOptions so that large chunked
+ // messages can be forwarded to the DLQ/RLQ topic successfully. Example:
+ // DLQ: &pulsar.DLQPolicy{
+ // MaxDeliveries: 3,
+ // DeadLetterTopic: "my-dlq-topic",
+ // ProducerOptions: pulsar.ProducerOptions{
+ // EnableChunking: true,
+ // DisableBatching: true,
+ // },
+ // }
EnableChunking bool
// ChunkMaxMessageSize is the max size of single chunk payload.
