@Override
    public Serializable read(final MySQLBinlogColumnDef columnDef, final MySQLPacketPayload payload) {
        int type = columnDef.getColumnMeta() >> 8;
        int length = columnDef.getColumnMeta() & 0xff;
        // unpack type & length, see https://bugs.mysql.com/bug.php?id=37426.
        if (0x30 != (type & 0x30)) {
            length += ((type & 0x30) ^ 0x30) << 4;
            type |= 0x30;
        }
        switch (MySQLBinaryColumnType.valueOf(type)) {
            case ENUM:
                return readEnumValue(length, payload);
            case SET:
                return payload.getByteBuf().readByte();
            case STRING:
                return new MySQLBinaryString(payload.readStringFixByBytes(readActualLength(length, payload)));
            default:
                throw new UnsupportedSQLOperationException(MySQLBinaryColumnType.valueOf(type).toString());
        }
    }
@Test
    void assertReadStringValue() {
        String expected = "test_value";
        columnDef.setColumnMeta(MySQLBinaryColumnType.STRING.getValue() << 8);
        when(payload.getByteBuf()).thenReturn(byteBuf);
        when(byteBuf.readUnsignedByte()).thenReturn((short) expected.length());
        when(payload.readStringFixByBytes(expected.length())).thenReturn(expected.getBytes());
        Serializable actual = new MySQLStringBinlogProtocolValue().read(columnDef, payload);
        assertInstanceOf(MySQLBinaryString.class, actual);
        assertThat(((MySQLBinaryString) actual).getBytes(), is(expected.getBytes()));
    }