gistfile1.txt

    public void create(final Request request, final Response response) throws Exception {
    	
    	SubPubMessage subPubMessage = null;
    
        try {
			subPubMessage = SubPubMessage.fromFormPost(request);
		} catch (Exception e) {
			log.error("Could not parse SubPub message, message not ack'd from SubPub.", e);
			throw new Exception("Could not parse SubPub message, message not ack'd from SubPub.", e);
		}

        try {
        	kafkaService.sendKafkaMessage(config.getTopicName(), subPubMessage.getPayload(), subPubMessage.getMessageType());
                countsClient.incrementGambitCount(KAFKA_MSG);
        } catch (Exception e) {
            log.error("Could not send to Kafka, message not ack'd from SubPub.", e);
        	throw new Exception("Could not send to Kafka, message not ack'd from SubPub.", e);
        }

    }