feature: Generate error codes from source

.github/workflows/ci.yaml

@@ -117,7 +117,7 @@ jobs:
  - uses: actions/checkout@v4
 
  - name: Start Kafka backend
- run: docker compose --file=container/compose.yml up -d
+ run: docker compose up -d
 
  - name: Set up Python
  uses: actions/setup-python@v5
@@ -166,7 +166,7 @@ jobs:
  key: schema-${{ hashFiles('codegen/fetch_schema.py') }}
  restore-keys: schema
 
- - run: pip install --upgrade .[all]
+ - run: pip install --upgrade -e .[all]
 
  - run: make fetch-schema-src
 

.gitignore

@@ -135,7 +135,9 @@ dmypy.json
 # pytype static type analyzer
 .pytype/
 
+# IDE
+/.idea
+
 # Project specific
 /schema/
-
-/.idea
+error-codes.txt

Makefile

@@ -4,7 +4,9 @@ fetch-schema-src:
 
 .PHONY: generate-schema
 generate-schema:
- python3 -m codegen
+ docker compose build java_tester
+ docker compose run --rm java_tester --print-error-codes > error-codes.txt
+ python3 -m codegen error-codes.txt
  pre-commit run --all-files || true
 
 .PHONY: build-schema

README.md

@@ -49,7 +49,7 @@ Apache Kafka® instance running on `127.0.0.1:9092`. There is a Docker Compose f
 `container/compose.yml` that you can use to conveniently start up an instance.
 
 ```shell
-$ docker compose --file=container/compose.yml up -d
+$ docker compose up -d kafka
 ```
 
 Run tests.

benchmarks/roundtrip-serialization.py

@@ -6,6 +6,7 @@
 
 import pyperf
 
+from kio.schema.errors import ErrorCode
 from kio.schema.metadata.v12 import MetadataResponse
 from kio.schema.metadata.v12.response import MetadataResponseBroker
 from kio.schema.metadata.v12.response import MetadataResponsePartition
@@ -14,7 +15,6 @@
 from kio.schema.types import TopicName
 from kio.serial import entity_reader
 from kio.serial import entity_writer
-from kio.static.constants import ErrorCode
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
 

codegen/__main__.py

@@ -1,8 +1,12 @@
+from . import generate_error_codes
 from . import generate_index
 from . import generate_schema
 from . import generate_tests
+from . import recreate_schema_path
 
 if __name__ == "__main__":
+ recreate_schema_path.main()
+ generate_error_codes.main()
  generate_schema.main()
  generate_tests.main()
  generate_index.main()

codegen/generate_error_codes.py

@@ -0,0 +1,106 @@
+# ruff: noqa: T201
+
+import pathlib
+import sys
+
+from dataclasses import dataclass
+from typing import Final
+from typing import Self
+
+from .introspect_schema import base_dir
+
+target_path: Final = base_dir / "src/kio/schema/errors.py"
+
+indent: Final = " "
+module_setup: Final = """\
+from __future__ import annotations
+
+import enum
+
+from typing import TYPE_CHECKING
+
+from kio.static.primitive import i16
+
+
+class ErrorCode(enum.IntEnum):
+ retriable: bool
+ value: i16
+
+ # Note: Pragma is needed to ignore the negative branch, the branch where the
+ # conditional check fails.
+ if not TYPE_CHECKING: # pragma: no cover
+
+ def __new__(cls, value: int, retriable: bool) -> ErrorCode:
+ normalized_value = i16(value)
+ obj = int.__new__(cls, normalized_value)
+ obj._value_ = normalized_value
+ obj.retriable = retriable
+ return obj
+
+"""
+
+
+def parse_name(value: str) -> str:
+ name = value.lower()
+ if not str.isidentifier(name):
+ raise ValueError(f"{value!r} is not a valid name")
+ return name
+
+
+def parse_bool(value: str) -> bool:
+ if value == "True":
+ return True
+ elif value == "False":
+ return False
+ else:
+ raise ValueError(f"{value!r} is not a valid bool")
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ErrorCode:
+ code: int
+ name: str
+ retriable: bool
+ message: str
+
+ @classmethod
+ def parse_line(cls, line: str) -> Self:
+ code, name, retriable, message = line.strip().split(" ", 3)
+ return cls(
+ code=int(code),
+ name=parse_name(name),
+ retriable=parse_bool(retriable),
+ message=message,
+ )
+
+
+def main() -> None:
+ try:
+ source_path = pathlib.Path(sys.argv[1])
+ except (IndexError, ValueError):
+ print(
+ "Error: must be called with path to extracted error codes as single argument",
+ file=sys.stderr,
+ )
+ raise SystemExit(1) from None
+
+ print("Generating error codes.", file=sys.stderr)
+
+ with (
+ source_path.open() as source_fd,
+ target_path.open("w") as target_fd,
+ ):
+ print(module_setup, file=target_fd)
+
+ for line in source_fd.readlines():
+ code = ErrorCode.parse_line(line)
+ print(
+ f"{indent}{code.name} = {code.code}, {code.retriable}",
+ file=target_fd,
+ )
+ if code.code != 0:
+ print(f'{indent}"""{code.message}"""', file=target_fd)
+
+
+if __name__ == "__main__":
+ main()

codegen/generate_schema.py

@@ -4,7 +4,6 @@
 from __future__ import annotations
 
 import pathlib
-import shutil
 import textwrap
 
 from collections import defaultdict
@@ -39,6 +38,7 @@
 from .parser import PrimitiveArrayType
 from .parser import PrimitiveField
 from .parser import parse_file
+from .util import create_package
 
 schema_repository_source: Final = "clients/src/main/resources/common/message/"
 imports_and_docstring: Final = '''\
@@ -50,6 +50,7 @@
 from dataclasses import dataclass, field
 from typing import Annotated, ClassVar
 import uuid
+from kio.schema.errors import ErrorCode
 from kio.static.primitive import i8
 from kio.static.primitive import i16
 from kio.static.primitive import i32
@@ -63,7 +64,6 @@
 from kio.static.primitive import i64Timedelta
 from kio.static.primitive import TZAware
 from kio.static.primitive import Records
-from kio.static.constants import ErrorCode
 from kio.static.constants import EntityType
 '''
 
@@ -607,11 +607,6 @@ def basic_name(schema_name: str) -> str:
  return to_snake_case(schema_name).removesuffix("_response").removesuffix("_request")
 
 
-def create_package(path: pathlib.Path) -> None:
- path.mkdir(exist_ok=True)
- (path / "__init__.py").touch(exist_ok=True)
-
-
 seen_custom_types = set[str]()
 custom_type_imports = """\
 from typing import NewType
@@ -697,8 +692,6 @@ def finalize_exports() -> None:
 def main() -> None:
  schema_output_path = pathlib.Path("src/kio/schema/")
  types_module_path = schema_output_path / "types.py"
- shutil.rmtree(schema_output_path)
- create_package(schema_output_path)
  schemas = (pathlib.Path("schema") / build_tag).glob("*.json")
  custom_types = set[CustomTypeDef]()
 

codegen/introspect_schema.py

@@ -8,8 +8,6 @@
 from typing import TYPE_CHECKING
 from typing import Final
 
-import kio.schema
-
 from kio.static.constants import EntityType
 
 # Chickens and eggs ...
@@ -18,7 +16,7 @@
 
 
 base_dir: Final = Path(__file__).parent.parent.resolve()
-schema_src_dir: Final = Path(kio.schema.__file__).parent.resolve()
+schema_src_dir: Final = base_dir / "src/kio/schema"
 ignore_modules: Final = frozenset({"index"})
 
 

codegen/recreate_schema_path.py

@@ -0,0 +1,22 @@
+# ruff: noqa: T201
+
+import pathlib
+import shutil
+import sys
+
+from .util import create_package
+
+
+def main() -> None:
+ print("Deleting existing schema directory.", file=sys.stderr)
+
+ schema_output_path = pathlib.Path("src/kio/schema/")
+ try:
+ shutil.rmtree(schema_output_path)
+ except FileNotFoundError:
+ pass
+ create_package(schema_output_path)
+
+
+if __name__ == "__main__":
+ main()

codegen/util.py

@@ -1,7 +1,14 @@
+import pathlib
+
 import pydantic
 
 
 class BaseModel(
  pydantic.BaseModel,
  allow_mutation=False,
 ): ...
+
+
+def create_package(path: pathlib.Path) -> None:
+ path.mkdir(exist_ok=True)
+ (path / "__init__.py").touch(exist_ok=True)

container/compose.yml → compose.yml

@@ -28,3 +28,7 @@ services:
  KAFKA_REPLICA_FETCH_MAX_BYTES: 104857600
  tmpfs:
  - /tmp
+
+ java_tester:
+ build:
+ context: java_tester

java_tester/Dockerfile

@@ -5,9 +5,11 @@ WORKDIR /java_tester
 ADD src src
 ADD build.gradle build.gradle
 
-RUN gradle --no-daemon distTar && tar -xvf build/distributions/java_tester.tar
+RUN --mount=type=cache,target=/root/.gradle \
+ gradle --no-daemon distTar \
+ && tar -xvf build/distributions/java_tester.tar
 
 FROM eclipse-temurin:17-jre
 COPY --from=build /java_tester/java_tester /java_tester
 
-CMD /java_tester/bin/java_tester
+ENTRYPOINT ["/java_tester/bin/java_tester"]

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

@@ -8,8 +8,10 @@
 import java.util.Arrays;
 import java.util.Base64;
 
+import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ObjectSerializationCache;
 import org.apache.kafka.common.protocol.Readable;
 
@@ -23,7 +25,28 @@ public class JavaTester {
 
  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
+ private static final String CMD_PRINT_ERROR_CODES = "--print-error-codes";
+
  public static void main(String[] args) throws Exception {
+ if (Arrays.stream(args).anyMatch(CMD_PRINT_ERROR_CODES::equals)) {
+ System.err.println("Printing error codes");
+
+ for (Errors error : Errors.values()) {
+ System.out.print(error.code());
+ System.out.print(" ");
+ System.out.print(error.name());
+ System.out.print(" ");
+ // https://github.com/apache/kafka/blob/84b2d5bedf4381ebfcf32a4671f096de937beb31/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java#L551
+ System.out.print(
+ error.exception() != null && error.exception() instanceof RetriableException ? "True" : "False"
+ );
+ System.out.print(" ");
+ System.out.println(error.message());
+ }
+
+ return;
+ }
+
  System.out.println("Java tester started");
  System.out.flush();
  try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {

pyproject.toml

@@ -135,6 +135,8 @@ extend-ignore = [
  "E501",
  # Allow function calls in argument defaults.
  "B008",
+ # Bad advice, builtin syntax is simpler than imports.
+ "SIM105",
 ]
 
 [tool.ruff.lint.isort]

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

@@ -6,9 +6,9 @@
 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.constants import ErrorCode
 from kio.static.primitive import i16
 from kio.static.primitive import i32Timedelta
 

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

@@ -6,9 +6,9 @@
 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.constants import ErrorCode
 from kio.static.primitive import i16
 from kio.static.primitive import i32Timedelta