Always defer stop

practo · May 5, 2021 · dd2bb21 · dd2bb21
1 parent 91c39a4
commit dd2bb21
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/redshiftsink/pkg/redshiftloader/load_processor.go b/redshiftsink/pkg/redshiftloader/load_processor.go
@@ -776,6 +776,8 @@ func (b *loadProcessor) processBatch(
 func (b *loadProcessor) Process(session sarama.ConsumerGroupSession, msgBuf []*serializer.Message) error {
 	start := time.Now()
 	b.metric.setStartRunning()
+	defer b.metric.setStopRunning()
+
 	b.setBatchId()
 	ctx := session.Context()
 
@@ -796,7 +798,6 @@ func (b *loadProcessor) Process(session sarama.ConsumerGroupSession, msgBuf []*s
 		return err
 	}
 	b.markOffset(session, msgBuf)
-	b.metric.setStopRunning()
 
 	var timeTaken string
 	secondsTaken := time.Since(start).Seconds()