Pulsar: Failed to decode long type fields in json schema message

[lynnmatrix]

Describe the bug
If message sent in json schema, long field will be decoded as int if its value below Integer.MAX_VALUE, other wise decoded as string.
For example, the json message below:

{
    "timestamp": 1585204833128
}

will be decoded as

{
    "timestamp": "1585204833128"
}

To Reproduce

Expected behavior
Long type fields encoded in json schema will be decoded as long.

Screenshots

Desktop (please complete the following information):

Additional context

[lynnmatrix]

GenericJsonRecord.java

    public Object getField(String fieldName) {
        JsonNode fn = this.jn.get(fieldName);
        if (fn.isContainerNode()) {
            AtomicInteger idx = new AtomicInteger(0);
            List<Field> fields = (List)Lists.newArrayList(fn.fieldNames()).stream().map((f) -> {
                return new Field(f, idx.getAndIncrement());
            }).collect(Collectors.toList());
            return new GenericJsonRecord(this.schemaVersion, fields, fn);
        } else if (fn.isBoolean()) {
            return fn.asBoolean();
        } else if (fn.isInt()) {
            return fn.asInt();
        } else if (fn.isFloatingPointNumber()) {
            return fn.asDouble();
        } else {
            return fn.isDouble() ? fn.asDouble() : fn.asText();
        }
    }

GenericJsonRecord::getField does not check whether the field is long type, and decode long field by 'fn.asText()'

[sijie]

@lynnmatrix Can you submit a pull request to fix it?

[lynnmatrix]

OK @sijie