Hiya, I’m late to this thread, and was just referred to the KIP from the
confluent mailing list. Nice idea!
I like the idea of being able to provide a Schema fallback for
JsonConverter. Perhaps an interface that could return the schema given the
JsonNode? For my use case, I have JSONSchemas resolvable via a
schema_uri field in my JSON data. I’d like to be able to implement an
interface that would look up the JSONSchema for the given message, and then
convert that JSONSchema to a Connect Schema.
Then, instead of inferSchema being a simple method that is either enabled
or not, we can set a schema.resolver.type. Pseudocode E.g.
interface JsonSchemaResolver {
public Schema apply(JsonNode value);
}
class InferSchemaResolver implements JsonSchemaResolver {
public Schema apply(JsonNode value) {
return inferSchema(value);
}
}
Or for me and my custom resolver…
class JsonSchemaURIResolver implements JsonSchemaResolver {
public JsonSchemaURIResolver(schemaUriPath) {
// …
}
public Schema apply(JsonNode value) {
schemaURI = value.at(schemaUriPath);
// cache JSON schema here.
return asConnectSchema(getJsonSchemaAtURI(schemaURI));
}
public Schema asConnectSchema(JsonNode schema) {
// Recursively traverse schema and pull convert “type” fields to
Connect Schema types.
}
}
And in config:
# perhaps the default:
schema.resolver.type=org.apache.kafka.connect.json.InferSchemaResolver
# Or for me and my custom resolver…
schema.resolver.type=org.apache.kafka.connect.json.JsonSchemaURIResolver
If we were really fancy, I suppose the Resolver interface wouldn't have to
be JsonNode specific. Then any Converter could be configured with a schema
resolver implementation that returned a Schema from an Object value.
(I believe this is similar to how Confluent’s AvroConverter works, but that
is Confluent Schema Registry specific.)
Anyway, just a thought.
- Andrew Otto
Senior Systems Engineer
Wikimedia Foundation
