@Override
    public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumber logSequenceNumber) {
        AbstractWALEvent result;
        byte[] bytes = new byte[data.remaining()];
        data.get(bytes);
        String dataText = new String(bytes, StandardCharsets.UTF_8);
        if (decodeWithTX) {
            result = decodeDataWithTX(dataText);
        } else {
            result = decodeDataIgnoreTX(dataText);
        }
        result.setLogSequenceNumber(logSequenceNumber);
        return result;
    }
@Test
    void assertParallelDecodeWithTx() {
        MppTableData tableData = new MppTableData();
        tableData.setTableName("public.test");
        tableData.setOpType("INSERT");
        tableData.setColumnsName(new String[]{"data"});
        tableData.setColumnsType(new String[]{"raw"});
        tableData.setColumnsVal(new String[]{"'7D'"});
        List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
        MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, true);
        List<AbstractWALEvent> actual = new LinkedList<>();
        for (String each : dataList) {
            actual.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
        }
        assertThat(actual.size(), is(4));
        assertInstanceOf(BeginTXEvent.class, actual.get(0));
        assertThat(((BeginTXEvent) actual.get(0)).getCsn(), is(951909L));
        assertThat(((WriteRowEvent) actual.get(1)).getAfterRow().get(0).toString(), is("7D"));
        assertThat(((WriteRowEvent) actual.get(2)).getAfterRow().get(0).toString(), is("7D"));
        assertThat(((CommitTXEvent) actual.get(3)).getXid(), is(1006076L));
        assertNull(((CommitTXEvent) actual.get(3)).getCsn());
    }