Support cloud events for KFServing custom framework

[yuzisun]

/kind feature

Describe the solution you'd like
As a user when doing feature transformation the payload can be protobuf/avro or some other binary format delivered via HTTP or KAFKA in cloudevent binary format(https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#32-binary-content-mode) but currently KFServing custom framework only supports JSON.

• I'd like to send cloud event to KFServer via KAFKA in following format and KFServer pass the decoded cloud event to preprocess or predict handler

------------------ Message -------------------
Topic Name: mytopic
------------------- key ----------------------
Key: mykey
------------------ headers -------------------
ce_specversion: "1.0"
ce_type: "com.example.someevent"
ce_source: "/mycontext/subcontext"
ce_id: "1234-1234-1234"
ce_time: "2018-04-05T03:56:24Z"
content-type: application/protobuf
------------------- value --------------------
      ... application data encoded in protobuf ...
-----------------------------------------------

             consume                       http(cloudevent)
KAFKA ----------> Kafka Event Source -----------------> InferenceService

def has_ce_headers(headers: typing.Dict[str, str]) -> bool:
    return (
        "ce-specversion" in headers
        and "ce-source" in headers
        and "ce-type" in headers
        and "ce-id" in headers
    )
    

 if has_ce_headers(self.request.headers):
        event = from_http(self.request.headers, self.request.body)
        body = event['data']
 else:
        try:
            body = json.loads(self.request.body)
        except json.decoder.JSONDecodeError as e:
            raise tornado.web.HTTPError(
                 status_code=HTTPStatus.BAD_REQUEST,
                 reason="Unrecognized request format: %s" % e)

• Currently HTTP headers are not exposed to the preprocess or predict handler, if a cloud event is sent to KFServer it is not possible for user to check the http headers which can be useful for tracing and examining the original event source(kafka or http).
  async def predict(self, request: Dict, headers: Dict) -> Dict:
async def preprocess(self, request: Dict, headers: Dict) -> Dict:

• Also it would be nice if KFServing can generate cloud event response so that it works with knative eventing channels seamlessly.

Anything else you would like to add:
[Miscellaneous information that will assist in solving the issue.]

[issue-label-bot]

Issue-Label Bot is automatically applying the labels:

Label	Probability
area/engprod	0.91

Please mark this comment with 👍 or 👎 to give our bot feedback!
Links: app homepage, dashboard and code for this bot.

[issue-label-bot]

Issue-Label Bot is automatically applying the labels:

Label	Probability
area/engprod	0.91

Please mark this comment with 👍 or 👎 to give our bot feedback!
Links: app homepage, dashboard and code for this bot.

[issue-label-bot]

Issue-Label Bot is automatically applying the labels:

Label	Probability
area/front-end	0.85

Please mark this comment with 👍 or 👎 to give our bot feedback!
Links: app homepage, dashboard and code for this bot.

[issue-label-bot]

Issue-Label Bot is automatically applying the labels:

Label	Probability
area/front-end	0.85

Please mark this comment with 👍 or 👎 to give our bot feedback!
Links: app homepage, dashboard and code for this bot.