fix: Ack handled messages in pulsar eventsource (#1455)

Signed-off-by: Arnar Gauti Ingason <arnarg@fastmail.com>
argoproj · Dec 9, 2021 · d446437 · d446437
1 parent 29b1182
commit d446437
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/eventsources/sources/pulsar/start.go b/eventsources/sources/pulsar/start.go
@@ -157,6 +157,8 @@ consumeMessages:
 			if err := el.handleOne(msg, dispatch, log); err != nil {
 				log.Errorw("failed to process a Pulsar event", zap.Error(err))
 				el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
+			} else {
+				consumer.Ack(msg.Message)
 			}
 		case <-ctx.Done():
 			consumer.Close()