feature: Build schema on upstream version 3.9.0-rc5 (#213)

This bumps schema to 3.9.0-rc5. The intention is that we can release
this as some pre-release, most likely an alpha, pending the upstream
final version.

This PR adds an `@xfail` to the compatibility test of
`UpdateRaftVoterResponse`. This model is hitting #215 which already
exists prior to schema vertsion 3.9.0. That bug should be addressed
separately.

codegen/__init__.py

@@ -2,4 +2,4 @@
 
 # Note: the kafka-clients dependency of the Java tester also needs updating when
 # this is bumped (in java_tester/build.gradle).
-build_tag: Final = "3.8.0"
+build_tag: Final = "3.9.0-rc5"

codegen/generate_tests.py

@@ -47,6 +47,7 @@ def test_{entity_snake_case}_roundtrip(instance: {entity_type}) -> None:
 """
 
 test_code_java = """\
+{xfail}
 @pytest.mark.java
 @given(instance=from_type({entity_type}))
 def test_{entity_snake_case}_java(instance: {entity_type}, java_tester: JavaTester) -> None:
@@ -86,11 +87,22 @@ def main() -> None:
                 entity_snake_case=to_snake_case(entity_type.__name__),
             )
         )
+
         if entity_type.__type__ is not EntityType.nested:
+            xfail = (
+                ""
+                if entity_type.__name__ not in "UpdateRaftVoterResponse"
+                else (
+                    "@pytest.mark.xfail("
+                    'reason="https://github.com/Aiven-Open/kio/issues/215"'
+                    ")"
+                )
+            )
             module_code[module_path].append(
                 test_code_java.format(
                     entity_type=entity_type.__name__,
                     entity_snake_case=to_snake_case(entity_type.__name__),
+                    xfail=xfail,
                 )
             )
 

codegen/parser.py

@@ -105,7 +105,7 @@ def get_type_hint(self, optional: bool = False) -> str:
             case Primitive.bool_:
                 hint = "bool"
             case Primitive.uuid:
-                hint = "uuid.UUID | None"
+                hint = "uuid.UUID"
             case Primitive.error_code:
                 hint = "ErrorCode"
             case Primitive.timedelta_i32:
@@ -117,7 +117,7 @@ def get_type_hint(self, optional: bool = False) -> str:
             case no_match:
                 assert_never(no_match)
 
-        if optional:
+        if optional or self is Primitive.uuid:
             return f"{hint} | None"
         return hint
 

compose.yml

@@ -1,7 +1,7 @@
 name: kio
 services:
   kafka:
-    image: apache/kafka:3.8.0
+    image: apache/kafka:3.9.0-rc5
     hostname: broker
     container_name: broker
     ports:

java_tester/build.gradle

@@ -10,10 +10,14 @@ java {
 
 repositories {
     mavenCentral()
+    // This allows installing release candidates.
+    maven {
+        url = uri("https://repository.apache.org/content/groups/staging/")
+    }
 }
 
 dependencies {
-    implementation 'org.apache.kafka:kafka-clients:3.8.0'
+    implementation 'org.apache.kafka:kafka-clients:3.9.0'
     implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
     implementation 'org.apache.commons:commons-text:1.9'
 }

src/kio/schema/add_raft_voter/v0/__init__.py

@@ -0,0 +1,7 @@
+from .request import AddRaftVoterRequest
+from .response import AddRaftVoterResponse
+
+__all__ = (
+    "AddRaftVoterRequest",
+    "AddRaftVoterResponse",
+)

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

@@ -0,0 +1,48 @@
+"""
+Generated from ``clients/src/main/resources/common/message/AddRaftVoterRequest.json``.
+"""
+
+import uuid
+
+from dataclasses import dataclass
+from dataclasses import field
+from typing import ClassVar
+
+from kio.schema.request_header.v2.header import RequestHeader
+from kio.static.constants import EntityType
+from kio.static.primitive import i16
+from kio.static.primitive import i32
+from kio.static.primitive import i32Timedelta
+from kio.static.primitive import u16
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class Listener:
+    __type__: ClassVar = EntityType.nested
+    __version__: ClassVar[i16] = i16(0)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(80)
+    __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
+    name: str = field(metadata={"kafka_type": "string"})
+    """The name of the endpoint"""
+    host: str = field(metadata={"kafka_type": "string"})
+    """The hostname"""
+    port: u16 = field(metadata={"kafka_type": "uint16"})
+    """The port"""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class AddRaftVoterRequest:
+    __type__: ClassVar = EntityType.request
+    __version__: ClassVar[i16] = i16(0)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(80)
+    __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
+    cluster_id: str | None = field(metadata={"kafka_type": "string"})
+    timeout: i32Timedelta = field(metadata={"kafka_type": "timedelta_i32"})
+    voter_id: i32 = field(metadata={"kafka_type": "int32"})
+    """The replica id of the voter getting added to the topic partition"""
+    voter_directory_id: uuid.UUID | None = field(metadata={"kafka_type": "uuid"})
+    """The directory id of the voter getting added to the topic partition"""
+    listeners: tuple[Listener, ...]
+    """The endpoints that can be used to communicate with the voter"""

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

@@ -0,0 +1,28 @@
+"""
+Generated from ``clients/src/main/resources/common/message/AddRaftVoterResponse.json``.
+"""
+
+from dataclasses import dataclass
+from dataclasses import field
+from typing import ClassVar
+
+from kio.schema.errors import ErrorCode
+from kio.schema.response_header.v1.header import ResponseHeader
+from kio.static.constants import EntityType
+from kio.static.primitive import i16
+from kio.static.primitive import i32Timedelta
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class AddRaftVoterResponse:
+    __type__: ClassVar = EntityType.response
+    __version__: ClassVar[i16] = i16(0)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(80)
+    __header_schema__: ClassVar[type[ResponseHeader]] = ResponseHeader
+    throttle_time: i32Timedelta = field(metadata={"kafka_type": "timedelta_i32"})
+    """The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota."""
+    error_code: ErrorCode = field(metadata={"kafka_type": "error_code"})
+    """The error code, or 0 if there was no error"""
+    error_message: str | None = field(metadata={"kafka_type": "string"})
+    """The error message, or null if there was no error."""

src/kio/schema/api_versions/v3/response.py

@@ -75,7 +75,7 @@ class ApiVersionsResponse:
     supported_features: tuple[SupportedFeatureKey, ...] = field(
         metadata={"tag": 0}, default=()
     )
-    """Features supported by the broker."""
+    """Features supported by the broker. Note: in v0-v3, features with MinSupportedVersion = 0 are omitted."""
     finalized_features_epoch: i64 = field(
         metadata={"kafka_type": "int64", "tag": 1}, default=i64(-1)
     )

src/kio/schema/api_versions/v4/__init__.py

@@ -0,0 +1,7 @@
+from .request import ApiVersionsRequest
+from .response import ApiVersionsResponse
+
+__all__ = (
+    "ApiVersionsRequest",
+    "ApiVersionsResponse",
+)

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

@@ -0,0 +1,24 @@
+"""
+Generated from ``clients/src/main/resources/common/message/ApiVersionsRequest.json``.
+"""
+
+from dataclasses import dataclass
+from dataclasses import field
+from typing import ClassVar
+
+from kio.schema.request_header.v2.header import RequestHeader
+from kio.static.constants import EntityType
+from kio.static.primitive import i16
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiVersionsRequest:
+    __type__: ClassVar = EntityType.request
+    __version__: ClassVar[i16] = i16(4)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(18)
+    __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
+    client_software_name: str = field(metadata={"kafka_type": "string"})
+    """The name of the client."""
+    client_software_version: str = field(metadata={"kafka_type": "string"})
+    """The version of the client."""

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

@@ -0,0 +1,90 @@
+"""
+Generated from ``clients/src/main/resources/common/message/ApiVersionsResponse.json``.
+"""
+
+from dataclasses import dataclass
+from dataclasses import field
+from typing import ClassVar
+
+from kio.schema.errors import ErrorCode
+from kio.schema.response_header.v0.header import ResponseHeader
+from kio.static.constants import EntityType
+from kio.static.primitive import i16
+from kio.static.primitive import i32Timedelta
+from kio.static.primitive import i64
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiVersion:
+    __type__: ClassVar = EntityType.nested
+    __version__: ClassVar[i16] = i16(4)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(18)
+    __header_schema__: ClassVar[type[ResponseHeader]] = ResponseHeader
+    api_key: i16 = field(metadata={"kafka_type": "int16"})
+    """The API index."""
+    min_version: i16 = field(metadata={"kafka_type": "int16"})
+    """The minimum supported version, inclusive."""
+    max_version: i16 = field(metadata={"kafka_type": "int16"})
+    """The maximum supported version, inclusive."""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class SupportedFeatureKey:
+    __type__: ClassVar = EntityType.nested
+    __version__: ClassVar[i16] = i16(4)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(18)
+    __header_schema__: ClassVar[type[ResponseHeader]] = ResponseHeader
+    name: str = field(metadata={"kafka_type": "string"})
+    """The name of the feature."""
+    min_version: i16 = field(metadata={"kafka_type": "int16"})
+    """The minimum supported version for the feature."""
+    max_version: i16 = field(metadata={"kafka_type": "int16"})
+    """The maximum supported version for the feature."""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class FinalizedFeatureKey:
+    __type__: ClassVar = EntityType.nested
+    __version__: ClassVar[i16] = i16(4)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(18)
+    __header_schema__: ClassVar[type[ResponseHeader]] = ResponseHeader
+    name: str = field(metadata={"kafka_type": "string"})
+    """The name of the feature."""
+    max_version_level: i16 = field(metadata={"kafka_type": "int16"})
+    """The cluster-wide finalized max version level for the feature."""
+    min_version_level: i16 = field(metadata={"kafka_type": "int16"})
+    """The cluster-wide finalized min version level for the feature."""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiVersionsResponse:
+    __type__: ClassVar = EntityType.response
+    __version__: ClassVar[i16] = i16(4)
+    __flexible__: ClassVar[bool] = True
+    __api_key__: ClassVar[i16] = i16(18)
+    __header_schema__: ClassVar[type[ResponseHeader]] = ResponseHeader
+    error_code: ErrorCode = field(metadata={"kafka_type": "error_code"})
+    """The top-level error code."""
+    api_keys: tuple[ApiVersion, ...]
+    """The APIs supported by the broker."""
+    throttle_time: i32Timedelta = field(metadata={"kafka_type": "timedelta_i32"})
+    """The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota."""
+    supported_features: tuple[SupportedFeatureKey, ...] = field(
+        metadata={"tag": 0}, default=()
+    )
+    """Features supported by the broker. Note: in v0-v3, features with MinSupportedVersion = 0 are omitted."""
+    finalized_features_epoch: i64 = field(
+        metadata={"kafka_type": "int64", "tag": 1}, default=i64(-1)
+    )
+    """The monotonically increasing epoch for the finalized features information. Valid values are >= 0. A value of -1 is special and represents unknown epoch."""
+    finalized_features: tuple[FinalizedFeatureKey, ...] = field(
+        metadata={"tag": 2}, default=()
+    )
+    """List of cluster-wide finalized features. The information is valid only if FinalizedFeaturesEpoch >= 0."""
+    zk_migration_ready: bool = field(
+        metadata={"kafka_type": "bool", "tag": 3}, default=False
+    )
+    """Set by a KRaft controller if the required configurations for ZK migration are present"""

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

@@ -22,7 +22,7 @@ class PartitionData:
     __api_key__: ClassVar[i16] = i16(53)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
     partition_index: i32 = field(metadata={"kafka_type": "int32"})
-    """The partition index."""
+    """The partition index"""
     leader_id: BrokerId = field(metadata={"kafka_type": "int32"})
     """The ID of the newly elected leader"""
     leader_epoch: i32 = field(metadata={"kafka_type": "int32"})
@@ -37,7 +37,7 @@ class TopicData:
     __api_key__: ClassVar[i16] = i16(53)
     __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
     topic_name: TopicName = field(metadata={"kafka_type": "string"})
-    """The topic name."""
+    """The topic name"""
     partitions: tuple[PartitionData, ...]
 
 

src/kio/schema/begin_quorum_epoch/v1/__init__.py

@@ -0,0 +1,7 @@
+from .request import BeginQuorumEpochRequest
+from .response import BeginQuorumEpochResponse
+
+__all__ = (
+    "BeginQuorumEpochRequest",
+    "BeginQuorumEpochResponse",
+)