private static <T> LogDataToJsonConverter<T> createNotNullConverter(
      Type type, LogData.FieldGetterFactory<T> fieldGetterFactory) {
    // The following switch case originates from org.apache.iceberg.types.TypeUtil
    // and org.apache.iceberg.flink.TypeToFlinkType
    switch (type.typeId()) {
      case BOOLEAN:
        return (source, context) -> context.mapper.getNodeFactory().booleanNode((Boolean) source);
      case INTEGER:
        return ((source, context) -> context.mapper.getNodeFactory().numberNode((int) source));
      case LONG:
        return ((source, context) -> context.mapper.getNodeFactory().numberNode((long) source));
      case FLOAT:
        return ((source, context) -> context.mapper.getNodeFactory().numberNode((float) source));
      case DOUBLE:
        return ((source, context) -> context.mapper.getNodeFactory().numberNode((double) source));
      case DATE:
        return createDateConverter();
      case TIME:
        // For the type: Flink only support TimeType with default precision (second) now. The
        // precision of time is
        // not supported in Flink, so we can think of it as a simple time type directly.
        // For the data: Flink uses int that support mills to represent time data, so it supports
        // mills precision.
        return createTimeConverter();
      case TIMESTAMP:
        Types.TimestampType timestamp = (Types.TimestampType) type;
        if (timestamp.shouldAdjustToUTC()) {
          return createTimestampWithLocalZone();
        } else {
          return createTimestampConverter();
        }
      case STRING:
        return ((source, context) -> context.mapper.getNodeFactory().textNode(source.toString()));
      case UUID:
      case FIXED:
      case BINARY:
        return ((source, context) -> context.mapper.getNodeFactory().binaryNode((byte[]) source));
      case DECIMAL:
        return ((source, context) ->
            context.mapper.getNodeFactory().numberNode((BigDecimal) source));
      case LIST:
        return createListConverter(type, fieldGetterFactory);
      case MAP:
        return createMapConverter(type, fieldGetterFactory);
      case STRUCT:
        return createStructConverter(type, fieldGetterFactory);
      default:
        throw new UnsupportedOperationException("Not Support to parse type: " + type);
    }
  }
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