feat: Close operator consumer when context is cancelled

operator/operator.go

@@ -290,6 +290,8 @@ func (o *Operator) Start(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
+			err = o.Close()
+			o.logger.Fatal("Error while closing", "err", err)
 			return nil
 		case err := <-metricsErrChan:
 			// TODO(samlaf); we should also register the service as unhealthy in the node api
@@ -327,6 +329,14 @@ func (o *Operator) Start(ctx context.Context) error {
 	}
 }
 
+func (o *Operator) Close() error {
+	if err := o.consumer.Close(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // Takes a CheckpointTaskCreatedLog struct as input and returns a TaskResponseHeader struct.
 // The TaskResponseHeader struct is the struct that is signed and sent to the contract as a task response.
 func (o *Operator) ProcessCheckpointTaskCreatedLog(checkpointTaskCreatedLog *taskmanager.ContractSFFLTaskManagerCheckpointTaskCreated) *taskmanager.CheckpointTaskResponse {