private static <T> LogDataToJsonConverter<T> createMapConverter(
      Type type, LogData.FieldGetterFactory<T> fieldGetterFactory) {
    Types.MapType map = type.asNestedType().asMapType();
    Types.NestedField keyField = map.field(map.keyId());
    Types.NestedField valueField = map.field(map.valueId());
    final LogDataToJsonConverter<T> valueConverter =
        createConverter(valueField.type(), fieldGetterFactory);
    final LogArrayData.ElementGetter valueGetter = LogArrayData.createElementGetter(valueField);
    final LogArrayData.ElementGetter keyGetter = LogArrayData.createElementGetter(keyField);
    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;
        node.removeAll();
      }
      LogMapData mapData = (LogMapData) source;
      LogArrayData keyArray = mapData.keyArray();
      LogArrayData valueArray = mapData.valueArray();
      int numElements = mapData.size();
      for (int i = 0; i < numElements; i++) {
        String fieldName;
        if (keyArray.isNullAt(i)) {
          // when map key is null
          throw new RuntimeException(
              "JSON format doesn't support to serialize map data with null keys. ");
        } else {
          Object obj = keyGetter.getElementOrNull(keyArray, i);
          fieldName = null != obj ? obj.toString() : null;
        }

        Object value = valueGetter.getElementOrNull(valueArray, i);

        LogDataToJsonConverter.FormatConverterContext subContext =
            new LogDataToJsonConverter.FormatConverterContext(context.mapper, node.get(fieldName));
        node.set(fieldName, valueConverter.convert(value, subContext));
      }

      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