@@ -0,0 +1,1454 @@
+/**%0A * Copyright 2016 Evokly S.A.%0A * See LICENSE file for License%0A **/%0A%0Apackage com.evokly.kafka.connect.mqtt;%0A%0Aimport static org.junit.Assert.assertEquals;%0A%0Aimport org.apache.kafka.connect.source.SourceRecord;%0Aimport org.eclipse.paho.client.mqttv3.MqttMessage;%0Aimport org.junit.Before;%0Aimport org.junit.Test;%0A%0Aimport java.nio.charset.StandardCharsets;%0Aimport java.util.List;%0A%0Apublic class MqttSourceTaskTest %7B%0A    private MqttSourceTask mTask;%0A%0A    /**%0A     * Several tests need similar objects created before they can run.%0A     */%0A    @Before%0A    public void beforeEach() %7B%0A        mTask = new MqttSourceTask();%0A    %7D%0A%0A    @Test%0A    public void testPoll() throws Exception %7B%0A        // empty queue%0A        assertEquals(mTask.mQueue.size(), 0);%0A%0A        // add dummy message to queue%0A        MqttMessage mqttMessage = new MqttMessage();%0A        mqttMessage.setPayload(%22test_message%22.getBytes(StandardCharsets.UTF_8));%0A        mTask.messageArrived(%22test_topic%22, mqttMessage);%0A%0A        // check message to be process%0A        assertEquals(mTask.mQueue.size(), 1);%0A%0A        // generate and validate SourceRecord%0A        List%3CSourceRecord%3E sourceRecords = mTask.poll();%0A%0A        assertEquals(sourceRecords.size(), 1);%0A        assertEquals(sourceRecords.get(0).key(), %22test_topic%22);%0A        assertEquals(new String((byte%5B%5D) sourceRecords.get(0).value(), %22UTF-8%22), %22test_message%22);%0A%0A        // empty queue%0A        assertEquals(mTask.mQueue.size(), 0);%0A    %7D%0A%7D%0A

validate queue and poll request