@Override
    protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
        while (in.readableBytes() >= 1 + MySQLBinlogEventHeader.MYSQL_BINLOG_EVENT_HEADER_LENGTH) {
            in.markReaderIndex();
            MySQLPacketPayload payload = new MySQLPacketPayload(in, ctx.channel().attr(CommonConstants.CHARSET_ATTRIBUTE_KEY).get());
            checkPayload(payload);
            MySQLBinlogEventHeader binlogEventHeader = new MySQLBinlogEventHeader(payload, binlogContext.getChecksumLength());
            if (!checkEventIntegrity(in, binlogEventHeader)) {
                return;
            }
            Optional<MySQLBaseBinlogEvent> binlogEvent = decodeEvent(binlogEventHeader, payload);
            if (!binlogEvent.isPresent()) {
                skipChecksum(binlogEventHeader.getEventType(), in);
                return;
            }
            if (binlogEvent.get() instanceof PlaceholderBinlogEvent) {
                out.add(binlogEvent.get());
                skipChecksum(binlogEventHeader.getEventType(), in);
                return;
            }
            if (decodeWithTX) {
                processEventWithTX(binlogEvent.get(), out);
            } else {
                processEventIgnoreTX(binlogEvent.get(), out);
            }
            skipChecksum(binlogEventHeader.getEventType(), in);
        }
    }
@Test
    void assertDecodeUpdateRowEvent() {
        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
        // the hex data is from update t_order set status = 'updated' where order_id = 1;
        byteBuf.writeBytes(StringUtil.decodeHexDump("00cb38a9621f010000004e0000000c7e000000007b00000000000100020004ffff08010000000000000001000000075355434345535308010000000000000001000000077570"
                + "6461746564e78cee6c"));
        byteBuf.writeBytes(StringUtil.decodeHexDump("006acb656410010000001f000000fa29000000001643000000000000b13f8340"));
        binlogContext.getTableMap().put(123L, tableMapEventPacket);
        when(tableMapEventPacket.getColumnDefs()).thenReturn(columnDefs);
        List<Object> decodedEvents = new LinkedList<>();
        binlogEventPacketDecoder.decode(channelHandlerContext, byteBuf, decodedEvents);
        assertThat(decodedEvents.size(), is(1));
        LinkedList<?> actualEventList = (LinkedList<?>) decodedEvents.get(0);
        assertThat(actualEventList.get(0), instanceOf(MySQLUpdateRowsBinlogEvent.class));
        MySQLUpdateRowsBinlogEvent actual = (MySQLUpdateRowsBinlogEvent) actualEventList.get(0);
        assertThat(actual.getBeforeRows().get(0), is(new Serializable[]{1L, 1, new MySQLBinaryString("SUCCESS".getBytes()), null}));
        assertThat(actual.getAfterRows().get(0), is(new Serializable[]{1L, 1, new MySQLBinaryString("updated".getBytes()), null}));
    }