Schema-registry: RecordNameStrategy throws SerializationException on null values.

[twbecker]

After the fix for #799, io.confluent.kafka.serializers.subject.RecordNameStrategy no longer accepts nulls, failing with the following exception:

org.apache.kafka.common.errors.SerializationException: In configuration value.subject.name.strategy = io.confluent.kafka.serializers.subject.RecordNameStrategy, the message value must only be an Avro record schema

This is because the record being serialized (which is nullable) is no longer passed directly to getSubjectName(). Instead the Schema is passed, which in the case of a null value has Schema.Type == Schema.Type.NULL. Since this isn't explicitly handled, the exception is thrown. This is a serious regression; obviously being able to produce nulls is critical for log compacted topics in particular.

[twbecker]

I think adding a check akin to the following to io.confluent.kafka.serializers.subject.RecordNameStrategy#subjectName will fix this:

    if (schema.getType() == Schema.Type.NULL) {
      return null;
    }

[mageshn]

@twbecker Thanks for reporting this issue. Would you be interested in making a PR for this?

[twbecker]

@mageshn https://github.com/confluentinc/schema-registry/pull/1006

[twbecker]

Ping @mageshn can I get a review on the PR for this?

[tkdrahn]

+1

[redsk]

Yes, please. I have the same problem

[cgraving]

Any further info on this? Is there a work around for sending tombstones to compacted topics with RecordNameStrategy?

[cgraving]

Also see this causing issues with Windowing and Suppression. Something like:

groupedStream.windowedBy(TimeWindows.of(ofSeconds(30)).grace(ofSeconds(1L)))  
                .reduce(WindowedStream::apply, Materialized.<String, SomeSpecificRecord, WindowStore<Bytes, byte[]>>as("REDUCER_STORE")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(valueSerde)
                )
          .suppress(Suppressed.untilWindowCloses(unbounded()))