changed BodyConsumer to be abstrac class and added handleError 

so the handler can release resources if there are exceptions while processing in pipeline.