KAFKA-14425; The Kafka protocol should support nullable structs

[dajac]

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-893%3A+The+Kafka+protocol+should+support+nullable+structs

Committer Checklist (excluded from commit message)

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

[dajac]

@cmccabe Could you review this one please? The vote is not closed yet but we have all the votes.

[cmccabe]

Thanks for this, @dajac . Can you add a test that the Invalid default for struct field message appears as expected when the default is set to something other than null? MessageDataGeneratorTest.java has some examples of tests like that

The other test that I really want is a test that a tagged nullable struct will be correctly ignored by code prior to this PR. Unfortunately I don't have a good suggestion for how to do that in junit or ducktape. Can you do some kind of manual test to ensure that this is the case?

(Once we have people using tagged nullable structs the ducktape cross-version compatibility tests will serve as a de-facto test of this, but that will come later)

[dajac]

Can you add a test that the Invalid default for struct field message appears as expected when the default is set to something other than null? MessageDataGeneratorTest.java has some examples of tests like that

That makes sense. Added tests.

The other test that I really want is a test that a tagged nullable struct will be correctly ignored by code prior to this PR. Unfortunately I don't have a good suggestion for how to do that in junit or ducktape. Can you do some kind of manual test to ensure that this is the case?

I did the following. With this PR, I used the following schema and code to serialize a message.

{
  "name": "TestMessage",
  "type": "header",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "struct1", "type": "Struct", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "tag": 1, "taggedVersions": "0+", "fields": [
        { "name": "int1", "type": "int32", "versions": "0+" }
      ]
    },
    { "name": "int0", "type": "int32", "versions": "0+", "tag": 0, "taggedVersions": "0+" }
  ]
}

        TestMessageData.Struct struct = new TestMessageData.Struct()
            .setInt1(100);
        TestMessageData message = new TestMessageData()
            .setStruct1(struct)
            .setInt0(200);

        ByteBuffer buffer = MessageUtil.toByteBuffer(message, (short) 0);
        FileChannel fc = new FileOutputStream("/tmp/test-message").getChannel();
        fc.write(buffer);
        fc.close();

Then, with trunk, I used the following schema and code to validate.

{
  "name": "TestMessage",
  "type": "header",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "int0", "type": "int32", "versions": "0+", "tag": 0, "taggedVersions": "0+" }
  ]
}

        FileChannel fc = new FileInputStream("/tmp/test-message").getChannel();
        ByteBuffer buffer = ByteBuffer.allocate((int) fc.size());
        fc.read(buffer);
        fc.close();
        buffer.flip();

        TestMessageData message = new TestMessageData();
        message.read(new ByteBufferAccessor(buffer), (short) 0);

        assertEquals(200, message.int0());
        assertEquals(1, message.unknownTaggedFields().size());

That worked as expected.

[dajac]

The KIP has been accepted.

[cmccabe]

LGTM