@Test
public void testAppend() throws Exception {

    LOGGER.info("Begin Seding message to Kafka................");
    source.start();
    //send message
    sendMessageToKafka();
    Thread.sleep(5000);
    //get message from channel
    Transaction transaction = channel.getTransaction();
    try{
        transaction.begin();
        Event event;
        while ((event = channel.take()) != null) {
            LOGGER.info("#get channel########" + new String(event.getBody(), "UTF-8"));
        }
        transaction.commit();
    } finally{
        transaction.close();
    }
}