private Object convert(JsonNode node, TypeInformation<?> info) {
		if (info.equals(Types.VOID) || node.isNull()) {
			return null;
		} else if (info.equals(Types.BOOLEAN)) {
			return node.asBoolean();
		} else if (info.equals(Types.STRING)) {
			return node.asText();
		} else if (info.equals(Types.BIG_DEC)) {
			return node.decimalValue();
		} else if (info.equals(Types.BIG_INT)) {
			return node.bigIntegerValue();
		} else if (info.equals(Types.SQL_DATE)) {
			return Date.valueOf(node.asText());
		} else if (info.equals(Types.SQL_TIME)) {
			// according to RFC 3339 every full-time must have a timezone;
			// until we have full timezone support, we only support UTC;
			// users can parse their time as string as a workaround
			final String time = node.asText();
			if (time.indexOf('Z') < 0 || time.indexOf('.') >= 0) {
				throw new IllegalStateException(
					"Invalid time format. Only a time in UTC timezone without milliseconds is supported yet. " +
						"Format: HH:mm:ss'Z'");
			}
			return Time.valueOf(time.substring(0, time.length() - 1));
		} else if (info.equals(Types.SQL_TIMESTAMP)) {
			// according to RFC 3339 every date-time must have a timezone;
			// until we have full timezone support, we only support UTC;
			// users can parse their time as string as a workaround
			final String timestamp = node.asText();
			if (timestamp.indexOf('Z') < 0) {
				throw new IllegalStateException(
					"Invalid timestamp format. Only a timestamp in UTC timezone is supported yet. " +
						"Format: yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
			}
			return Timestamp.valueOf(timestamp.substring(0, timestamp.length() - 1).replace('T', ' '));
		} else if (info instanceof RowTypeInfo) {
			return convertRow(node, (RowTypeInfo) info);
		} else if (info instanceof ObjectArrayTypeInfo) {
			return convertObjectArray(node, ((ObjectArrayTypeInfo) info).getComponentInfo());
		} else if (info instanceof BasicArrayTypeInfo) {
			return convertObjectArray(node, ((BasicArrayTypeInfo) info).getComponentInfo());
		} else if (info instanceof PrimitiveArrayTypeInfo &&
				((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
			return convertByteArray(node);
		} else {
			// for types that were specified without JSON schema
			// e.g. POJOs
			try {
				return objectMapper.treeToValue(node, info.getTypeClass());
			} catch (JsonProcessingException e) {
				throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
			}
		}
	}
Configures the failure behaviour if a JSON field is missing.

<p>By default, a missing field is ignored and the field is set to null.

@param failOnMissingField Flag indicating whether to fail or not on a missing field.
kafka-base/kafka-base-source/src/main/java/org/apache/flink/formats/json/DTJsonRowDeserializationSchema.java