[Mohith26]

Fixes #1395

Motivation

With batching enabled, the broker deduplicates a batch using only its sequence_id (lowest) and highest_sequence_id. The Go client had two gaps there:

1. It never set highest_sequence_id, so the broker recorded only the first sequence ID of each batch.
2. It let a resent message share a batch with new messages.

Against a standalone 4.2.5 broker with deduplication enabled, that gives:

Sent rounds (each round is one batch)	Stored on main	Expected
[0, 1], then [0 (resend), 2]	0, 1 (2 is lost)	0, 1, 2
[0, 1, 2], then [1 (resend)], then [3]	0, 1, 2, 1 again, 3	0, 1, 2, 3

The first row is the case from the issue (data loss), the second is a duplicate getting through.

Modifications

• batchContainer tracks the highest sequence ID it holds and sets highest_sequence_id on both the batch CommandSend and MessageMetadata (also per key for the key-based builder). FlushBatch and pendingItem carry it.
• partitionProducer tracks the last pushed sequence ID (seeded from ProducerSuccess.last_sequence_id). A batched message whose ID is not above it flushes the current batch and goes into its own batch, and the next message flushes again, so potential duplicates are never mixed with new messages. This is the same rule as the Java client ([Fix-6273] Flush the potential duplicated message when add messages to a batch. pulsar#6326), including its warn/info logs.
• LastSequenceID() now reports the highest acknowledged sequence ID instead of the first ID of the last acknowledged batch, matching the Java client.

Producers that use generated sequence IDs are not affected: IDs only increase, so batches are built exactly as before (covered by a test).

Verifying this change

• pulsar/internal/batch_builder_test.go: decodes flushed frames and checks sequence_id / highest_sequence_id for the default and key-based builders, and that the value resets between batches.
• pulsar/producer_partition_test.go: drives internalSend with a mock connection and checks how messages are split into batches (issue scenario, consecutive resends, resend from the middle of an in-flight batch, generated IDs) and that LastSequenceID() follows acks. Runs without a broker, also with -race.
• pulsar/producer_test.go: TestProducerDeduplicationWithBatchedResends enables deduplication on a fresh namespace and checks what the consumer actually receives for both rows in the table above. It fails on main and passes with this change (run locally against a 4.2.5 standalone).
• go vet and golangci-lint run ./pulsar/... (v2.12.2) are clean; go test ./pulsar/internal/... passes.

This change was prepared with AI assistance (per the ASF generative tooling guidance); all tests listed above were run locally.

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

• Dependencies (does it add or upgrade a dependency): no
• The public API: no (LastSequenceID() semantics now match the Java client for batches)
• The schema: no
• The default values of configurations: no
• The wire protocol: yes, the client now sets the existing optional highest_sequence_id field on batched sends

[nodece]

Thanks for the fix — I verified it locally on the PR branch and the analysis and implementation check out:

• Confirmed master never sets highest_sequence_id anywhere in batch_builder.go, and the isolation rule (flush before and after a potentially-duplicated message) reproduces the expected batch splits, matching the Java ProducerImpl behavior including the warn/info log wording.
• Confirmed newPartitionProducer blocks on grabCnx before runEventsLoop starts, so lastSequenceIDPushed is always seeded (to -1 for a fresh producer) before any internalSend — the zero value can't leak into the duplicate check.
• pendingItem is only constructed in writeData, so highestSequenceID has no missed call site; the key-based builder sets it per key container; internalSingleSend correctly passes (sid, sid).
• New unit tests pass incl. -race; ./pulsar/internal/... passes; go vet clean.
• The LastSequenceID() change also makes the value monotonic (the old code stored pi.sequenceID unconditionally per receipt, so an out-of-order receipt could regress it), and the existing interface doc already describes the new semantics.

Two non-blocking questions:

1. Reconnect seeding — p.lastSequenceIDPushed is only synced from ProducerSuccess.lastSequenceId inside if p.sequenceIDGenerator == nil (grabCnx), i.e. only on the first connect; on reconnects the generator is non-nil so it keeps the local value. The Java client re-syncs lastSequenceIDPushed from the broker on every connection. With user-supplied explicit sequence IDs and a producer name, that means after a reconnect (where e.g. another instance of the same producer name pushed higher IDs) this client's potential-duplicate detection is less conservative than Java's. Was restricting it to the first connect intentional? Moving p.lastSequenceIDPushed = lastSequenceID out of the if (or taking the max with the local value) would match Java.

2. Minor: batchContainer.reset() doesn't clear bc.highestSequenceID; correctness currently relies on the first Add of each batch reassigning it unconditionally (covered by the "next batch must not inherit" assertion). Clearing it in reset() would make that invariant explicit.

[Mohith26]

Thanks for verifying it so thoroughly.

1. Reconnect seeding: restricting it to the first connect was intentional, and I believe it matches Java. In ProducerImpl.connectionOpened the broker's lastSequenceId is only applied under if (this.msgIdGenerator == 0 && conf.getInitialSequenceId() == null) (L2295), with the comment that the broker's value is ignored on subsequent reconnects, and that branch only sets lastSequenceIdPublished and msgIdGenerator. lastSequenceIdPushed is never synced from ProducerSuccess; it only advances in updateLastSeqPushed as ops are pushed (L3007). So on reconnect both clients keep the local value, and on first connect this PR is slightly more conservative than Java by seeding lastSequenceIDPushed from the broker. There is also a practical reason not to set it on reconnect here: grabCnx then runs on the reconnect goroutine while internalSend reads and writes lastSequenceIDPushed on the events loop, so it would need to become atomic. I added a comment in grabCnx explaining this (fa386fb). Happy to do a max-with-broker variant as a follow-up if you'd rather be more conservative than Java there.
2. reset(): done in fa386fb. batchContainer.reset() now clears highestSequenceID, and TestBatchContainerSetsHighestSequenceID asserts it is zero after Flush (that assertion fails without the change).

go test ./pulsar/internal/..., the three producer batching tests with -race, go vet ./pulsar/..., and golangci-lint v2.12.2 on ./pulsar/internal/ pass.