private static <T> LogDataToJsonConverter<T> createStructConverter(
      Type type, LogData.FieldGetterFactory<T> fieldGetterFactory) {
    final Types.StructType structType = type.asNestedType().asStructType();
    final List<Types.NestedField> fields = structType.fields();
    final Type[] fieldTypes = fields.stream().map(Types.NestedField::type).toArray(Type[]::new);
    final String[] fieldNames = fields.stream().map(Types.NestedField::name).toArray(String[]::new);
    final List<LogDataToJsonConverter<T>> fieldConverterList =
        Arrays.stream(fieldTypes)
            .map(type1 -> createConverter(type1, fieldGetterFactory))
            .collect(Collectors.toList());
    final int fieldCount = fields.size();
    List<LogData.FieldGetter<T>> fieldGetterList = new ArrayList<>();
    for (int i = 0; i < fieldCount; i++) {
      fieldGetterList.add(fieldGetterFactory.createFieldGetter(fieldTypes[i], i));
    }

    return (source, context) -> {
      ObjectNode node;
      // reuse could be a NullNode if last record is null.
      if (context.node == null || context.node.isNull()) {
        node = context.mapper.createObjectNode();
      } else {
        node = (ObjectNode) context.node;
      }
      T actualValue = (T) source;
      for (int i = 0; i < fieldCount; i++) {
        String fieldName = fieldNames[i];
        try {
          LogDataToJsonConverter.FormatConverterContext subContext =
              new LogDataToJsonConverter.FormatConverterContext(
                  context.mapper, node.get(fieldName));

          Object field = fieldGetterList.get(i).getFieldOrNull(actualValue, i);
          node.set(fieldName, fieldConverterList.get(i).convert(field, subContext));
        } catch (Throwable t) {
          throw new RuntimeException(
              String.format("Fail to serialize at field: %s.", fieldName), t);
        }
      }
      return node;
    };
  }
Runtime converter that converts {@link LogData} into {@link JsonNode}

@param <T> indicate an actual value wrapped within {@link LogData}
amoro-format-iceberg/src/main/java/org/apache/amoro/log/LogDataToJsonConverters.java