private JsonToLogDataConverter<T> createNotNullConverter(Type type) {
    switch (type.typeId()) {
      case BOOLEAN:
        return this::convertToBoolean;
      case INTEGER:
        return this::convertToInt;
      case LONG:
        return this::convertToLong;
      case FLOAT:
        return this::convertToFloat;
      case DOUBLE:
        return this::convertToDouble;
      case DATE:
        return this::convertToDate;
      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 this::convertToTime;
      case TIMESTAMP:
        Types.TimestampType timestamp = (Types.TimestampType) type;
        if (timestamp.shouldAdjustToUTC()) {
          return this::convertToTimestampWithLocalZone;
        } else {
          return this::convertToTimestamp;
        }
      case STRING:
        return this::convertToString;
      case UUID:
      case FIXED:
      case BINARY:
        return this::convertToBytes;
      case DECIMAL:
        return this::convertDecimal;
      case LIST:
        return createListConverter(type);
      case MAP:
        return createMapConverter(type);
      case STRUCT:
        return createStructConverter(type);
      default:
        throw new UnsupportedOperationException("Not Support to parse type: " + type);
    }
  }
Converter to convert {@link JsonNode} to log data.

@param <T> to indicate the log data type
amoro-format-iceberg/src/main/java/org/apache/amoro/log/JsonToLogDataConverters.java