[Issue 1446][Consumer] Fix consumer can't consume resent chunked messages (#1464)

Master Issue: #1446
related issue apache/pulsar#21070 and apache/pulsar#21101

### Motivation
Current, when the producer resend the chunked message like this:
```
M1: UUID: 0, ChunkID: 0
M2: UUID: 0, ChunkID: 0 // Resend the first chunk
M3: UUID: 0, ChunkID: 1
```
When the consumer received the M2, it will find that it's already tracking the UUID:0 chunked messages, and will then discard the message M1 and M2. This will lead to unable to consume the whole chunked message even though it's already persisted in the Pulsar topic.

Here is the code logic:
```Go
	if ctx == nil || ctx.chunkedMsgBuffer == nil || chunkID != ctx.lastChunkedMsgID+1 {
		lastChunkedMsgID := -1
		totalChunks := -1
		if ctx != nil {
			lastChunkedMsgID = int(ctx.lastChunkedMsgID)
			totalChunks = int(ctx.totalChunks)
			ctx.chunkedMsgBuffer.Clear()
		}
		pc.log.Warnf(fmt.Sprintf(
			"Received unexpected chunk messageId %s, last-chunk-id %d, chunkId = %d, total-chunks %d",
			msgID.String(), lastChunkedMsgID, chunkID, totalChunks))
		pc.chunkedMsgCtxMap.remove(uuid)
		pc.availablePermits.inc()
		return nil
	}
```
The bug can be easily reproduced using the testcase `TestChunkWithReconnection` and `TestResendChunkMessages` introduced by this PR.

### Modifications
The current chunk processing strategy is consistent with the behavior of the Java client:
https://github.com/apache/pulsar/blob/52a4d5ee84fad6af2736376a6fcdd1bc41e7c52f/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1579

When receiving the new duplicated first chunk of a chunked message, the consumer discard the current chunked message context and create a new context to track the following messages. For the case mentioned in Motivation, the M1 will be released and the consumer will assemble M2 and M3 as the chunked message.

Author: geniusjoe

pulsar/consumer_partition.go

@@ -1461,35 +1461,161 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 		partitionIdx: pc.partitionIdx,
 	}
 
-	if msgMeta.GetChunkId() == 0 {
+	if msgMeta.GetChunkId() != msgMeta.GetNumChunksFromMsg()-1 {
+		pc.availablePermits.inc()
+	}
+	if chunkID == 0 {
+		// Handle ack hole case when receive duplicated chunks.
+		// There are two situation that receives chunks with the same sequence ID and chunk ID.
+		// Situation 1 - Message redeliver:
+		// For example:
+		//     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+		//     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+		//     Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:1
+		//     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:2
+		//     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:3
+		// In this case, chunk-3 and chunk-4 have the same msgID with chunk-1 and chunk-2.
+		// This may be caused by message redeliver, we can't ack any chunk in this case here.
+		// Situation 2 - Corrupted chunk message
+		// For example:
+		//     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+		//     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+		//     Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3
+		//     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+		//     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+		// In this case, all the chunks with different msgIDs and are persistent in the topic.
+		// But Chunk-1 and Chunk-2 belong to a corrupted chunk message that must be skipped since
+		// they will not be delivered to end users. So we should ack them here to avoid ack hole.
+		ctx := pc.chunkedMsgCtxMap.get(uuid)
+		if ctx != nil {
+			isCorruptedChunkMessageDetected := true
+			for _, previousChunkMsgID := range ctx.chunkedMsgIDs {
+				if previousChunkMsgID == nil {
+					continue
+				}
+				if previousChunkMsgID.equal(msgID) {
+					isCorruptedChunkMessageDetected = false
+					break
+				}
+			}
+			if isCorruptedChunkMessageDetected {
+				ctx.discard(pc)
+			}
+			// The first chunk of a new chunked-message received
+			// before receiving other chunks of previous chunked-message
+			// so, remove previous chunked-message from map and release buffer
+			pc.log.Warnf(fmt.Sprintf(
+				"[%s] [%s] Receive a duplicated chunk id=0 message with messageId [%s], sequenceId [%d], "+
+					"uuid [%s]. Remove previous chunk context with lastChunkedMsgID [%d]",
+				pc.name,
+				pc.options.subscription,
+				msgID.String(),
+				msgMeta.GetSequenceId(),
+				msgMeta.GetUuid(),
+				ctx.lastChunkedMsgID,
+			))
+			ctx.chunkedMsgBuffer.Clear()
+			pc.chunkedMsgCtxMap.remove(uuid)
+		}
 		pc.chunkedMsgCtxMap.addIfAbsent(uuid,
 			numChunks,
 			totalChunksSize,
 		)
 	}
 
+	// discard message if chunk is out-of-order
 	ctx := pc.chunkedMsgCtxMap.get(uuid)
-
 	if ctx == nil || ctx.chunkedMsgBuffer == nil || chunkID != ctx.lastChunkedMsgID+1 {
+		// Filter and ack duplicated chunks instead of discard ctx.
+		// For example:
+		//     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+		//     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+		//     Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
+		//     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
+		//     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
+		//     Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6
+		// We should filter and ack chunk-4 and chunk-5.
+		if ctx != nil && chunkID <= ctx.lastChunkedMsgID {
+			pc.log.Warnf(fmt.Sprintf(
+				"[%s] [%s] Receive a duplicated chunk message with messageId [%s], "+
+					"last-chunk-Id [%d], chunkId [%d], sequenceId [%d], uuid [%s]",
+				pc.name,
+				pc.options.subscription,
+				msgID.String(),
+				ctx.lastChunkedMsgID,
+				chunkID,
+				msgMeta.GetSequenceId(),
+				msgMeta.GetUuid(),
+			))
+			// Just like the above logic of receiving the first chunk again.
+			// We only ack this chunk in the message duplication case.
+			isCorruptedChunkMessageDetected := true
+			for _, previousChunkMsgID := range ctx.chunkedMsgIDs {
+				if previousChunkMsgID == nil {
+					continue
+				}
+				if previousChunkMsgID.equal(msgID) {
+					isCorruptedChunkMessageDetected = false
+					break
+				}
+			}
+			if isCorruptedChunkMessageDetected {
+				pc.AckID(toTrackingMessageID(msgID))
+			}
+			return nil
+		}
+		// Chunked messages rely on TCP to ensure that chunk IDs are strictly increasing within a partition.
+		// If the current chunk ID is greater than ctx.lastChunkedMsgID + 1,
+		// it indicates that the current chunk is corrupted and may require resource cleanup.
 		lastChunkedMsgID := -1
 		totalChunks := -1
 		if ctx != nil {
 			lastChunkedMsgID = int(ctx.lastChunkedMsgID)
 			totalChunks = int(ctx.totalChunks)
-			ctx.chunkedMsgBuffer.Clear()
 		}
 		pc.log.Warnf(fmt.Sprintf(
-			"Received unexpected chunk messageId %s, last-chunk-id %d, chunkId = %d, total-chunks %d",
-			msgID.String(), lastChunkedMsgID, chunkID, totalChunks))
-		pc.chunkedMsgCtxMap.remove(uuid)
-		pc.availablePermits.inc()
+			"[%s] [%s] Received unexpected chunk messageId [%s], last-chunk-id [%d], "+
+				"chunkId = [%d], total-chunks [%d], sequenceId [%d], uuid [%s]",
+			pc.Topic(),
+			pc.options.subscription,
+			msgID.String(),
+			lastChunkedMsgID,
+			chunkID,
+			totalChunks,
+			msgMeta.GetSequenceId(),
+			msgMeta.GetUuid()),
+		)
+		if ctx != nil {
+			ctx.chunkedMsgBuffer.Clear()
+			pc.chunkedMsgCtxMap.remove(uuid)
+		}
+		// Consider a scenario where MaxPendingChunkedMessage is set to 1,
+		// and we have two messages (A and B), each consisting of three chunks:
+		// A chunks are Chunk-1, Chunk-2, Chunk-6 and B chunks are Chunk-3, Chunk-4, Chunk-5
+		// The consumer receives them in the following order:
+		//     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
+		//     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
+		//     since MaxPendingChunkedMessage is 1, the context for A is removed
+		//     Chunk-3 sequence ID: 1, chunk ID: 0, msgID: 1:3
+		//     Chunk-4 sequence ID: 1, chunk ID: 1, msgID: 1:4
+		//     Chunk-5 sequence ID: 1, chunk ID: 2, msgID: 1:5
+		//     Chunk-6 sequence ID: 0, chunk ID: 2, msgID: 1:6
+		// If we acknowledge Chunk-6 here, message A would be lost.
+		// This is unexpected, as the user would expect A to be successfully consumed after redelivery.
+		// So the correct logic should be:
+		// If AutoAckIncompleteChunk is true, then acknowledge the message.
+		// Otherwise, do nothing so that the message can be redelivered in the future.
+		if pc.options.autoAckIncompleteChunk {
+			pc.AckID(toTrackingMessageID(msgID))
+		}
 		return nil
 	}
 
+	// The chunk ID meets the expected value,
+	// so we add the current chunk to the corresponding chunkedMsgCtx.
 	ctx.append(chunkID, msgID, compressedPayload)
 
 	if msgMeta.GetChunkId() != msgMeta.GetNumChunksFromMsg()-1 {
-		pc.availablePermits.inc()
 		return nil
 	}
 
@@ -2518,6 +2644,7 @@ func (c *chunkedMsgCtxMap) discardOldestChunkMessage(autoAck bool) {
 	if autoAck {
 		ctx.discard(c.pc)
 	}
+	ctx.chunkedMsgBuffer.Clear()
 	delete(c.chunkedMsgCtxs, oldest)
 	c.pc.log.Infof("Chunked message [%s] has been removed from chunkedMsgCtxMap", oldest)
 }
@@ -2535,6 +2662,7 @@ func (c *chunkedMsgCtxMap) discardChunkMessage(uuid string, autoAck bool) {
 	if autoAck {
 		ctx.discard(c.pc)
 	}
+	ctx.chunkedMsgBuffer.Clear()
 	delete(c.chunkedMsgCtxs, uuid)
 	e := c.pendingQueue.Front()
 	for ; e != nil; e = e.Next() {

pulsar/message_chunking_test.go

@@ -21,14 +21,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"math/rand"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/log"
 
 	"google.golang.org/protobuf/proto"
 
@@ -578,3 +581,167 @@ func sendSingleChunk(p Producer, uuid string, chunkID int, totalChunks int) {
 		uint32(internal.MaxMessageSize),
 	)
 }
+
+func TestChunkWithReconnection(t *testing.T) {
+	sLogger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	client, err := NewClient(ClientOptions{
+		URL:    lookupURL,
+		Logger: log.NewLoggerWithSlog(sLogger),
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:               topic,
+		DisableBatching:     true,
+		EnableChunking:      true,
+		ChunkMaxMessageSize: 100,
+		MaxPendingMessages:  200000,
+		SendTimeout:         60 * time.Second,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+
+	c, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		Type:             Exclusive,
+		SubscriptionName: "chunk-subscriber",
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, c)
+	defer c.Close()
+
+	// Reduce publish rate to prevent the producer sending messages too fast
+	url := adminURL + "/" + "admin/v2/persistent/public/default/" + topic + "/publishRate"
+	makeHTTPCall(t, http.MethodPost, url, "{\"publishThrottlingRateInMsg\": 1,\"publishThrottlingRateInByte\": 100}")
+	// Need to wait some time to let the rate limiter take effect
+	time.Sleep(2 * time.Second)
+
+	// payload/ChunkMaxMessageSize = 1000/100 = 10 msg, and publishThrottlingRateInMsg = 1
+	// so that this chunk msg will send finish after 10 seconds
+	producer.SendAsync(context.Background(), &ProducerMessage{
+		Payload: createTestMessagePayload(1000),
+	}, func(_ MessageID, _ *ProducerMessage, err error) {
+		assert.Nil(t, err)
+	})
+	assert.NoError(t, err)
+
+	time.Sleep(5 * time.Second)
+	//	trigger topic unload to test sending chunk msg with reconnection
+	url = adminURL + "/" + "admin/v2/persistent/public/default/" + topic + "/unload"
+	makeHTTPCall(t, http.MethodPut, url, "")
+	// Need to wait some time to receive all chunk messages
+	time.Sleep(10 * time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	msg, err := c.Receive(ctx)
+	cancel()
+	assert.NoError(t, err)
+	assert.NotNil(t, msg.ID())
+}
+
+func TestResendChunkMessages(t *testing.T) {
+	sLogger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	client, err := NewClient(ClientOptions{
+		URL:    lookupURL,
+		Logger: log.NewLoggerWithSlog(sLogger),
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:               topic,
+		DisableBatching:     true,
+		EnableChunking:      true,
+		ChunkMaxMessageSize: 100,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+
+	c, err := client.Subscribe(ConsumerOptions{
+		Topic:                    topic,
+		Type:                     Exclusive,
+		SubscriptionName:         "chunk-subscriber",
+		MaxPendingChunkedMessage: 10,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, c)
+	defer c.Close()
+
+	sendSingleChunk(producer, "0", 0, 2)
+	sendSingleChunk(producer, "0", 0, 2) // Resending the first chunk
+	sendSingleChunk(producer, "1", 0, 3) // This is for testing the interwoven chunked message
+	sendSingleChunk(producer, "1", 1, 3)
+	sendSingleChunk(producer, "1", 0, 3) // Resending the UUID-1 chunked message
+	sendSingleChunk(producer, "0", 1, 2)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	msg, err := c.Receive(ctx)
+	cancel()
+	assert.NoError(t, err)
+	assert.Equal(t, "chunk-0-0|chunk-0-1|", string(msg.Payload()))
+	c.Ack(msg)
+
+	sendSingleChunk(producer, "1", 1, 3)
+	sendSingleChunk(producer, "1", 2, 3)
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	msg, err = c.Receive(ctx)
+	cancel()
+	assert.NoError(t, err)
+	assert.Equal(t, "chunk-1-0|chunk-1-1|chunk-1-2|", string(msg.Payload()))
+	c.Ack(msg)
+}
+
+func TestResendChunkWithAckHoleMessages(t *testing.T) {
+	sLogger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	client, err := NewClient(ClientOptions{
+		URL:    lookupURL,
+		Logger: log.NewLoggerWithSlog(sLogger),
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:               topic,
+		DisableBatching:     true,
+		EnableChunking:      true,
+		ChunkMaxMessageSize: 100,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+
+	c, err := client.Subscribe(ConsumerOptions{
+		Topic:                    topic,
+		Type:                     Exclusive,
+		SubscriptionName:         "chunk-subscriber",
+		MaxPendingChunkedMessage: 10,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, c)
+	defer c.Close()
+
+	sendSingleChunk(producer, "0", 0, 4)
+	sendSingleChunk(producer, "0", 1, 4)
+	sendSingleChunk(producer, "0", 2, 4)
+	sendSingleChunk(producer, "0", 1, 4) // Resending previous chunk
+	sendSingleChunk(producer, "0", 2, 4)
+	sendSingleChunk(producer, "0", 3, 4)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	msg, err := c.Receive(ctx)
+	cancel()
+	assert.NoError(t, err)
+	assert.Equal(t, "chunk-0-0|chunk-0-1|chunk-0-2|chunk-0-3|", string(msg.Payload()))
+	c.Ack(msg)
+
+	sendSingleChunk(producer, "1", 0, 4)
+	sendSingleChunk(producer, "1", 1, 4)
+	sendSingleChunk(producer, "1", 4, 4) // send broken chunk
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	msg, err = c.Receive(ctx)
+	cancel()
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+}