// processMessage unmarshals the provided kafka message into an event and calls the handler.
// After the message is handled, it is committed.
processMessage unmarshals the provided kafka message into an event and calls the handler.
After the message is handled, it is committed.
// processMessage unmarshals the provided kafka message into an event and calls the handler . // After the message is handled it is committed .
func processMessage(ctx context.Context, message kafka.Message, handler Handler) {
	event, err := unmarshal(message)
	if err != nil {
		log.Error(ctx, "failed to unmarshal event", err)
		message.Commit()
		return
	}
	ctx = dprequest.WithRequestId(ctx, event.TraceID)
	log.Info(ctx, "event received", log.Data{"event": event})
	err = handler.Handle(ctx, event)
	if err != nil {
		log.Error(ctx, "failed to handle event", err)
		message.Commit()
		return
	}
	log.Info(ctx, "event processed - committing message", log.Data{"event": event})
	message.Commit()
	log.Info(ctx, "message committed", log.Data{"event": event})
}
func processMessage ( ctx context . Context , message kafka . Message , handler Handler ) { event , err := unmarshal ( message ) if err != nil { log . Error ( ctx , " " , err ) message . Commit ( ) return } ctx = dprequest . WithRequestId ( ctx , event . TraceID ) log . Info ( ctx , " " , log . Data { " " : event } ) err = handler . Handle ( ctx , event ) if err != nil { log . Error ( ctx , " " , err ) message . Commit ( ) return } log . Info ( ctx , " " , log . Data { " " : event } ) message . Commit ( ) log . Info ( ctx , " " , log . Data { " " : event } ) }
processMessage unmarshals the provided kafka message into an event and calls the handler.
processMessage unmarshals the provided kafka message into an event and calls the handler .
// unmarshal - commit on failure (consuming the message again would result in the same error) // handle - commit on failure (implement error handling to not commit if message needs to be consumed again)