queue/kafka: a fenced-out member's late commit can move the new owner's committed offset

[Zaba505]

What happens

When a queue/kafka Service is fenced out of its consumer group, it can still commit the offset it had pending from before the fence. Kafka accepts that commit under the member's new generation. If another member owns the partition by then, the late commit can move the new owner's committed offset backward or forward.

At-least-once still holds:

• Backward means the new owner's records after that offset are processed again, after its next restart or rebalance.
• Forward only skips records the fenced member did process.

So no record is lost. But this is the case the Kafka proposal's revision listed as a pre-reset bug the rewrite fixed: "a late commit could rewind the new owner's". The package doc also says the opposite of what happens against Kafka.

Why

The design assumed that a fenced member's partitions are lost, and the Service stops a lost partition without committing (service.go#L664-L689 skips the commit when p.lost is set).

Against Kafka itself, franz-go learns of the fence from the refused commit (UNKNOWN_MEMBER_ID), not from a heartbeat. It rejoins as a new member, and the partition the member still held comes back through OnPartitionsRevoked, not OnPartitionsLost. Two things then combine:

1. commit treats a commit refused because the member isn't in the group's current generation as "not a failure: franz-go rejoins, p.pending stays pending, and the next commit covers it".
2. The revoke stops the partition, and run commits p.pending on the way out. The member has rejoined by then, so the commit carries the new generation and Kafka accepts it.

kfake only shows the "lost" path, because its tests inject UNKNOWN_MEMBER_ID on a heartbeat, so the unit tests never reach this.

How it was found

#731's container test, TestKafka_a_member_whose_revoke_outlasts_RebalanceTimeout_is_fenced, fences a member against apache/kafka-native. Its PR, #750, records the revoke-instead-of-loss difference and this late commit under "Judgment calls". In that test the late commit equals what the new owner committed, so nothing moved. Moving an offset hasn't been observed, only reasoned from the code and the commit Kafka accepted.

The docs

doc.go#L90-L93 says: "When the member is fenced out of the group, its partitions are lost: another member may own them already, so the Service stops them after their record in flight without committing". Against Kafka, the partition is revoked and committed. The Failures table's row for a fenced-out member ("the records are committed by a later commit, or processed again by the partition's new owner") describes what the code does, but not that the later commit can belong to a partition this member no longer owns.

Expected

A commit refused because the member was fenced out should end that member's claim on what it had pending. Its new owner, if there is one, decides the partition's committed offset. Some ways to get there:

• on a refused commit, drop p.pending for that partition, or mark the partition as lost, so neither the revoke nor a later batch commits it;
• or record the group generation each partition was assigned under, and commit only while the client is still in that generation.

Either way, a container test should show the new owner's committed offset is unchanged after the fenced member's revoke, and the doc's fencing paragraph and Failures row should say what happens against Kafka.

None of this changes queue's or queue/kafka's API, Delivery's guarantees, or how the Service runs on humus.Main.

References

• Proposal: queue processing and the Kafka runtime, and its revision
• story(kafka): consume Kafka topics as a humus Service, at least or at most once #728, the Service (story(kafka): consume Kafka topics as a humus Service, at least or at most once #747)
• story(kafka): test the consumer against Kafka in a container, in a CI job of its own #731, the container tests (story(kafka): test the consumer against Kafka in a container, in a CI job of its own #750)

[Zaba505]

Triage

Verdict: this is a reproducible bug in queue/kafka. It isn't a misuse of humus, and it isn't a feature request. It's also worse than reported: it breaks AtMostOnce too, and a record can be processed twice. #758 is filed to fix it.

Is this a misuse of humus?

No. The end user picks a Delivery, sources the Settings and writes the Processor. Commits, fences and rebalances belong to the Service. The Processor can't see the group's membership or its commits, so the end user has no way to guard against this.

The trigger in #750's test is a revoke that outlasts RebalanceTimeout, and doc.go says the revoke "has to fit in RebalanceTimeout". So you could argue that a slow Processor is the end user's problem. But doc.go also promises what happens when a member is fenced anyway: the Service stops its partitions "without committing". Fences are expected and documented, so the Service has to behave as documented when one happens. Some fences have nothing to do with the Processor, either. Take a member cut off from the coordinator for longer than SessionTimeout. If its next commit reaches Kafka before its next heartbeat, franz-go takes the same path: it rejoins as a new member and reports the partitions as revoked. I got that from franz-go's code and didn't reproduce it.

Under AtLeastOnce, you could argue that a rewind only causes duplicates, which the contract allows. But nothing the end user configures bounds how far back it goes. The committed offset returns to wherever the fenced member stopped, however far the new owner has got since then. And #683 listed a late commit that rewinds the new owner's offset as a pre-reset bug that the rewrite fixed. Under AtMostOnce, the end user's Processor doesn't need to be idempotent, because doc.go promises that "a record is never processed twice".

Is it a feature request?

No. What happens contradicts three things:

• Proposal: queue processing and the Kafka runtime #683's design: "On loss … stops the partitions without committing";
• doc.go's "Revocation and loss" paragraph;
• AtMostOnce's guarantee.

The fix doesn't change the API.

Is it reproducible?

Yes. I wrote two container tests based on TestKafka_a_member_whose_revoke_outlasts_RebalanceTimeout_is_fenced, and ran them against apache/kafka-native:4.3.1 with franz-go v1.22.2, at 4289e89. Each test failed in 4 runs out of 4.

AtLeastOnce: the committed offset moves backward. After the fence, the test produced 10 more records to the partition the fenced member still held. Its new owner processed them and committed offset 11. Then the test released the fenced member's record in flight.

1. The fenced member's commit was refused with UNKNOWN_MEMBER_ID. Its partition goroutine then took the batch waiting behind it and processed offset 1.
2. franz-go rejoined as a new member (join returned UnknownMemberID, rejoining without a member id). Its next session reported the partition as lost from the assignment diff, and franz-go called OnPartitionsRevoked.
3. The revoke committed offset 2 under the new member ID, and Kafka accepted it (updated committed 0{0=>2}).
4. The group's committed offset went from 11 back to 2. A fresh member then processed offsets 2 to 10 again.

AtMostOnce: records are processed twice. This used the same fence, and the new owner processed the same 10 records. This time, the test released the moved partition's record first, so franz-go began to rejoin. Then it released the kept partition's record, once franz-go had logged blocking commits from join&sync.

1. The Service called the commit for the batch of offsets 1 to 10. The commit waited for the join and sync, then went out under the new member ID, even though that member's sync had assigned it nothing (synced assigned:, then issuing commit 0:{0 11} and updated committed 0{1=>11}).
2. Kafka didn't refuse the commit, so the Service processed the batch. Both the fenced member and the new owner processed offsets 1 to 10.
3. No commit was refused before this one, so nothing in the Service saw the fence.

So two lines of the report need correcting. AtMostOnce's guarantee doesn't hold. And a moved offset has now been observed.

Why

The report has the first half: franz-go learns of the fence and reports the partitions as revoked, not lost. Three more facts complete it:

• Kafka's classic group protocol checks a commit's member ID and generation, but not whether that member is assigned the partition. A fenced member rejoins under a new member ID, so Kafka accepts every commit it sends after the rejoin, for any partition.
• franz-go fills in the member ID when it sends the commit, not when CommitRecords is called. A commit called during a join and sync waits for it to finish, then picks up the new member ID.
• franz-go reports the old member's partitions through OnPartitionsRevoked, so the Service can't tell this case from a cooperative revoke. That may be worth raising with franz-go, but the fix below doesn't depend on it.

On the options under Expected

• Dropping p.pending, or marking the partition lost, when a commit is refused isn't enough. In the AtLeastOnce run, the partition processed offset 1 after its commit was refused, so it had a record pending again when the revoke came. In the AtMostOnce run, no commit was refused at all.
• The generation is the wrong key. Under cooperative-sticky, every rebalance moves the generation on, while a partition the member keeps gets no new OnPartitionsAssigned entry. A partition tied to the generation it was assigned under would stop committing after any rebalance. A fence changes the member ID.
• The check has to run when franz-go sends the commit, because of that wait for the join and sync.

I prototyped a fix on 4289e89 that works:

• Each partition records the member ID it's held under. OnPartitionsAssigned sets it from cl.GroupMetadata(), for every partition the Service still holds. franz-go calls that callback at every session, "even if nothing new is assigned", so a member that gets a partition back after a fence holds it under its new member ID.
• commitRecord commits through kgo.PreCommitFnContext. Its function sees the OffsetCommitRequest that franz-go is about to send, and cancels the commit when req.MemberID isn't the partition's member ID.
• notMember treats the cancelled commit like a refused one. Under AtLeastOnce the record stays pending, and under AtMostOnce the batch is skipped and logged.

With the prototype, both reproductions pass. The committed offset stays at 11, and the fenced member processes nothing more of the partition it kept. The five existing container tests pass, twice each, and the unit tests pass under -race, with 95.3% coverage.

The prototype (service.go, against 4289e89)

diff --git a/queue/kafka/service.go b/queue/kafka/service.go
index 06265217..f1e17d28 100644
--- a/queue/kafka/service.go
+++ b/queue/kafka/service.go
@@ -19,4 +19,5 @@ import (
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/twmb/franz-go/pkg/kmsg"
 	"github.com/twmb/franz-go/plugin/kotel"
 	"go.opentelemetry.io/otel/attribute"
@@ -306,4 +307,9 @@ type partition struct {
 	lost atomic.Bool
 
+	// member is the member ID the partition is held under, set by every
+	// OnPartitionsAssigned while the partition is held. A commit franz-go
+	// would send under another member ID is not sent.
+	member atomic.Pointer[string]
+
 	// pending is the last record run processed and has not committed. Only
 	// run uses it.
@@ -459,5 +465,6 @@ func refused(err error) bool {
 // franz-go rejoins the group on it, so it is not the Service's failure.
 func notMember(err error) bool {
-	return errors.Is(err, kerr.UnknownMemberID) ||
+	return errors.Is(err, errNotHeld) ||
+		errors.Is(err, kerr.UnknownMemberID) ||
 		errors.Is(err, kerr.IllegalGeneration) ||
 		errors.Is(err, kerr.FencedInstanceID) ||
@@ -465,4 +472,9 @@ func notMember(err error) bool {
 }
 
+// errNotHeld is a commit not sent because franz-go would send it under a
+// member ID the partition is not held under: the member was fenced out and
+// rejoined as a new member.
+var errNotHeld = errors.New("the partition is not held under the group's current member ID")
+
 // fail records err as the error Serve returns, if it is the first, and
 // stops polling, so the Service stops as on shutdown.
@@ -491,4 +503,5 @@ func (g *group) partition(tp topicPartition) *partition {
 func (g *group) assigned(_ context.Context, cl *kgo.Client, assigned map[string][]int32) {
 	g.logPartitions("partitions assigned", assigned)
+	member, _ := cl.GroupMetadata()
 	g.mu.Lock()
 	defer g.mu.Unlock()
@@ -496,4 +509,7 @@ func (g *group) assigned(_ context.Context, cl *kgo.Client, assigned map[string]
 		return
 	}
+	for _, p := range g.parts {
+		p.member.Store(&member)
+	}
 	for topic, partitions := range assigned {
 		for _, id := range partitions {
@@ -513,4 +529,5 @@ func (g *group) assigned(_ context.Context, cl *kgo.Client, assigned map[string]
 				},
 			}
+			p.member.Store(&member)
 			g.parts[tp] = p
 			go g.run(cl, p)
@@ -691,5 +708,11 @@ func (g *group) commit(cl *kgo.Client, p *partition) bool {
 func (g *group) commitRecord(cl *kgo.Client, p *partition, r *kgo.Record) error {
 	start := time.Now()
-	err := cl.CommitRecords(g.work, r)
+	ctx := kgo.PreCommitFnContext(g.work, func(req *kmsg.OffsetCommitRequest) error {
+		if m := p.member.Load(); m == nil || req.MemberID != *m {
+			return errNotHeld
+		}
+		return nil
+	})
+	err := cl.CommitRecords(ctx, r)
 	attrs := append(p.attrs[:len(p.attrs):len(p.attrs)], semconv.MessagingOperationTypeSettle)
 	if err != nil {

#758 has the acceptance criteria: the guard, container tests for both deliveries, and corrections to doc.go's fencing paragraph and its Failures row.

[Zaba505]

Fixed by #766, merged as def58a7.

What changed: each partition now records the member ID it is held under: that of the session that assigned it, and of each later session that kept it (OnPartitionsAssigned re-stamps every held partition). Every commit goes through kgo.PreCommitFnContext, which cancels the commit when franz-go would send it under any other member ID. The check runs when franz-go sends the commit, so a commit called during a join and sync is checked against the member ID the join returned. The Service treats a commit it doesn't send like a refused one. Under AtLeastOnce the record stays pending, and nothing more is committed once the partition is revoked or lost. Under AtMostOnce the batch is not processed, and it is logged at WARN with its offsets.

So a fenced member that rejoins as a new member commits nothing more for the partitions it held, whether franz-go reports them as revoked or as lost. Their new owner decides the committed offset. Both reproductions from the triage are now container tests (TestKafka_at_least_once_a_fenced_member_does_not_rewind_the_new_owners_commit and TestKafka_at_most_once_a_fenced_member_does_not_commit_or_process_a_batch_polled_while_it_rejoins), alongside a kfake unit test. Without the guard they fail as reported here: the committed offset goes from 11 to 2, and the AtMostOnce batch goes out under the new member ID. doc.go's fencing paragraph and Failures row, and the docs site's Kafka page, now describe this.