@Override
    public R apply(R record) {
        final Object value = operatingValue(record);
        final Schema schema = operatingSchema(record);
        if (value == null && schema == null) {
            return record;
        }
        requireSchema(schema, "updating schema metadata");
        final boolean isArray = schema.type() == Schema.Type.ARRAY;
        final boolean isMap = schema.type() == Schema.Type.MAP;
        final Schema updatedSchema = new ConnectSchema(
                schema.type(),
                schema.isOptional(),
                schema.defaultValue(),
                schemaName != null ? schemaName : schema.name(),
                schemaVersion != null ? schemaVersion : schema.version(),
                schema.doc(),
                schema.parameters(),
                schema.fields(),
                isMap ? schema.keySchema() : null,
                isMap || isArray ? schema.valueSchema() : null
        );
        log.trace("Applying SetSchemaMetadata SMT. Original schema: {}, updated schema: {}",
            schema, updatedSchema);
        return newRecord(record, updatedSchema);
    }
@Test
    public void valueSchemaRequired() {
        final SinkRecord record = new SinkRecord("", 0, null, null, null, 42, 0);
        assertThrows(DataException.class, () -> xform.apply(record));
    }