Handle null arrays

codegen/generate_schema.py

@@ -364,8 +364,14 @@ def generate_entity_array_field(
     field: EntityArrayField,
     version: int,
 ) -> str:
+    optional = (
+        field.nullableVersions.matches(version) if field.nullableVersions else False
+    )
     field_call = format_array_field_call(field, version)
-    return f"    {to_snake_case(field.name)}: tuple[{field.type}, ...]{field_call}\n"
+    opt = " | None" if optional else ""
+    return (
+        f"    {to_snake_case(field.name)}: tuple[{field.type}, ...]{opt}{field_call}\n"
+    )
 
 
 def entity_annotation(field: EntityField | CommonStructField, optional: bool) -> str:
@@ -413,9 +419,13 @@ def generate_common_struct_array_field(
     field: CommonStructArrayField,
     version: int,
 ) -> str:
+    optional = (
+        field.nullableVersions.matches(version) if field.nullableVersions else False
+    )
     field_call = format_array_field_call(field, version)
+    opt = " | None" if optional else ""
     return (
-        f"    {to_snake_case(field.name)}: tuple[{field.type.struct.name}, ...]"
+        f"    {to_snake_case(field.name)}: tuple[{field.type.struct.name}, ...]{opt}"
         f"{field_call}\n"
     )
 

java_tester/src/main/java/io/aiven/kio/java_tester/CollectionCreator.java

@@ -1,5 +1,6 @@
 package io.aiven.kio.java_tester;
 
+import com.fasterxml.jackson.databind.node.NullNode;
 import java.lang.reflect.Method;
 import java.util.AbstractCollection;
 import java.util.ArrayList;
@@ -17,7 +18,7 @@ class CollectionCreator extends BaseCreator {
     CollectionCreator(RootMessageInfo rootMessageInfo,
                       JsonNode fieldValue, String fieldName, Schema fieldSchema) throws Exception {
         super(rootMessageInfo);
-        if (!fieldValue.isArray()) {
+        if (!fieldValue.isArray() && !fieldValue.isNull()) {
             throw new Exception("The value of " + fieldName + " must be array but was " + fieldValue);
         }
         this.fieldValue = fieldValue;
@@ -28,6 +29,9 @@ class CollectionCreator extends BaseCreator {
     AbstractCollection<Object> createAbstractCollection(
         Class<AbstractCollection<Object>> collectionClazz
     ) throws Exception {
+        if (fieldValue.isNull()) {
+            return null;
+        }
         AbstractCollection<Object> collection = collectionClazz.getDeclaredConstructor().newInstance();
         Class<?> elementClazz = getCollectionElementClass(collectionClazz);
         fillCollectionFromChildren(elementClazz, collection);
@@ -47,6 +51,9 @@ private static Class<Object> getCollectionElementClass(Class<AbstractCollection<
     }
 
     List<?> createList() throws Exception {
+        if (fieldValue.isNull()) {
+            return null;
+        }
         final String elementTypeInSchema;
         {
             String tmp = fieldSchema.type();

src/kio/schema/consumer_group_heartbeat/v0/request.py

@@ -87,7 +87,7 @@ class ConsumerGroupHeartbeatRequest:
     """null if it didn't change since the last heartbeat; the subscribed topic regex otherwise"""
     server_assignor: str | None = field(metadata={"kafka_type": "string"}, default=None)
     """null if not used or if it didn't change since the last heartbeat; the server side assignor to use otherwise."""
-    client_assignors: tuple[Assignor, ...]
+    client_assignors: tuple[Assignor, ...] | None
     """null if not used or if it didn't change since the last heartbeat; the list of client-side assignors otherwise."""
-    topic_partitions: tuple[TopicPartitions, ...]
+    topic_partitions: tuple[TopicPartitions, ...] | None
     """null if it didn't change since the last heartbeat; the partitions owned by the member."""

src/kio/schema/create_partitions/v0/request.py

@@ -41,7 +41,7 @@ class CreatePartitionsTopic:
     """The topic name."""
     count: i32 = field(metadata={"kafka_type": "int32"})
     """The new partition count."""
-    assignments: tuple[CreatePartitionsAssignment, ...]
+    assignments: tuple[CreatePartitionsAssignment, ...] | None
     """The new partition assignments."""
 
 

src/kio/schema/create_partitions/v1/request.py

@@ -41,7 +41,7 @@ class CreatePartitionsTopic:
     """The topic name."""
     count: i32 = field(metadata={"kafka_type": "int32"})
     """The new partition count."""
-    assignments: tuple[CreatePartitionsAssignment, ...]
+    assignments: tuple[CreatePartitionsAssignment, ...] | None
     """The new partition assignments."""
 
 

src/kio/schema/create_partitions/v2/request.py

@@ -41,7 +41,7 @@ class CreatePartitionsTopic:
     """The topic name."""
     count: i32 = field(metadata={"kafka_type": "int32"})
     """The new partition count."""
-    assignments: tuple[CreatePartitionsAssignment, ...]
+    assignments: tuple[CreatePartitionsAssignment, ...] | None
     """The new partition assignments."""
 
 

src/kio/schema/create_partitions/v3/request.py

@@ -41,7 +41,7 @@ class CreatePartitionsTopic:
     """The topic name."""
     count: i32 = field(metadata={"kafka_type": "int32"})
     """The new partition count."""
-    assignments: tuple[CreatePartitionsAssignment, ...]
+    assignments: tuple[CreatePartitionsAssignment, ...] | None
     """The new partition assignments."""
 
 

src/kio/schema/create_topics/v5/response.py

@@ -58,7 +58,7 @@ class CreatableTopicResult:
     """Number of partitions of the topic."""
     replication_factor: i16 = field(metadata={"kafka_type": "int16"}, default=i16(-1))
     """Replication factor of the topic."""
-    configs: tuple[CreatableTopicConfigs, ...]
+    configs: tuple[CreatableTopicConfigs, ...] | None
     """Configuration of the topic."""
 
 

src/kio/schema/create_topics/v6/response.py

@@ -58,7 +58,7 @@ class CreatableTopicResult:
     """Number of partitions of the topic."""
     replication_factor: i16 = field(metadata={"kafka_type": "int16"}, default=i16(-1))
     """Replication factor of the topic."""
-    configs: tuple[CreatableTopicConfigs, ...]
+    configs: tuple[CreatableTopicConfigs, ...] | None
     """Configuration of the topic."""
 
 

src/kio/schema/create_topics/v7/response.py

@@ -62,7 +62,7 @@ class CreatableTopicResult:
     """Number of partitions of the topic."""
     replication_factor: i16 = field(metadata={"kafka_type": "int16"}, default=i16(-1))
     """Replication factor of the topic."""
-    configs: tuple[CreatableTopicConfigs, ...]
+    configs: tuple[CreatableTopicConfigs, ...] | None
     """Configuration of the topic."""
 
 

src/kio/schema/describe_client_quotas/v0/response.py

@@ -68,5 +68,5 @@ class DescribeClientQuotasResponse:
     """The error code, or `0` if the quota description succeeded."""
     error_message: str | None = field(metadata={"kafka_type": "string"})
     """The error message, or `null` if the quota description succeeded."""
-    entries: tuple[EntryData, ...]
+    entries: tuple[EntryData, ...] | None
     """A result entry."""

src/kio/schema/describe_client_quotas/v1/response.py

@@ -68,5 +68,5 @@ class DescribeClientQuotasResponse:
     """The error code, or `0` if the quota description succeeded."""
     error_message: str | None = field(metadata={"kafka_type": "string"})
     """The error message, or `null` if the quota description succeeded."""
-    entries: tuple[EntryData, ...]
+    entries: tuple[EntryData, ...] | None
     """A result entry."""

src/kio/schema/describe_delegation_token/v0/request.py

@@ -33,5 +33,5 @@ class DescribeDelegationTokenRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(41)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    owners: tuple[DescribeDelegationTokenOwner, ...]
+    owners: tuple[DescribeDelegationTokenOwner, ...] | None
     """Each owner that we want to describe delegation tokens for, or null to describe all tokens."""

src/kio/schema/describe_delegation_token/v1/request.py

@@ -33,5 +33,5 @@ class DescribeDelegationTokenRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(41)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    owners: tuple[DescribeDelegationTokenOwner, ...]
+    owners: tuple[DescribeDelegationTokenOwner, ...] | None
     """Each owner that we want to describe delegation tokens for, or null to describe all tokens."""

src/kio/schema/describe_delegation_token/v2/request.py

@@ -33,5 +33,5 @@ class DescribeDelegationTokenRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(41)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    owners: tuple[DescribeDelegationTokenOwner, ...]
+    owners: tuple[DescribeDelegationTokenOwner, ...] | None
     """Each owner that we want to describe delegation tokens for, or null to describe all tokens."""

src/kio/schema/describe_delegation_token/v3/request.py

@@ -33,5 +33,5 @@ class DescribeDelegationTokenRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(41)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    owners: tuple[DescribeDelegationTokenOwner, ...]
+    owners: tuple[DescribeDelegationTokenOwner, ...] | None
     """Each owner that we want to describe delegation tokens for, or null to describe all tokens."""

src/kio/schema/describe_log_dirs/v0/request.py

@@ -35,5 +35,5 @@ class DescribeLogDirsRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(35)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[DescribableLogDirTopic, ...]
+    topics: tuple[DescribableLogDirTopic, ...] | None
     """Each topic that we want to describe log directories for, or null for all topics."""

src/kio/schema/describe_log_dirs/v1/request.py

@@ -35,5 +35,5 @@ class DescribeLogDirsRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(35)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[DescribableLogDirTopic, ...]
+    topics: tuple[DescribableLogDirTopic, ...] | None
     """Each topic that we want to describe log directories for, or null for all topics."""

src/kio/schema/describe_log_dirs/v2/request.py

@@ -35,5 +35,5 @@ class DescribeLogDirsRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(35)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[DescribableLogDirTopic, ...]
+    topics: tuple[DescribableLogDirTopic, ...] | None
     """Each topic that we want to describe log directories for, or null for all topics."""

src/kio/schema/describe_log_dirs/v3/request.py

@@ -35,5 +35,5 @@ class DescribeLogDirsRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(35)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[DescribableLogDirTopic, ...]
+    topics: tuple[DescribableLogDirTopic, ...] | None
     """Each topic that we want to describe log directories for, or null for all topics."""

src/kio/schema/describe_log_dirs/v4/request.py

@@ -35,5 +35,5 @@ class DescribeLogDirsRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(35)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[DescribableLogDirTopic, ...]
+    topics: tuple[DescribableLogDirTopic, ...] | None
     """Each topic that we want to describe log directories for, or null for all topics."""

src/kio/schema/describe_user_scram_credentials/v0/request.py

@@ -31,5 +31,5 @@ class DescribeUserScramCredentialsRequest:
     __flexible__: ClassVar[bool] = True
     __api_key__: ClassVar[i16] = i16(50)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    users: tuple[UserName, ...]
+    users: tuple[UserName, ...] | None
     """The users to describe, or null/empty to describe all users."""

src/kio/schema/elect_leaders/v0/request.py

@@ -38,7 +38,7 @@ class ElectLeadersRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(43)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topic_partitions: tuple[TopicPartitions, ...]
+    topic_partitions: tuple[TopicPartitions, ...] | None
     """The topic partitions to elect leaders."""
     timeout: i32Timedelta = field(
         metadata={"kafka_type": "timedelta_i32"},

src/kio/schema/elect_leaders/v1/request.py

@@ -41,7 +41,7 @@ class ElectLeadersRequest:
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
     election_type: i8 = field(metadata={"kafka_type": "int8"})
     """Type of elections to conduct for the partition. A value of '0' elects the preferred replica. A value of '1' elects the first live replica if there are no in-sync replica."""
-    topic_partitions: tuple[TopicPartitions, ...]
+    topic_partitions: tuple[TopicPartitions, ...] | None
     """The topic partitions to elect leaders."""
     timeout: i32Timedelta = field(
         metadata={"kafka_type": "timedelta_i32"},

src/kio/schema/elect_leaders/v2/request.py

@@ -41,7 +41,7 @@ class ElectLeadersRequest:
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
     election_type: i8 = field(metadata={"kafka_type": "int8"})
     """Type of elections to conduct for the partition. A value of '0' elects the preferred replica. A value of '1' elects the first live replica if there are no in-sync replica."""
-    topic_partitions: tuple[TopicPartitions, ...]
+    topic_partitions: tuple[TopicPartitions, ...] | None
     """The topic partitions to elect leaders."""
     timeout: i32Timedelta = field(
         metadata={"kafka_type": "timedelta_i32"},

src/kio/schema/fetch/v10/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v11/response.py

@@ -50,7 +50,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     preferred_read_replica: BrokerId = field(
         metadata={"kafka_type": "int32"}, default=BrokerId(-1)

src/kio/schema/fetch/v12/response.py

@@ -94,7 +94,7 @@ class PartitionData:
     )
     snapshot_id: SnapshotId = field(metadata={"tag": 2}, default=SnapshotId())
     """In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     preferred_read_replica: BrokerId = field(
         metadata={"kafka_type": "int32"}, default=BrokerId(-1)

src/kio/schema/fetch/v13/response.py

@@ -95,7 +95,7 @@ class PartitionData:
     )
     snapshot_id: SnapshotId = field(metadata={"tag": 2}, default=SnapshotId())
     """In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     preferred_read_replica: BrokerId = field(
         metadata={"kafka_type": "int32"}, default=BrokerId(-1)

src/kio/schema/fetch/v14/response.py

@@ -95,7 +95,7 @@ class PartitionData:
     )
     snapshot_id: SnapshotId = field(metadata={"tag": 2}, default=SnapshotId())
     """In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     preferred_read_replica: BrokerId = field(
         metadata={"kafka_type": "int32"}, default=BrokerId(-1)

src/kio/schema/fetch/v15/response.py

@@ -95,7 +95,7 @@ class PartitionData:
     )
     snapshot_id: SnapshotId = field(metadata={"tag": 2}, default=SnapshotId())
     """In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     preferred_read_replica: BrokerId = field(
         metadata={"kafka_type": "int32"}, default=BrokerId(-1)

src/kio/schema/fetch/v4/response.py

@@ -47,7 +47,7 @@ class PartitionData:
     """The current high water mark."""
     last_stable_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v5/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v6/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v7/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v8/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/fetch/v9/response.py

@@ -49,7 +49,7 @@ class PartitionData:
     """The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)"""
     log_start_offset: i64 = field(metadata={"kafka_type": "int64"}, default=i64(-1))
     """The current log start offset."""
-    aborted_transactions: tuple[AbortedTransaction, ...]
+    aborted_transactions: tuple[AbortedTransaction, ...] | None
     """The aborted transactions."""
     records: bytes | None = field(metadata={"kafka_type": "records"})
     """The record data."""

src/kio/schema/list_partition_reassignments/v0/request.py

@@ -45,5 +45,5 @@ class ListPartitionReassignmentsRequest:
         default=i32Timedelta.parse(datetime.timedelta(milliseconds=60000)),
     )
     """The time in ms to wait for the request to complete."""
-    topics: tuple[ListPartitionReassignmentsTopics, ...]
+    topics: tuple[ListPartitionReassignmentsTopics, ...] | None
     """The topics to list partition reassignments for, or null to list everything."""

src/kio/schema/metadata/v1/request.py

@@ -32,5 +32,5 @@ class MetadataRequest:
     __flexible__: ClassVar[bool] = False
     __api_key__: ClassVar[i16] = i16(3)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
-    topics: tuple[MetadataRequestTopic, ...]
+    topics: tuple[MetadataRequestTopic, ...] | None
     """The topics to fetch metadata for."""