KAFKA-16858: Throw DataException from validateValue on array and map schemas without inner schemas

[gharris1727]

The SchemaBuilder interface allows for objects with null valueSchema and/or keySchema to be constructed. These currently cause the validateValue to throw an NPE on non-empty containers. This changes empty and non-empty containers to throw DataException instead.

The first commit rewrites some existing tests, and adds assertions for the prior behavior. Look at only changes in the later commits to see the actual behavior change.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[gharris1727]

Hey @C0urante @yashmayya Could either of you PTAL? Thanks!

[C0urante]

Can you expound on the motivation for this change? IIRC we began to allow null element schemas in order to support schema inference for empty collections parsed by the Values class, which is valid IMO.

[gharris1727]

Can you expound on the motivation for this change? IIRC we began to allow null element schemas in order to support schema inference for empty collections parsed by the Values class, which is valid IMO.

Framed in that way, I also think it would make sense to try to infer a schema for empty lists. But I think that decision is undesirable in the broader context of the Connect ecosystem.

I tried to determine how a null schema is interpreted in the Connect type-system. When used in a ConnectRecord, it's a top-type, as it allows the Object value to be any type. The way Values uses it is also as a top-type, because a heterogeneous array infers a null schema. The current behavior for validateValue is as a bottom-type, as no value is valid for a null schema (although this seems to be an accident).

Much of our infrastructure is not prepared for element schemas to be null, such as the SchemaBuilder#array() and SchemaBuilder#map() methods, which still have null guards:

kafka/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java

Lines 361 to 383 in 55d38ef

	public static SchemaBuilder array(Schema valueSchema) {
	if (null == valueSchema)
	throw new SchemaBuilderException("valueSchema cannot be null.");
	SchemaBuilder builder = new SchemaBuilder(Type.ARRAY);
	builder.valueSchema = valueSchema;
	return builder;
	}
	/**
	* @param keySchema the schema for keys in the map
	* @param valueSchema the schema for values in the map
	* @return a new {@link Schema.Type#MAP} SchemaBuilder
	*/
	public static SchemaBuilder map(Schema keySchema, Schema valueSchema) {
	if (null == keySchema)
	throw new SchemaBuilderException("keySchema cannot be null.");
	if (null == valueSchema)
	throw new SchemaBuilderException("valueSchema cannot be null.");
	SchemaBuilder builder = new SchemaBuilder(Type.MAP);
	builder.keySchema = keySchema;
	builder.valueSchema = valueSchema;
	return builder;
	}

The field builder has a null guard:

kafka/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaBuilder.java

Lines 326 to 327 in 55d38ef

	if (null == fieldSchema)
	throw new SchemaBuilderException("fieldSchema for field " + fieldName + " cannot be null.");

and obviously the validateValue method was written without null schemas in-mind.

If we changed all of the above functions to remove the null guards, it would permit schemaless data within schema'd data. This would allow data sources to break the single schema vs schemaless check that most SMTs use to determine if they're operating on schema'd vs schemaless data. Rather than receiving DataExceptions when assembling the structs in the data sources, it would likely surface as NPEs in data sinks.

So I don't think we can change validateValue to be top-type without risking more bad data and NPEs overall. I also don't think it's viable to make ConnectRecord bottom-type, as schemaless data is intentionally supported. I think the best path forward is to attempt to align with the SchemaBuilder semantics before the SchemaBuilder(Type) constructor was made public, and disallow null element schemas entirely. Then, the assumption that the root Schema in ConnectRecord should be the only nullable schema will be correct. Of course, we still have to change the Values class to be more conservative with the types it outputs, which I opened this ticket for: https://issues.apache.org/jira/browse/KAFKA-16870 .

One of the good things for us is that the null-schema'd collections from the Values class and SimpleHeaderConverter don't commonly make it into the key or value, or as fields in a struct, because AK doesn't have HeaderTo$Key or HeaderTo$Value. This ticket actually isn't motivated by the Values parsing for that reason.

[C0urante]

Thanks Greg, I find the rationale about all-or-nothing expectations for schemas pretty convincing. Left a few comments on the implementation and tests, nothing major though.

[C0urante]

The field name doesn't seem like it'll be very useful here; can we just use the nameless variant? Something like "array element" could possibly work but too but it'd still lack the name of the field, which is IMO the only really useful piece of info.

It also seems a little strange with the error messages implying that the problematic value is a top-level value for the field, instead of a sub-value of it: Invalid Java object for schema with type STRING: class java.lang.Boolean for field: "entry". Something like Invalid Java object for schema with type STRING: class java.lang.Boolean for element of array field: "field" could be less ambiguous.

[gharris1727]

I agree that the field name sucks here. I don't think that justifies keeping the old behavior, especially where field "null" appears in the exception :)

I've implemented your suggestion with a new validateValue method, so that existing external callers passing a field name will keep the existing quoting.

[C0urante]

Ah, nice touch with the location variant. LGTM!

[gharris1727]

Test failures appear unrelated, and the connect tests pass for me locally.