Skip to content

fix(subscriber): propagate handler error to engage startSubscriber backoff #3218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NitinKumar004 wants to merge 2 commits into
base: development
Choose a base branch
from
+6 −2
Open

fix(subscriber): propagate handler error to engage startSubscriber backoff #3218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
567d369
fix(subscriber): propagate handler error to engage startSubscriber ba…
NitinKumar004 May 14, 2026
23bbc15
Merge branch 'development' into fix/subscriber-error-swallowed
Umang01-hash May 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion pkg/gofr/subscriber.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,11 @@ func (s *SubscriptionManager) startSubscriber(ctx context.Context, topic string,
s.container.Logger.Errorf("error in subscription for topic %s: %v", topic, err)

delay = time.Second * 2

continue
}

delay = 0
}
}
}
Expand Down Expand Up @@ -80,7 +84,7 @@ func (s *SubscriptionManager) handleSubscription(ctx context.Context, topic stri
}
if err != nil {
s.container.Logger.Errorf("error in handler for topic %s: %v", topic, err)
return nil
return err
}

if msg.Committer != nil {
Expand Down
2 changes: 1 addition & 1 deletion pkg/gofr/subscriber_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func TestSubscriptionManager_handleSubscription_HandlerErrorDoesNotCommit(t *tes
s := SubscriptionManager{container: c}

err := s.handleSubscription(t.Context(), topicHandleSubErr, func(*Context) error { return errHandlerFail })
require.NoError(t, err)
require.ErrorIs(t, err, errHandlerFail, "handler error must propagate so startSubscriber can back off")
require.NotNil(t, ps.lastCommitter)
require.Zero(t, ps.lastCommitter.n, "handler error must not commit")
}
Loading