[Bug] HTTP source connector stop after receiving an invalid request

[Fungx]

Search before asking

• [X] I had searched in the issues and found no similar issues.

Environment

Linux

EventMesh version

master

What happened

HTTP source connector stop after receiving an request without attribute subject and never response again.

How to reproduce

Reproduce

1. start http connector server
2. send an invalid request without subject, such as

   curl --location --request POST 'http://localhost:3755/test' \
   --header 'ce-id: 11' \
   --header 'ce-specversion: 1.0' \
   --header 'ce-type: com.example.someevent' \
   --header 'ce-source: /mycontext' \
   # --header 'ce-subject: test' \
   --header 'Content-Type: text/plain' \
   --data-raw 'aaa'

3. http connector server stop forever

Analysis

In HttpSourceConnector#poll, the conversion from cloudevent to connectRecord is performed.

If a cloudevent contains null subject, connectRecord.addExtension("topic", event.getSubject()) will throw unexpected NPE, which will stop the connector server.

Fix

Check subject before saving cloudevents. We don't need to check other attributes like id, source, type,etc. Because they are required arrtibutes of cloudevents and cloudevents-sdk will do the check in toEvent().

VertxMessageFactory.createReader(ctx.request())
                    .map(r -> {
                        CloudEvent event = r.toEvent();
                        if (event.getSubject() == null) {
                            throw new IllegalStateException("attribute 'subject' cannot be null");
                        }
                        return event;
                    })
setting others...

Debug logs

No response

Are you willing to submit PR?

• [X] Yes I am willing to submit a PR!

Code of Conduct

• [X] I agree to follow this project's Code of Conduct *

[Pil0tXia]

Good. Is there a way for the sender to get that there is an error in one of the packets sent without looking at the EM logs?

[Fungx]

Good. Is there a way for the sender to get that there is an error in one of the packets sent without looking at the EM logs?

Sender can get it from the http status code. 400 will be returned if there is an error in the conversion.