public static Pair<GenericRowData, Integer> getRowData(Map<String, Object> record, String[] allColumns,
            DataType[] allTypes, boolean replaceLineBreak) {
        GenericRowData genericRowData = new GenericRowData(RowKind.INSERT, allColumns.length);
        int byteSize = 0;
        for (int index = 0; index < allColumns.length; index++) {
            String columnName = allColumns[index];
            LogicalType logicalType = allTypes[index].getLogicalType();
            LogicalTypeRoot typeRoot = logicalType.getTypeRoot();
            Object raw = record.get(columnName);
            byteSize += raw == null ? 0 : String.valueOf(raw).getBytes(StandardCharsets.UTF_8).length;
            switch (typeRoot) {
                case BOOLEAN:
                    genericRowData.setField(index, raw != null ? Boolean.parseBoolean(String.valueOf(raw)) : null);
                    break;
                case VARBINARY:
                case BINARY:
                    byte[] bytes = null;
                    if (raw instanceof byte[]) {
                        bytes = (byte[]) raw;
                    } else if (raw instanceof String) {
                        bytes = ((String) raw).getBytes(StandardCharsets.UTF_8);
                    }
                    genericRowData.setField(index, bytes);
                    break;
                case DECIMAL:
                    genericRowData.setField(index, raw != null ? new BigDecimal(String.valueOf(raw)) : null);
                    break;
                case DOUBLE:
                    genericRowData.setField(index, raw != null ? Double.valueOf(String.valueOf(raw)) : null);
                    break;
                case FLOAT:
                    genericRowData.setField(index, raw != null ? Float.valueOf(String.valueOf(raw)) : null);
                    break;
                case INTEGER:
                    genericRowData.setField(index, raw != null ? Integer.valueOf(String.valueOf(raw)) : null);
                    break;
                case BIGINT:
                    genericRowData.setField(index, raw != null ? Long.valueOf(String.valueOf(raw)) : null);
                    break;
                case TINYINT:
                    genericRowData.setField(index, raw != null ? Short.valueOf(String.valueOf(raw)) : null);
                    break;
                case CHAR:
                case VARCHAR:
                    String value = null;
                    if (raw != null) {
                        value = String.valueOf(record.get(columnName));
                        if (replaceLineBreak) {
                            value = value.replaceAll("[\r\n]", " ");
                        }
                    }
                    genericRowData.setField(index, value);
                    break;
                case DATE:
                case TIMESTAMP_WITHOUT_TIME_ZONE:
                case TIMESTAMP_WITH_TIME_ZONE:
                case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                    if (raw instanceof String) {
                        genericRowData.setField(index, parseDate((String) raw));
                    } else if (raw instanceof Long) {
                        genericRowData.setField(index, new Date((Long) raw));
                    }
                    break;
                default:
                    break;
            }
        }
        return new ImmutablePair<>(genericRowData, byteSize);
    }
convert json data to flink generic row data

@param record json data
@param allColumns hive column names
@param allTypes hive column types
@param replaceLineBreak if replace line break to blank to avoid data corrupt when hive text table
@return generic row data and byte size of the data
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/util/HiveTableUtil.java