feature(eventbus/kafka): retry eventhandling until success

Previously the eventbus was lossy in case of eventhandlers returning error.

eventbus/kafka/eventbus.go

@@ -319,23 +319,33 @@ func (b *EventBus) handle(m eh.EventMatcher, h eh.EventHandler, r *kafka.Reader)
 			continue
 		}
 
-		if err := handler(b.cctx, msg); err != nil {
+		for {
 			select {
-			case b.errCh <- err:
+			case <-b.cctx.Done():
+				return
 			default:
-				log.Printf("eventhorizon: missed error in Kafka event bus: %s", err)
 			}
 
-			continue
-		}
+			if err := handler(b.cctx, msg); err != nil {
+				select {
+				case b.errCh <- err:
+				default:
+					log.Printf("eventhorizon: missed error in Kafka event bus: %s", err)
+				}
+
+				time.Sleep(time.Second)
+			} else {
+				// Use a new context to always finish the commit.
+				if err := r.CommitMessages(context.Background(), msg); err != nil {
+					err = fmt.Errorf("could not commit message: %w", err)
+					select {
+					case b.errCh <- &eh.EventBusError{Err: err}:
+					default:
+						log.Printf("eventhorizon: missed error in Kafka event bus: %s", err)
+					}
+				}
 
-		// Use a new context to always finish the commit.
-		if err := r.CommitMessages(context.Background(), msg); err != nil {
-			err = fmt.Errorf("could not commit message: %w", err)
-			select {
-			case b.errCh <- &eh.EventBusError{Err: err}:
-			default:
-				log.Printf("eventhorizon: missed error in Kafka event bus: %s", err)
+				break
 			}
 		}
 	}