Always defer stop

practo · May 5, 2021 · 264fb00 · 264fb00
1 parent fc2b90c
commit 264fb00
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/pkg/redshiftloader/load_processor.go b/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()