Why JSON isn’t the same as JSON Schema in Kafka Connect converters and ksqlDB (Viewing Kafka messages bytes as hex)

$ echo '{"id": "2", "host": "test-machine", "body": "hello this is a test"}' | \
  kafka-json-schema-console-producer --broker-list localhost:9092  --property schema.registry.url=http://localhost:8081 --topic my_topic_jsonsr \
    --property value.schema='{ "type": "object", "properties": { "id": { "type": "string" }, "host": { "type": "string" }, "body": { "type": "string" } } }'

ksql> CREATE STREAM MY_STREAM WITH (KAFKA_TOPIC='my_topic_jsonsr', VALUE_FORMAT='JSON_SR');

 Message
----------------
 Stream created
----------------
ksql> DESCRIBE MY_STREAM;

Name                 : MY_STREAM
 Field | Type
-------------------------
 HOST  | VARCHAR(STRING)
 ID    | VARCHAR(STRING)
 BODY  | VARCHAR(STRING)
-------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;
ksql> SET 'auto.offset.reset' = 'earliest';
>
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
ksql> SELECT * FROM MY_STREAM EMIT CHANGES LIMIT 1;
+-------------+----+----------------------+
|HOST         |ID  |BODY                  |
+-------------+----+----------------------+
|test-machine |2   |hello this is a test  |
Limit Reached
Query terminated

ksql> CREATE STREAM MY_STREAM_02 WITH (KAFKA_TOPIC='my_topic_jsonsr', VALUE_FORMAT='JSON');
No columns supplied.

ksql> CREATE STREAM MY_STREAM_02 (HOST VARCHAR, ID VARCHAR, BODY VARCHAR)
        WITH (KAFKA_TOPIC='my_topic_jsonsr', VALUE_FORMAT='JSON');

 Message
----------------
 Stream created
----------------
ksql> DESCRIBE MY_STREAM_02;

Name                 : MY_STREAM_02
 Field | Type
-------------------------
 HOST  | VARCHAR(STRING)
 ID    | VARCHAR(STRING)
 BODY  | VARCHAR(STRING)
-------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;

ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' from 'earliest' to 'earliest'.
ksql> SELECT * FROM MY_STREAM_02 EMIT CHANGES LIMIT 1;
+--------+--------+---------+
|HOST    |ID      |BODY     |
+--------+--------+---------+

Press CTRL-C to interrupt

ksql> PRINT my_topic_jsonsr FROM BEGINNING LIMIT 1;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: JSON_SR or KAFKA_STRING
rowtime: 2021/03/09 14:08:14.436 Z, key: <null>, value: {"id":"2","host":"test-machine","body":"hello this is a test"}, partition: 0
Topic printing ceased
ksql>

Value format: JSON_SR or KAFKA_STRING

WARN stream-thread [_confluent-ksql-confluent_rmoff_01transient_6449533791924466400_1615299701177-972676ef-317f-4d3b-a30b-66d8ff86f577-StreamThread-1] task [0_0] Skipping record due to deserialization error. topic=[my_topic_jsonsr] partition=[0] offset=[0] (org.apache.kafka.streams.processor.internals.RecordDeserializer:88)
org.apache.kafka.common.errors.SerializationException: Failed to deserialize value from topic: my_topic_jsonsr. Invalid UTF-32 character 0x567a2269 (above 0x0010ffff) at char #1, byte #7)
Caused by: java.io.CharConversionException: Invalid UTF-32 character 0x567a2269 (above 0x0010ffff) at char #1, byte #7)
        at com.fasterxml.jackson.core.io.UTF32Reader.reportInvalid(UTF32Reader.java:195)
        at com.fasterxml.jackson.core.io.UTF32Reader.read(UTF32Reader.java:158)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._loadMore(ReaderBasedJsonParser.java:248)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._skipWSOrEnd(ReaderBasedJsonParser.java:2359)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:671)
        at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4247)
        at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:2734)
        at io.confluent.ksql.serde.json.KsqlJsonDeserializer.deserialize(KsqlJsonDeserializer.java:115)
        at io.confluent.ksql.serde.connect.ConnectFormat$StructToListDeserializer.deserialize(ConnectFormat.java:224)
        at io.confluent.ksql.serde.connect.ConnectFormat$StructToListDeserializer.deserialize(ConnectFormat.java:203)
        at io.confluent.ksql.serde.GenericDeserializer.deserialize(GenericDeserializer.java:59)
        at io.confluent.ksql.logging.processing.LoggingDeserializer.tryDeserialize(LoggingDeserializer.java:60)
        at io.confluent.ksql.logging.processing.LoggingDeserializer.deserialize(LoggingDeserializer.java:47)
        at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60)
        at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58)
        at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
        at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:176)
        at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:112)
        at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:185)
        at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:891)
        at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:1038)
        at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:842)
        at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:657)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:559)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:539)

$ kafkacat -b localhost:9092 -C -t my_topic_jsonsr -p 0 -o 0

V{"id":"2","host":"test-machine","body":"hello this is a test"}

$ kafkacat -b localhost:9092 -C -t my_topic_jsonsr -p 0 -o 0 -u | hexdump -C

00000000  00 00 00 00 56 7b 22 69  64 22 3a 22 32 22 2c 22  |....V{"id":"2","|
00000010  68 6f 73 74 22 3a 22 74  65 73 74 2d 6d 61 63 68  |host":"test-mach|
00000020  69 6e 65 22 2c 22 62 6f  64 79 22 3a 22 68 65 6c  |ine","body":"hel|
00000030  6c 6f 20 74 68 69 73 20  69 73 20 61 20 74 65 73  |lo this is a tes|
% Reached end of topic my_topic_jsonsr [0] at offset 1