fix: return handler error from subscriber instead of swallowing it

handleSubscription was logging handler errors but returning nil,
preventing the retry logic in startSubscriber from triggering.
Now the error is propagated so failed messages are retried.

Fixes #3215

Author: NitinKumar004

pkg/gofr/subscriber.go

@@ -69,7 +69,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 {

pkg/gofr/subscriber_test.go

@@ -4,7 +4,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"gofr.dev/pkg/gofr/container"
 	"gofr.dev/pkg/gofr/datasource"
 	"gofr.dev/pkg/gofr/datasource/pubsub"
 	"gofr.dev/pkg/gofr/datasource/pubsub/kafka"
@@ -56,3 +61,46 @@ func (mockSubscriber) Subscribe(ctx context.Context, topic string) (*pubsub.Mess
 func (mockSubscriber) Close() error {
 	return nil
 }
+
+var errHandler = errors.New("handler error")
+
+func TestHandleSubscription_HandlerErrorReturned(t *testing.T) {
+	testContainer, _ := container.NewMockContainer(t)
+	testContainer.PubSub = mockSubscriber{}
+
+	sm := newSubscriptionManager(testContainer)
+
+	err := sm.handleSubscription(context.Background(), "test-topic", func(_ *Context) error {
+		return errHandler
+	})
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, errHandler)
+}
+
+func TestHandleSubscription_SuccessfulHandler(t *testing.T) {
+	testContainer, _ := container.NewMockContainer(t)
+	testContainer.PubSub = mockSubscriber{}
+
+	sm := newSubscriptionManager(testContainer)
+
+	err := sm.handleSubscription(context.Background(), "test-topic", func(_ *Context) error {
+		return nil
+	})
+
+	assert.NoError(t, err)
+}
+
+func TestHandleSubscription_SubscribeError(t *testing.T) {
+	testContainer, _ := container.NewMockContainer(t)
+	testContainer.PubSub = mockSubscriber{}
+
+	sm := newSubscriptionManager(testContainer)
+
+	err := sm.handleSubscription(context.Background(), "test-err", func(_ *Context) error {
+		return nil
+	})
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, kafka.ErrConsumerGroupNotProvided)
+}