public byte[] serialize(LogData<T> element) {
    // 4 bytes version + 4 bytes upstreamId + 8 bytes EpicNo + 1 byte flip + 1 byte rowKind + n
    // bytes object data
    MessageBytes messageBytes = new MessageBytes();

    messageBytes
        .append(element.getVersionBytes())
        .append(element.getUpstreamIdBytes())
        .append(element.getEpicNoBytes())
        .append(element.getFlipByte())
        .append(element.getChangeActionByte());

    if (element.getFlip()) {
      // would ignore serializing actual value if flip is true.
      return messageBytes.toBytes();
    }

    // append n bytes data bytes
    if (node == null) {
      node = mapper.createObjectNode();
      converterContext =
          new LogDataToJsonConverters.LogDataToJsonConverter.FormatConverterContext(mapper, node);
    }

    try {
      convertRow(element);
      byte[] actualDataBytes = mapper.writeValueAsBytes(node);
      messageBytes.append(actualDataBytes);
    } catch (Throwable t) {
      throw new RuntimeException("Could not serialize row '" + element + "'. ", t);
    }
    return messageBytes.toBytes();
  }
Object mapper that is used to create output JSON objects.
amoro-format-iceberg/src/main/java/org/apache/amoro/log/LogDataJsonSerialization.java