mimaison commented on code in PR #19449:
URL: https://github.com/apache/kafka/pull/19449#discussion_r2177525246
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java:
##########
@@ -345,13 +360,16 @@ public SchemaAndValue toConnectData(String topic, byte[]
value) {
throw new DataException("Converting byte[] to Kafka Connect data
failed due to serialization error: ", e);
}
- if (config.schemasEnabled() && (!jsonValue.isObject() ||
jsonValue.size() != 2 || !jsonValue.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)
|| !jsonValue.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)))
- throw new DataException("JsonConverter with schemas.enable
requires \"schema\" and \"payload\" fields and may not contain additional
fields." +
+ if (config.schemasEnabled()) {
+ if (this.schema != null) {
Review Comment:
We can just use `schema` here like you did just below
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java:
##########
@@ -68,6 +68,10 @@ public class JsonConverter implements Converter,
HeaderConverter, Versioned {
private static final Map<Schema.Type, JsonToConnectTypeConverter>
TO_CONNECT_CONVERTERS = new EnumMap<>(Schema.Type.class);
+ // if a schema is provided in config, this schema will
+ // be used for all messages
+ private Schema schema = null;
Review Comment:
Can we put this with the other non-static, non-final fields below?
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverterConfig.java:
##########
@@ -36,6 +37,10 @@ public final class JsonConverterConfig extends
ConverterConfig {
private static final String SCHEMAS_ENABLE_DISPLAY = "Enable Schemas";
public static final String SCHEMAS_CACHE_SIZE_CONFIG =
"schemas.cache.size";
+ public static final String SCHEMA_CONTENT_CONFIG = "schema.content";
Review Comment:
Can we keep the `schemas.cache.size` fields together instead of adding these
new fields in the middle?
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java:
##########
@@ -291,6 +295,17 @@ public void configure(Map<String, ?> configs) {
fromConnectSchemaCache = new SynchronizedCache<>(new
LRUCache<>(config.schemaCacheSize()));
toConnectSchemaCache = new SynchronizedCache<>(new
LRUCache<>(config.schemaCacheSize()));
+
+ try {
+ final byte[] schemaContent = config.schemaContent();
+ if (schemaContent != null && schemaContent.length > 0) {
Review Comment:
I wonder if the empty string case should be handled in JsonConverterConfig,
WDYT?
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java:
##########
@@ -291,6 +295,17 @@ public void configure(Map<String, ?> configs) {
fromConnectSchemaCache = new SynchronizedCache<>(new
LRUCache<>(config.schemaCacheSize()));
toConnectSchemaCache = new SynchronizedCache<>(new
LRUCache<>(config.schemaCacheSize()));
+
+ try {
+ final byte[] schemaContent = config.schemaContent();
+ if (schemaContent != null && schemaContent.length > 0) {
+ final JsonNode schemaNode = deserializer.deserialize("",
schemaContent);
+ this.schema = asConnectSchema(schemaNode);
+ }
+ } catch (SerializationException e) {
+ throw new DataException("Failed to parse schema in converter
config due to serialization error: ", e);
+ }
+
Review Comment:
We don't need this extra line
##########
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java:
##########
@@ -68,6 +68,10 @@ public class JsonConverter implements Converter,
HeaderConverter, Versioned {
private static final Map<Schema.Type, JsonToConnectTypeConverter>
TO_CONNECT_CONVERTERS = new EnumMap<>(Schema.Type.class);
+ // if a schema is provided in config, this schema will
+ // be used for all messages
Review Comment:
A short comment like this can fit on a single line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
