KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification by DL1231 · Pull Request #20614 · apache/kafka

DL1231 · 2025-09-30T01:04:12Z

This patch introduces a clear separation between nullable and
non-nullable data structures. The key changes include:

Differentiates between nullable and non-nullable versions of
RECORDS, COMPACT_RECORDS, and Schema types.
Adds explicit nullable type names for ArrayOf and CompactArrayOf.
Introduces a new, concise syntax for representing types:
- {} for struct, ?{} for nullable struct
- [T] for array, ?[T] for nullable array
- (T) for compact array, ?(T) for nullable compact array
Declares shared schemas as non-nullable Schema by default. A field
that references a shared schema and is nullable must be explicitly
declared as a new NullableSchema(X).
Add UTs to verify the consistency between schema and message
serialization.

Reviewers: Jun Rao junrao@gmail.com, Chia-Ping Tsai
chia7712@gmail.com

junrao

@DL1231 : Thanks for the PR. This is a bit more complicated than I originally thought. Left a few comments.

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

Regarding the implementation of the nullable vs non-nullable types. We use 3 different approaches. (a) For bytes, we implement two independent classes BYTES and NULLABLE_BYTES. (b) For array, we use one class ArraryOf, which takes a nullable param. (c) For schema, we implement NULLABLE_SCHEMA as a subclass of SCHEMA. Is it possible to pick one approach to implement all nullable types in a consistent way? Perhaps (b) or (c) is a bit better since it allows more code sharing.
In the generated html, could we introduce notations for 4 different types of arrays (nullable vs non-nullable, compact vs non-compact)?
This is an existing issue and can probably be done in a separate PR. All static classes in Field except TaggedFieldsSection are not really being used. We can probably remove them.

DL1231 · 2025-10-22T10:06:40Z

@junrao : Thanks for your review.

Is it possible to pick one approach to implement all nullable types in a consistent way?

I think (c) might be more suitable, as it not only allows for more code reuse but also enables better separation of logic between nullable and non-nullable types.
What do you think about addressing this issue in a separate PR? The changes required to modify the implementation of all nullable types might be a bit more involved.

In the generated html, could we introduce notations for 4 different types of arrays (nullable vs non-nullable, compact vs non-compact)?

How about adding the array type after the []? For example:

ConsumerGroupHeartbeat Response (Version: 0) => throttle_time_ms error_code error_message member_id member_epoch heartbeat_interval_ms assignment _tagged_fields 
  throttle_time_ms => INT32
  error_code => INT16
  error_message => COMPACT_NULLABLE_STRING
  member_id => COMPACT_NULLABLE_STRING
  member_epoch => INT32
  heartbeat_interval_ms => INT32
  assignment => NULLABLE_STRUCT [topic_partitions]COMPACT_ARRAY _tagged_fields 
    topic_partitions => STRUCT topic_id [partitions]COMPACT_ARRAY _tagged_fields 
      topic_id => UUID
      partitions => INT32

All static classes in Field except TaggedFieldsSection are not really being used. We can probably remove them.

Filed KAFKA-19822 to track this case.

Should we add STRUCT to the top level? I guess the top level struct can never be null.

I agree that we probably don't need to. As you rightly pointed out, an empty request or response serves no purpose.

chia7712

@DL1231 thanks for this patch

junrao

@DL1231 : Thanks for the updated. PR.

I think (c) might be more suitable, as it not only allows for more code reuse but also enables better separation of logic between nullable and non-nullable types.
What do you think about addressing this issue in a separate PR? The changes required to modify the implementation of all nullable types might be a bit more involved.

Sounds good.

topic_partitions => STRUCT topic_id [partitions]COMPACT_ARRAY _tagged_fields

How about we use [T], [T]?, (T) and (T)? to represent array, nullable array, compacted array and nullable compacted array, respectively?

Also, could we add the STRUCT keyword to the top level schema in the generated html?

Finally, could you rebase the PR to pick up a fix for flaky test #20713?

DL1231 · 2025-10-25T07:12:44Z

@junrao : Thanks for your review.

Filed KAFKA-19833 to track this issue.

How about we use [T], [T]?, (T) and (T)? to represent array, nullable array, compacted array and nullable compacted array, respectively?
Also, could we add the STRUCT keyword to the top level schema in the generated html?

The generated HTML looks like this:

ConsumerGroupHeartbeat Response (Version: 0) => STRUCT throttle_time_ms error_code error_message member_id member_epoch heartbeat_interval_ms assignment 
  throttle_time_ms => INT32
  error_code => INT16
  error_message => COMPACT_NULLABLE_STRING
  member_id => COMPACT_NULLABLE_STRING
  member_epoch => INT32
  heartbeat_interval_ms => INT32
  assignment => NULLABLE_STRUCT (topic_partitions) 
    topic_partitions => STRUCT topic_id (partitions) 
      topic_id => UUID
      partitions => INT32

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao · 2025-10-31T23:04:41Z

        for (Iterator<StructSpec> iter = structRegistry.commonStructs(); iter.hasNext(); ) {
            StructSpec struct = iter.next();
-            generateSchemas(struct.name(), struct, message.struct().versions());
+            generateSchemas(struct.name(), struct, message.struct().versions(), Versions.NONE);


This is a bit problematic. A shared schema could be used by multiple fields. Some of them can be nullable and some others can be non-nullable. Not sure what's the best approach to address this issue. One potential way is to only support Schema for now. The generated code already handles null just with Schema. So far, for non-generated code usage, it seems that there hasn't been a need for a nullable schema. So, we could punt on that until there is a need.

Thanks for pointing this out. Filed KAFKA-19870 to track it.

@DL1231 : This one is important. So, I think we need to get this part right in this PR, instead of a followup one.

So, we could punt on that until there is a need.

Sorry, I misunderstood your point earlier. I will address this issue asap.

A shared schema could be used by multiple fields. Some of them can be nullable and some others can be non-nullable.

@junrao Pardon me, I may be misunderstanding you comment, but IIRC, the common struct does not support nullable property. So using Version.None should be good in this case

@junrao If Y is nullable, then declare that field as new NullableSchema(X);
if Z is non-nullable, then reference X directly.
X, by default, should be declared as new Schema(). WDYT?

Or, we could reject the json file if the common struct is used in both nullable and non-nullable definition. I think this may be reasonable since it should not be “common” if it has different definitions.

Or, we could reject the json file if the common struct is used in both nullable and non-nullable definition. I think this may be reasonable since it should not be “common” if it has different definitions.

This seems arbitrary. If we allow a struct field to be null, it seems that we should allow it regardless of how the struct is defined.

If Y is nullable, then declare that field as new NullableSchema(X);

This feels awkward to me. The generated code explicitly generates code that handles nulls. So, NullableSchema(X) is unnecessary and will likely confuse people.

The current method parameter uses Version.None by default, indicating that the common struct only supports Schema.
Should we keep the existing logic unchanged?

If Y is nullable, then declare that field as new NullableSchema(X);
if Z is non-nullable, then reference X directly.
X, by default, should be declared as new Schema().

@DL1231 : Thinking a bit more. I feel the above solution that you proposed probably works the best. We will need to change the constructor of NullableSchema to take a Schema. We will use this approach for both shared and non-shared schema when generating the classes.

junrao

@DL1231 : Added a couple of more comments.

junrao · 2025-11-18T23:57:34Z

        for (Iterator<StructSpec> iter = structRegistry.commonStructs(); iter.hasNext(); ) {
            StructSpec struct = iter.next();
-            generateSchemas(struct.name(), struct, message.struct().versions());
+            generateSchemas(struct.name(), struct, message.struct().versions(), Versions.NONE);


If Y is nullable, then declare that field as new NullableSchema(X);
if Z is non-nullable, then reference X directly.
X, by default, should be declared as new Schema().

@DL1231 : Thinking a bit more. I feel the above solution that you proposed probably works the best. We will need to change the constructor of NullableSchema to take a Schema. We will use this approach for both shared and non-shared schema when generating the classes.

DL1231 · 2025-11-19T02:48:04Z

@junrao Thanks for the review. I have updated the PR. The generated HTML looks like this:

ConsumerGroupHeartbeat Response (Version: 0) => { throttle_time_ms error_code error_message member_id member_epoch heartbeat_interval_ms assignment }
  throttle_time_ms => INT32
  error_code => INT16
  error_message => COMPACT_NULLABLE_STRING
  member_id => COMPACT_NULLABLE_STRING
  member_epoch => INT32
  heartbeat_interval_ms => INT32
  assignment => ?{ (topic_partitions) }
    topic_partitions => { topic_id (partitions) }
      topic_id => UUID
      partitions => INT32

junrao

@DL1231 : Thanks for the updated PR. A few more comments. Also, could you summarize the changes to generated code and the html doc?

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao

@DL1231 : Thanks for the updated PR. A few more comments.

junrao · 2025-11-26T19:55:07Z

-            RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+            RECORDS, COMPACT_RECORDS, NULLABLE_RECORDS, COMPACT_NULLABLE_RECORDS,
+            new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING), ArrayOf.nullable(STRING), CompactArrayOf.nullable(STRING),
+            new Schema(), new NullableSchema(new Schema())};


This is an existing issue. For COMPACT_BYTES and COMPACT_NULLABLE_BYTES, could you add a space in front of "Then N bytes follow. ?

Also, for all Array types, could we add a period at the end of the documentation to be consistent?

Thanks very much for your detailed and patient review. I have updated the PR—please take another look when you have time.

junrao

@DL1231 : Thanks for the updated PR. LGTM. Since the PR has evolved quite a bit from the original goal, could you adjust the title of the jira/PR to reflect the actual changes?

@chia7712 : Do you want to take another look at the PR?

chia7712 · 2025-11-27T18:38:13Z

Do you want to take another look at the PR?

yes, will take a look later!

chia7712

@DL1231 thanks for this great patch. overall LGTM. I have just one small comment remaining

chia7712 · 2025-11-28T05:41:33Z

+
+    @Override
+    public String leftBracket() {
+        return "?{";


Given that the Array types documentation include the symbol, should it also be included in the documentation for consistency?

ditto for Schema

Thanks for the review, I have updated the PR. PTAL

chia7712 · 2025-11-28T13:46:07Z

the flaky is already traced by https://issues.apache.org/jira/browse/KAFKA-18952

…protocol specification (apache#20614) This patch introduces a clear separation between nullable and non-nullable data structures. The key changes include: 1. Differentiates between nullable and non-nullable versions of `RECORDS`, `COMPACT_RECORDS`, and `Schema` types. 2. Adds explicit nullable type names for `ArrayOf` and `CompactArrayOf`. 3. Introduces a new, concise syntax for representing types: - `{}` for struct, `?{}` for nullable struct - `[T]` for array, `?[T]` for nullable array - `(T)` for compact array, `?(T)` for nullable compact array 4. Declares shared schemas as non-nullable `Schema` by default. A field that references a shared schema and is nullable must be explicitly declared as a new `NullableSchema(X)`. 5. Add UTs to verify the consistency between schema and message serialization. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

KAFKA-19634: document the encoding of nullable struct

fa69309

github-actions Bot added triage PRs from the community clients small Small PRs labels Sep 30, 2025

rename struct

6f1464e

chia7712 reviewed Sep 30, 2025

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java Outdated

junrao reviewed Sep 30, 2025

View reviewed changes

github-actions Bot removed the triage PRs from the community label Oct 1, 2025

add NULLABLE_SCHEMA, NULLABLE_RECORDS, COMPACT_NULLABLE_RECORDS

1caeda2

github-actions Bot added streams generator RPC and Record code generator and removed small Small PRs labels Oct 9, 2025

fix comment

89a2259

junrao added the ci-approved label Oct 15, 2025

junrao reviewed Oct 15, 2025

View reviewed changes

DL1231 added 2 commits October 20, 2025 10:51

fix comment

7d85132

Merge remote-tracking branch 'origin/trunk' into KAFKA-19634

7082547

DL1231 force-pushed the KAFKA-19634 branch from e7a03f0 to 1d4a7cc Compare October 20, 2025 06:16

fix build

2781f65

DL1231 force-pushed the KAFKA-19634 branch from d0ddd2e to 2781f65 Compare October 20, 2025 07:05

junrao reviewed Oct 21, 2025

View reviewed changes

fix comment

126340a

chia7712 reviewed Oct 22, 2025

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java Outdated

Comment thread generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java Outdated

fix comment

78c732d

junrao reviewed Oct 24, 2025

View reviewed changes

DL1231 added 2 commits October 25, 2025 11:41

Merge remote-tracking branch 'origin/trunk' into KAFKA-19634

e08b51f

fix comment

a5b4e42

junrao reviewed Nov 3, 2025

View reviewed changes

Merge remote-tracking branch 'origin/trunk' into KAFKA-19634

8e9e3de

junrao reviewed Nov 19, 2025

View reviewed changes

DL1231 added 2 commits November 19, 2025 10:48

address comment

9c3b19c

Merge remote-tracking branch 'origin/trunk' into KAFKA-19634

1d9037d

junrao reviewed Nov 19, 2025

View reviewed changes

DL1231 added 2 commits November 20, 2025 10:22

address comment

cec1e3d

revert replace

e0a02ef

junrao reviewed Nov 22, 2025

View reviewed changes

add UT

4088462

DL1231 requested a review from junrao November 24, 2025 11:45

DL1231 added 2 commits November 24, 2025 19:47

Merge remote-tracking branch 'origin/trunk' into KAFKA-19634

9af2ee6

fix typo

3bd76d1

junrao reviewed Nov 24, 2025

View reviewed changes

rename UT method

4ef1b1c

junrao reviewed Nov 26, 2025

View reviewed changes

address comment

e8e12ed

junrao reviewed Nov 26, 2025

View reviewed changes

address comment

7d7409f

junrao approved these changes Nov 27, 2025

View reviewed changes

DL1231 changed the title ~~KAFKA-19634: Document the encoding of nullable struct~~ KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification Nov 28, 2025

chia7712 reviewed Nov 28, 2025

View reviewed changes

address comment

1bbce2b

chia7712 approved these changes Nov 28, 2025

View reviewed changes

chia7712 merged commit 58d62d1 into apache:trunk Nov 28, 2025
22 of 24 checks passed

DL1231 mentioned this pull request Nov 29, 2025

KAFKA-19833: Reduce code duplication in nullable protocol types #21019

Merged

Conversation

DL1231 commented Sep 30, 2025 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

DL1231 commented Oct 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

DL1231 commented Oct 25, 2025

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DL1231 Nov 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

DL1231 commented Nov 19, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

DL1231 commented Sep 30, 2025 •

edited by github-actions Bot

Loading

DL1231 commented Oct 22, 2025 •

edited

Loading

DL1231 Nov 10, 2025 •

edited

Loading

DL1231 commented Nov 19, 2025 •

edited

Loading