@Test
  public void testLogDataSerialize() throws IOException {

    LogDataJsonSerialization<RowData> logDataJsonSerialization =
        new LogDataJsonSerialization<>(userSchema, LogRecordV1.FIELD_GETTER_FACTORY);
    GenericRowData rowData = new GenericRowData(4);
    rowData.setField(0, true);
    rowData.setField(1, 1);
    rowData.setField(2, 123456789L);
    rowData.setField(
        3,
        new GenericArrayData(
            new StringData[] {
              null, StringData.fromString("b"), null, StringData.fromString("c"), null
            }));
    LogData<RowData> logData =
        new LogRecordV1(
            FormatVersion.FORMAT_VERSION_V1,
            IdGenerator.generateUpstreamId(),
            123455L,
            false,
            ChangeAction.INSERT,
            rowData);

    byte[] bytes = logDataJsonSerialization.serialize(logData);

    Assert.assertNotNull(bytes);
    String actualJson = new String(Bytes.subByte(bytes, 18, bytes.length - 18));
    String expected =
        "{\"f_boolean\":true,\"f_int\":1,\"f_long\":123456789,\"f_list_string\":[null,\"b\",null,\"c\",null]}";
    assertEquals(expected, actualJson);

    LogDataJsonDeserialization<RowData> logDataJsonDeserialization =
        new LogDataJsonDeserialization<>(
            userSchema, LogRecordV1.factory, LogRecordV1.arrayFactory, LogRecordV1.mapFactory);
    LogData<RowData> result = logDataJsonDeserialization.deserialize(bytes);
    Assert.assertNotNull(result);
    check(logData, result);
  }
This is a {@link LogRecordV1} log data test, include all data types.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/shuffle/TestLogRecordV1.java