@@ -0,0 +1,2580 @@
+package forklift.connectors;%0A%0Aimport forklift.Forklift;%0Aimport forklift.integration.server.TestServiceManager;%0Aimport forklift.producers.ForkliftSerializer;%0Aimport forklift.source.TopicSource;%0A%0Aimport org.junit.AfterClass;%0Aimport org.junit.Assert;%0Aimport org.junit.BeforeClass;%0Aimport org.junit.Test;%0A%0Aimport java.nio.ByteBuffer;%0A%0Apublic class KafkaConnectorSerializerTests %7B%0A    private static TestServiceManager serviceManager;%0A    private static ForkliftSerializer connectorSerializer;%0A    private static final TopicSource testSource = new TopicSource(%22test-string-topic%22);%0A%0A    @BeforeClass%0A    public static void setup() throws Exception %7B%0A        serviceManager = new TestServiceManager();%0A        serviceManager.start();%0A%0A        Forklift forklift = serviceManager.newManagedForkliftInstance();%0A        connectorSerializer = (ForkliftSerializer) forklift.getConnector();%0A    %7D%0A%0A    @AfterClass%0A    public static void teardown() %7B%0A        serviceManager.stop();%0A    %7D%0A%0A    private byte%5B%5D subbytes(byte%5B%5D bytes, int start, int end) %7B%0A        byte%5B%5D result = new byte%5Bend - start%5D;%0A        System.arraycopy(bytes, start, result, 0, end - start);%0A        return result;%0A    %7D%0A%0A    // the confluent avro serializer writes 1 magic byte (0x00)%0A    // followed by 4 bytes of the id of the schema to deserialize%0A    // the following avro data%0A    private byte%5B%5D stripHeader(byte%5B%5D bytes) %7B%0A        return subbytes(bytes, 5, bytes.length);%0A    %7D%0A%0A    private void assertBytesEqual(byte%5B%5D expected, byte%5B%5D actual) %7B%0A        Assert.assertArrayEquals(%0A            %22Expected bytes '%22 + hexString(expected) +%0A            %22', found bytes '%22 + hexString(actual) + %22'%22,%0A            expected, actual);%0A    %7D%0A%0A    private static final char%5B%5D hexChar = %220123456789ABCDEF%22.toCharArray();%0A    private String hexString(byte%5B%5D bytes) %7B%0A        char%5B%5D result = new char%5B2 * bytes.length%5D;%0A%0A        for (int i = 0; i %3C bytes.length; i++) %7B%0A            result%5B2 * i%5D = hexChar%5Bbytes%5Bi%5D & 0x0F%5D;%0A            result%5B2 * i + 1%5D = hexChar%5B(bytes%5Bi%5D %3E%3E 4) & 0x0F%5D;%0A        %7D%0A%0A        return new String(result);%0A    %7D%0A%0A    @Test%0A    public void testSampleStringSerializesCorrectly() %7B%0A        final byte%5B%5D sampleStringSerialized = connectorSerializer.serializeForSource(testSource, %22hello%22);%0A        final byte%5B%5D sampleStringAvro = stripHeader(sampleStringSerialized);%0A        final byte%5B%5D expectedSampleStringAvro = new byte%5B%5D%7B%0A            10, // double the length of the string%0A            'h', 'e', 'l', 'l', 'o'%0A        %7D;%0A%0A%0A        assertBytesEqual(expectedSampleStringAvro, sampleStringAvro);%0A    %7D%0A%7D%0A

Add small test that the kafka ConnectorSerializer is serializing avro correctly