Merge pull request #111 from scribd/laynax/SERF-1677/rec-mw-kafka

[SERF1677] Add sentry recovery to Kafka consumer

pkg/pubsub/kafka/partitionconsumer.go

@@ -2,7 +2,9 @@ package kafka
 
 import (
 	"context"
+	"time"
 
+	"github.com/getsentry/sentry-go"
 	"github.com/twmb/franz-go/pkg/kgo"
 
 	sdkkafka "github.com/scribd/go-sdk/pkg/instrumentation/kafka"
@@ -18,6 +20,14 @@ type pconsumer struct {
 }
 
 func (pc *pconsumer) consume(cl *kgo.Client, logger sdklogger.Logger, shouldCommit bool, handler func(*kgo.Record)) {
+	defer func() {
+		if rec := recover(); rec != nil {
+			sentry.CurrentHub().Recover(rec)
+			sentry.Flush(time.Second * 5)
+			logger.Fatalf("kafka consumer: panic error: %v", rec)
+		}
+	}()
+
 	defer close(pc.done)
 
 	for {