Deduplication fails when the batch message contains duplicate message and valid message

[hegelwin]

Expected behavior

When a producer resends a duplicate SequenceID while batching is enabled, the client should flush the current batch and start a new one.
This guarantees that a batch never contains both old/duplicate and new SequenceIds, allowing the broker‑side deduplication logic to drop only the duplicates and accept the genuinely new messages.

This is the behavior of the Java client since PR apache/pulsar#6326.

Actual behavior

With the Go client (v0.15.0) the duplicate and new messages are added to the same batch.
Because the outer CommandSend carries only the lowest SequenceId in the batch, the broker cannot determine that newer messages are present; it treats the entire batch as a duplicate and drops it — including the fresh messages.

Steps to reproduce

package main

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
)

func main() {
	// ---------- set up ----------
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar://localhost:6650",
		OperationTimeout:  5 * time.Second,
		ConnectionTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalf("client: %v", err)
	}
	defer client.Close()

	const topic = "persistent://public/default/dedupe-go-batch-bug"
	const producerName = "go-dedupe-test"

	prod, err := client.CreateProducer(pulsar.ProducerOptions{
		Topic:                   topic,
		Name:                    producerName,
		DisableBatching:         false,
		BatchingMaxPublishDelay: 1 * time.Second,
	})
	if err != nil {
		log.Fatalf("producer: %v", err)
	}
	defer prod.Close()

	// helper to send one async msg and wait for ack
	send := func(seq int64, wg *sync.WaitGroup) {
		wg.Add(1)
		prod.SendAsync(context.Background(),
			&pulsar.ProducerMessage{
				Payload:    []byte(fmt.Sprintf("seq-%d", seq)),
				SequenceID: &seq,
			},
			func(id pulsar.MessageID, _ *pulsar.ProducerMessage, err error) {
				log.Printf("Send cb seq=%d err=%v", seq, err)
				wg.Done()
			})
	}

	// ---------- 1) first batch: seq 0,1 ----------
	var wg sync.WaitGroup
	send(0, &wg)
	send(1, &wg)
	wg.Wait()                   // ensure broker acks first batch
	time.Sleep(1 * time.Second) // let batch window expire

	// ---------- 2) second batch with duplicate + new ----------
	//   we issue them back‑to‑back so they land in the SAME batch
	send(0, &wg) // duplicate
	send(2, &wg) // new
	wg.Wait()
	prod.Flush()

	// ---------- consume to see what really persisted ----------
	cons, _ := client.Subscribe(pulsar.ConsumerOptions{
		Topic:                       topic,
		SubscriptionName:            "check",
		Type:                        pulsar.Exclusive,
		SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
	})
	defer cons.Close()

	got := map[string]struct{}{}
	for {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		msg, err := cons.Receive(ctx)
		cancel()
		if err != nil {
			break
		}
		got[string(msg.Payload())] = struct{}{}
		cons.Ack(msg)
	}

	fmt.Println("messages stored:")
	for k := range got {
		fmt.Println("  ", k)
	}
}

System configuration

Pulsar version: 4.0.5