[KAFKA-14716] Connect schema does not allow struct default values

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Assignee: Daniel Urban
• Reporter: Daniel Urban
• Created: 14/Feb/23 16:29
• Updated: 15/Feb/23 09:38
• Resolved: 15/Feb/23 09:38

Description

The ConnectSchema API should allow specifying a composite (struct) default value for a field, but with the current API, it is impossible to do so.

1. There is a circular dependency between creating a struct as a default value and creating the schema which holds it as the default value. The Struct constructor expects a Schema object, and the default value setter of SchemaBuilder checks schema conformity by using the ConnectSchema.validateValue, which in turn uses ConnectSchema.equals. This can only be bypassed if the struct references a SchemaBuilder instance, and defaultValue is called on that builder instance, but this goes against the Struct docs stating that "Struct objects must specify a complete {@link Schema} up front".
2. ConnectSchema.equals is not prepared to be used with other Schema implementations, so equals checks between ConnectSchema and SchemaBuilder instances will always fail. This is only causing an issue if equals has to be used for schema conformity checks.

Code examples:

Working code (mind that the schema referenced by the Struct is a SchemaBuilder, and it is mutated after the Struct is constructed):

@Test
public void testCompositeDefault() {
    SchemaBuilder nestedSchema = SchemaBuilder.struct()
            .field("bar", Schema.STRING_SCHEMA);
    Struct nestedDefault = new Struct(nestedSchema);
    nestedDefault.put("bar", "default_value");

    Schema schema = SchemaBuilder.struct()
            .field("foo",
                    nestedSchema
                            .defaultValue(nestedDefault)
                            .build()
            )
            .build();
} 

Not working code (but better aligned with the current API and docs - 2 separate Schema instances used by the Struct and the field, only diff is the default value between the 2):

 @Test
public void testCompositeDefault() {
    Schema nestedSchema = SchemaBuilder.struct()
            .field("bar", Schema.STRING_SCHEMA)
            .build();
    Struct nestedDefault = new Struct(nestedSchema);
    nestedDefault.put("bar", "default_value");

    Schema schema = SchemaBuilder.struct()
            .field("foo",
                    SchemaBuilder
                            .struct()
                            .field("bar", Schema.STRING_SCHEMA)
                            .defaultValue(nestedDefault)
                            .build()
            )
            .build();
}

Issue Links

• duplicates: KAFKA-12694 Specifying a struct-based defaultValue on a SchemaBuilder causes a DataException