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

#### 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* `SequenceId`s, 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](https://github.com/apache/pulsar/pull/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

```go
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

Start by running the provided Go reproduction against Pulsar 4.0.5, then trace the Go client's batching and sequence-ID handling for SendAsync. Done means a duplicate SequenceID flushes the current batch before the new message is added, so consuming the topic retains the genuinely new message while dropping only the duplicate.

Written by the indexing model from the issue text.