@Override
    public SchemaAndValue toConnectHeader(String topic, String headerKey, byte[] value) {
        if (value == null) {
            return NULL_SCHEMA_AND_VALUE;
        }
        try {
            String str = new String(value, UTF_8);
            return Values.parseString(str);
        } catch (NoSuchElementException e) {
            throw new DataException("Failed to deserialize value for header '" + headerKey + "' on topic '" + topic + "'", e);
        } catch (Throwable t) {
            LOG.warn("Failed to deserialize value for header '{}' on topic '{}', so using byte array", headerKey, topic, t);
            return new SchemaAndValue(Schema.BYTES_SCHEMA, value);
        }
    }
A {@link HeaderConverter} that serializes header values as strings and that deserializes header values to the most appropriate
numeric, boolean, array, or map representation. Schemas are not serialized, but are inferred upon deserialization when possible.
connect/api/src/main/java/org/apache/kafka/connect/storage/SimpleHeaderConverter.java