MessageController.java

@Controller
@ResponseBody
@RequestMapping("/message")
public class MessageController {

	ObjectMapper mapper = new ObjectMapper();

	@Autowired
	CamelContext camelContext;

	@Autowired
	@Qualifier("KafkaRouteProducer")
	RouteBuilder kafkaRouteProducer;

	@Autowired
	@Qualifier("KafkaRouteConsumer")
	RouteBuilder kafkaRouteConsumer;

	@EndpointInject(uri = "direct:kafkaRoute")
	ProducerTemplate kafkaProducer;

	ConsumerTemplate kafkaConsumer;

	@PostConstruct
	public void setup() {
		try {
			camelContext.addRoutes(kafkaRouteProducer);
			camelContext.addRoutes(kafkaRouteConsumer);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * GET to produce a message for Kafaka.
	 * 
	 * @param request
	 *            the {@link HttpServletRequest} object.
	 * @param response
	 *            the {@link HttpServletResponse} object.
	 */
	@RequestMapping(method = RequestMethod.GET)
	public void get(HttpServletRequest request, HttpServletResponse response) {
		try {
			kafkaProducer.sendBody("direct:kafkaRoute", "This is a message from the /message route!");
		} catch (Exception exception) {
			exception.printStackTrace();
		}
	}

	/**
	 * POST a notification, send it as a message to Kafka.
	 * 
	 * @param request
	 *            the {@link HttpServletRequest} object.
	 * @param response
	 *            the {@link HttpServletResponse} object.
	 * @param notification
	 *            the {@link Notification} to be posted.
	 */
	@RequestMapping(method = RequestMethod.POST)
	public void post(HttpServletRequest request, HttpServletResponse response, @RequestBody Notification notification) {
		try {
			kafkaProducer.sendBody("direct:kafkaRoute", mapper.writeValueAsString(notification));
		} catch (Exception exception) {
			exception.printStackTrace();
		}
	}
}