chore: Enable Java compat tests for records types (#176)

Fixes #100 (and unlocks first non-alpha release 🎉).

codegen/generate_schema.py

@@ -67,6 +67,7 @@
 from kio.static.primitive import i32Timedelta
 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
 '''

codegen/generate_tests.py

@@ -86,15 +86,7 @@ def main() -> None:
  entity_snake_case=to_snake_case(entity_type.__name__),
  )
  )
- if (
- entity_type.__type__ is not EntityType.nested
- and entity_type.__name__
- not in {
- "ProduceRequest", # Records
- "FetchResponse", # Records
- "FetchSnapshotResponse", # Records
- }
- ):
+ if entity_type.__type__ is not EntityType.nested:
  module_code[module_path].append(
  test_code_java.format(
  entity_type=entity_type.__name__,

codegen/parser.py

@@ -98,8 +98,10 @@ def get_type_hint(self, optional: bool = False) -> str:
  hint = "f64"
  case Primitive.string:
  hint = "str"
- case Primitive.bytes_ | Primitive.records:
+ case Primitive.bytes_:
  hint = "bytes"
+ case Primitive.records:
+ hint = "Records"
  case Primitive.bool_:
  hint = "bool"
  case Primitive.uuid:

container/compose.yml

@@ -1,4 +1,3 @@
-version: '2'
 name: kio
 services:
  # Adopted from

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

@@ -7,6 +7,7 @@
 
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.record.BaseRecords;
+import org.apache.kafka.common.record.MemoryRecords;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.commons.text.CaseUtils;
@@ -88,7 +89,10 @@ T create(JsonNode json) throws Exception {
  List<?> list = new CollectionCreator(rootMessageInfo, fieldValue, fieldName, fieldSchema).createList();
  setter.invoke(instance, list);
  } else if (BaseRecords.class.isAssignableFrom(parameterType)) {
- throw new Exception("Not implemented");
+ ByteBuffer buffer = getByteBuffer(fieldValue, fieldName);
+ if (buffer != null) {
+ setter.invoke(instance, MemoryRecords.readableRecords(buffer));
+ }
  } else {
  Object o = new ObjectCreator<>(
  rootMessageInfo, new EntityClass<>(parameterType), fieldSchema).create(fieldValue);

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

@@ -12,6 +12,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i64
@@ -30,7 +31,7 @@ class PartitionData:
  """The error code, or 0 if there was no fetch error."""
  high_watermark: i64 = field(metadata={"kafka_type": "int64"})
  """The current high water mark."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -12,6 +12,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -31,7 +32,7 @@ class PartitionData:
  """The error code, or 0 if there was no fetch error."""
  high_watermark: i64 = field(metadata={"kafka_type": "int64"})
  """The current high water mark."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -14,6 +14,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -56,7 +57,7 @@ class PartitionData:
  metadata={"kafka_type": "int32"}, default=BrokerId(-1)
  )
  """The preferred read replica for the consumer to use on its next fetch request"""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -14,6 +14,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -100,7 +101,7 @@ class PartitionData:
  metadata={"kafka_type": "int32"}, default=BrokerId(-1)
  )
  """The preferred read replica for the consumer to use on its next fetch request"""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -15,6 +15,7 @@
 from kio.schema.types import ProducerId
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -101,7 +102,7 @@ class PartitionData:
  metadata={"kafka_type": "int32"}, default=BrokerId(-1)
  )
  """The preferred read replica for the consumer to use on its next fetch request"""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -15,6 +15,7 @@
 from kio.schema.types import ProducerId
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -101,7 +102,7 @@ class PartitionData:
  metadata={"kafka_type": "int32"}, default=BrokerId(-1)
  )
  """The preferred read replica for the consumer to use on its next fetch request"""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -15,6 +15,7 @@
 from kio.schema.types import ProducerId
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -101,7 +102,7 @@ class PartitionData:
  metadata={"kafka_type": "int32"}, default=BrokerId(-1)
  )
  """The preferred read replica for the consumer to use on its next fetch request"""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -12,6 +12,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -31,7 +32,7 @@ class PartitionData:
  """The error code, or 0 if there was no fetch error."""
  high_watermark: i64 = field(metadata={"kafka_type": "int64"})
  """The current high water mark."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -12,6 +12,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -31,7 +32,7 @@ class PartitionData:
  """The error code, or 0 if there was no fetch error."""
  high_watermark: i64 = field(metadata={"kafka_type": "int64"})
  """The current high water mark."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -49,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)"""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -51,7 +52,7 @@ class PartitionData:
  """The current log start offset."""
  aborted_transactions: tuple[AbortedTransaction, ...] | None
  """The aborted transactions."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data."""
 
 

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

@@ -13,6 +13,7 @@
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
 from kio.static.constants import ErrorCode
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -61,7 +62,7 @@ class PartitionSnapshot:
  """The total size of the snapshot."""
  position: i64 = field(metadata={"kafka_type": "int64"})
  """The starting byte position within the snapshot included in the Bytes field."""
- unaligned_records: bytes = field(metadata={"kafka_type": "records"})
+ unaligned_records: Records = field(metadata={"kafka_type": "records"})
  """Snapshot data in records format which may not be aligned on an offset boundary"""
 
 

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

@@ -11,6 +11,7 @@
 from kio.schema.request_header.v1.header import RequestHeader
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -25,7 +26,7 @@ class PartitionProduceData:
  __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
  index: i32 = field(metadata={"kafka_type": "int32"})
  """The partition index."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data to be produced."""
 
 

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

@@ -11,6 +11,7 @@
 from kio.schema.request_header.v1.header import RequestHeader
 from kio.schema.types import TopicName
 from kio.static.constants import EntityType
+from kio.static.primitive import Records
 from kio.static.primitive import i16
 from kio.static.primitive import i32
 from kio.static.primitive import i32Timedelta
@@ -25,7 +26,7 @@ class PartitionProduceData:
  __header_schema__: ClassVar[type[RequestHeader]] = RequestHeader
  index: i32 = field(metadata={"kafka_type": "int32"})
  """The partition index."""
- records: bytes | None = field(metadata={"kafka_type": "records"})
+ records: Records | None = field(metadata={"kafka_type": "records"})
  """The record data to be produced."""