You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/10/20 04:00:36 UTC
(pulsar-client-go) branch fix-1541-waitforexclusive-stale-topic-epoch created (now 0a946574)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a change to branch fix-1541-waitforexclusive-stale-topic-epoch
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
at 0a946574 [fix][producer] Recover WaitForExclusive producer fenced by stale topic epoch
This branch includes the following new commits:
new 0a946574 [fix][producer] Recover WaitForExclusive producer fenced by stale topic epoch
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(pulsar-client-go) 01/01: [fix][producer] Recover WaitForExclusive producer fenced by stale topic epoch
Posted by zi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch fix-1541-waitforexclusive-stale-topic-epoch
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
commit 0a946574365229c53a3ab6c709f30c27ece6b531
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Oct 20 12:00:23 2026 +0800
[fix][producer] Recover WaitForExclusive producer fenced by stale topic epoch
A WaitForExclusive producer permanently stopped reconnecting after a broker
restart. When the exclusive ownership changed while the producer was
disconnected (e.g. a queued producer was promoted during the broker graceful
shutdown), the topic epoch moved forward, and the broker rejected the
reconnection with "ProducerFenced: Topic epoch has already moved". The client
treated every fenced error as fatal, failing pending messages and closing the
producer.
Drop the stale topic epoch and retry in that case: the broker either grants
the ownership back with a new epoch or queues the producer behind the current
owner, which is the WaitForExclusive semantics. Other fenced errors, such as
being fenced out by an ExclusiveWithFencing producer, remain fatal, consistent
with the Java client.
Also only track the topic epoch when the broker actually assigns one in the
ProducerSuccess response, instead of fabricating epoch 0, matching the
Optional semantics of the Java client.
Fixes #1541
Signed-off-by: Zixuan Liu <no...@gmail.com>
---
pulsar/producer_partition.go | 33 ++++++++++++---
pulsar/producer_test.go | 95 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 123 insertions(+), 5 deletions(-)
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 833e9f4b..9d00e08f 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -79,6 +79,7 @@ const (
errMsgTopicTerminated = "TopicTerminatedError"
errMsgProducerBlockedQuotaExceededException = "ProducerBlockedQuotaExceededException"
errMsgProducerFenced = "ProducerFenced"
+ errMsgTopicEpochMoved = "Topic epoch has already moved"
)
func init() {
@@ -346,8 +347,12 @@ func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
}
p.producerName = res.Response.ProducerSuccess.GetProducerName()
- nextTopicEpoch := res.Response.ProducerSuccess.GetTopicEpoch()
- p.topicEpoch = &nextTopicEpoch
+ if res.Response.ProducerSuccess.TopicEpoch != nil {
+ p.topicEpoch = proto.Uint64(res.Response.ProducerSuccess.GetTopicEpoch())
+ } else {
+ // Drop any previously tracked epoch if the broker did not assign one.
+ p.topicEpoch = nil
+ }
if p.options.Encryption != nil {
p.encryptor = internalcrypto.NewProducerEncryptor(p.options.Encryption.Keys,
@@ -530,9 +535,15 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
}
if strings.Contains(errMsg, errMsgProducerFenced) {
- p.log.Warn("Producer was fenced, failing pending messages, stop reconnecting")
- p.doClose(errors.Join(ErrProducerFenced, err))
- return struct{}{}, nil
+ if p.canRecoverFromFenced(errMsg) {
+ p.log.Warn("Producer was fenced because the topic epoch has already moved, " +
+ "dropping the stale topic epoch and retrying")
+ p.topicEpoch = nil
+ } else {
+ p.log.Warn("Producer was fenced, failing pending messages, stop reconnecting")
+ p.doClose(errors.Join(ErrProducerFenced, err))
+ return struct{}{}, nil
+ }
}
p.log.WithError(err).Warn("Failed to reconnect to broker, will retry later.")
@@ -556,6 +567,18 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
})
}
+// canRecoverFromFenced reports whether a ProducerFenced failure received while
+// reconnecting can be recovered from by dropping the stale topic epoch and
+// retrying, instead of permanently fencing the producer. This only applies to
+// WaitForExclusive producers rejected because the topic epoch has already
+// moved forward: on the next attempt the broker either grants the ownership
+// back with a new epoch or queues the producer behind the current owner.
+func (p *partitionProducer) canRecoverFromFenced(errMsg string) bool {
+ return p.options.ProducerAccessMode == ProducerAccessModeWaitForExclusive &&
+ p.topicEpoch != nil &&
+ strings.Contains(errMsg, errMsgTopicEpochMoved)
+}
+
func (p *partitionProducer) runEventsLoop() {
go func() {
for {
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 23310ac6..fe073fac 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -2075,6 +2075,101 @@ func TestWaitForExclusiveProducer(t *testing.T) {
wg.Wait()
}
+func TestCanRecoverFromFenced(t *testing.T) {
+ fencedWithEpochMoved := "server error: ProducerFenced: " +
+ "org.apache.pulsar.broker.service.BrokerServiceException$ProducerFencedException: " +
+ "Topic epoch has already moved. Current epoch: 86, Producer epoch: 85"
+
+ tests := []struct {
+ name string
+ accessMode ProducerAccessMode
+ topicEpoch *uint64
+ errMsg string
+ expected bool
+ }{
+ {
+ name: "wait for exclusive producer with stale topic epoch",
+ accessMode: ProducerAccessModeWaitForExclusive,
+ topicEpoch: proto.Uint64(85),
+ errMsg: fencedWithEpochMoved,
+ expected: true,
+ },
+ {
+ name: "exclusive producer is not recoverable",
+ accessMode: ProducerAccessModeExclusive,
+ topicEpoch: proto.Uint64(85),
+ errMsg: fencedWithEpochMoved,
+ expected: false,
+ },
+ {
+ name: "no topic epoch tracked",
+ accessMode: ProducerAccessModeWaitForExclusive,
+ topicEpoch: nil,
+ errMsg: fencedWithEpochMoved,
+ expected: false,
+ },
+ {
+ name: "fenced for another reason",
+ accessMode: ProducerAccessModeWaitForExclusive,
+ topicEpoch: proto.Uint64(85),
+ errMsg: "server error: ProducerFenced: Fenced out",
+ expected: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ p := &partitionProducer{
+ options: &ProducerOptions{ProducerAccessMode: tt.accessMode},
+ topicEpoch: tt.topicEpoch,
+ }
+ assert.Equal(t, tt.expected, p.canRecoverFromFenced(tt.errMsg))
+ })
+ }
+}
+
+func TestWaitForExclusiveProducerRecoverFromFencedTopicEpoch(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: serviceURL,
+ })
+ require.NoError(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+
+ // Advance the topic epoch by letting a producer own the topic and close.
+ first, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ ProducerAccessMode: ProducerAccessModeWaitForExclusive,
+ })
+ require.NoError(t, err)
+ first.Close()
+
+ second, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ ProducerAccessMode: ProducerAccessModeWaitForExclusive,
+ })
+ require.NoError(t, err)
+ defer second.Close()
+
+ pp := second.(*producer).getProducer(0).(*partitionProducer)
+
+ // Forge a stale topic epoch, as if the epoch had moved forward while the
+ // producer was disconnected (e.g. after a broker restart), then force a
+ // reconnection which the broker rejects with ProducerFenced.
+ pp.topicEpoch = proto.Uint64(0)
+ pp._getConn().Close()
+
+ var msgID MessageID
+ require.Eventually(t, func() bool {
+ msgID, err = second.Send(context.Background(), &ProducerMessage{
+ Payload: []byte("test"),
+ })
+ return err == nil
+ }, 30*time.Second, 100*time.Millisecond)
+ assert.NotNil(t, msgID)
+}
+
func TestMemLimitRejectProducerMessages(t *testing.T) {
c, err := NewClient(ClientOptions{