public LogData<T> deserialize(byte[] message) throws IOException {
    try {
      checkArgument(message != null, "message is null.");
      checkArgument(message.length >= ROW_BEGINNING_POS, "message is illegal.");
      byte[] versionBytes = Bytes.subByte(message, 0, 4);
      byte[] upstreamIdBytes = Bytes.subByte(message, 4, 4);
      long epicNo = Longs.fromByteArray(Bytes.subByte(message, 8, 8));
      byte flip = Bytes.subByte(message, 16, 1)[0];
      byte changeActionByte = Bytes.subByte(message, 17, 1)[0];

      boolean flipBoolean = FlipUtil.convertToBoolean(flip);

      T actualValue;
      if (flipBoolean) {
        // we can ignore actual value which should be empty, when flip is true.
        return factory.create(null, versionBytes, upstreamIdBytes, epicNo, true, changeActionByte);
      }

      byte[] actualValueBytes = Bytes.subByte(message, 18, message.length - 18);
      final JsonNode root = objectMapper.readTree(actualValueBytes);
      actualValue = (T) jsonToLogDataConverter.convert(root, null);
      return factory.create(
          actualValue, versionBytes, upstreamIdBytes, epicNo, false, changeActionByte);
    } catch (Throwable t) {
      LOG.error("", t);
      throw t;
    }
  }
Object mapper for parsing the JSON.
amoro-format-iceberg/src/main/java/org/apache/amoro/log/LogDataJsonDeserialization.java