You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2025/09/30 01:04:15 UTC
[PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
DL1231 opened a new pull request, #20614:
URL: https://github.com/apache/kafka/pull/20614
This patch adds documentation for nullable struct encoding in the protocol specification.
For nullable structs, the encoding is defined as follows:
- The first byte is an INT8 null indicator (-1 for null, 1 for non-null)
- If non-null, the remaining bytes contain the serialization of each field in order
- If null, no additional bytes follow the null indicator
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2443696830
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
<pre>DescribeUserScramCredentials Request (Version: 0) => [users] _tagged_fields
users => NULLABLE_STRUCT name _tagged_fields
name => COMPACT_STRING
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2502203020
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
}
// Generate schemas for inline structures
generateSchemas(message.dataClassName(), message.struct(),
- message.struct().versions());
+ message.struct().versions(), Versions.NONE);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2482868156
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
}
// Generate schemas for inline structures
generateSchemas(message.dataClassName(), message.struct(),
- message.struct().versions());
+ message.struct().versions(), Versions.NONE);
Review Comment:
Could we add a comment that the top-level struct can never be null?
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2422175741
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -916,40 +916,71 @@ public String documentation() {
}
};
- public static final DocumentedType COMPACT_RECORDS = new DocumentedType() {
- @Override
- public boolean isNullable() {
- return true;
- }
+ public static final DocumentedType RECORDS = new DocumentedType() {
@Override
public void write(ByteBuffer buffer, Object o) {
- if (o == null) {
- COMPACT_NULLABLE_BYTES.write(buffer, null);
- } else if (o instanceof MemoryRecords) {
+ if (o instanceof MemoryRecords) {
MemoryRecords records = (MemoryRecords) o;
- COMPACT_NULLABLE_BYTES.write(buffer, records.buffer().duplicate());
+ BYTES.write(buffer, records.buffer().duplicate());
} else {
throw new IllegalArgumentException("Unexpected record type: " + o.getClass());
}
}
@Override
public MemoryRecords read(ByteBuffer buffer) {
- ByteBuffer recordsBuffer = (ByteBuffer) COMPACT_NULLABLE_BYTES.read(buffer);
- if (recordsBuffer == null) {
- return null;
- } else {
- return MemoryRecords.readableRecords(recordsBuffer);
- }
+ ByteBuffer recordsBuffer = (ByteBuffer) BYTES.read(buffer);
+ return MemoryRecords.readableRecords(recordsBuffer);
}
@Override
public int sizeOf(Object o) {
- if (o == null) {
- return 1;
+ BaseRecords records = (BaseRecords) o;
+ return 4 + records.sizeInBytes();
+ }
+
+ @Override
+ public String typeName() {
+ return "RECORDS";
+ }
+
+ @Override
+ public BaseRecords validate(Object item) {
+ if (item instanceof BaseRecords)
Review Comment:
This is an existing issue. Since RECORDS currently only supports MemoryRecords, we probably should validate that too to be consistent.
##########
clients/src/main/resources/common/message/README.md:
##########
@@ -102,7 +104,7 @@ Guide](https://kafka.apache.org/protocol.html).
Nullable Fields
---------------
Booleans, ints, and floats can never be null. However, fields that are strings,
-bytes, uuid, records, or arrays may optionally be "nullable". When a field is
+bytes, records, struct, or arrays may optionally be "nullable". When a field is
Review Comment:
Could we add uuid to the first sentence?
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
Review Comment:
How does NullableSchema fit here?
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
Could you post a few request/response to see how the generated html look like?
##########
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java:
##########
@@ -342,7 +342,7 @@ private static boolean retainsBufferReference(Schema schema) {
Schema.Visitor detector = new Schema.Visitor() {
@Override
public void visit(Type field) {
- if (field == BYTES || field == NULLABLE_BYTES || field == RECORDS ||
+ if (field == BYTES || field == NULLABLE_BYTES || field == NULLABLE_RECORDS ||
Review Comment:
This should include all 4 variants of RECORDS.
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -317,8 +317,11 @@ private String fieldTypeToSchemaType(FieldType type,
fieldTypeToSchemaType(arrayType.elementType(), false, version, fieldFlexibleVersions, false));
}
} else if (type.isStruct()) {
- return String.format("%s.SCHEMA_%d", type,
+ if (nullable)
+ headerGenerator.addImport(MessageGenerator.NULLABLE_SCHEMA_CLASS);
+ String schemaType = String.format("%s.SCHEMA_%d", type,
floorVersion(type.toString(), version));
+ return nullable ? String.format("new NullableSchema(%s)", schemaType) : schemaType;
Review Comment:
This generates the following code. Should we just change Assignment.SCHEMA_0 to be NullableSchema?
` new Field("assignment", new NullableSchema(Assignment.SCHEMA_0), "null if not provided; the assignment otherwise."),`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2525837647
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
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 is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2558156454
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
Review Comment:
It appears that Struct does not have default values. If we don't explicitly set the value, it will throw an exception:
<pre>
Error computing size for field 'my_boolean': Missing value for field 'my_boolean' which has no default value.
org.apache.kafka.common.protocol.types.SchemaException: Error computing size for field 'my_boolean': Missing value for field 'my_boolean' which has no default value.
at app//org.apache.kafka.common.protocol.types.Schema.sizeOf(Schema.java:141)
at app//org.apache.kafka.common.protocol.types.Struct.sizeOf(Struct.java:201)
at app//org.apache.kafka.common.message.ProtocolSerializationConsistencyTest.checkSchemaAndMessageSerializationConsistency(ProtocolSerializationConsistencyTest.java:154)
at app//org.apache.kafka.common.message.ProtocolSerializationConsistencyTest.testNonFlexibleWithNullValue(ProtocolSerializationConsistencyTest.java:99)
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2557562323
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
Review Comment:
Hmm, we are not setting null value here as the method name indicates.
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
Review Comment:
Hmm, myIntArray has no default in version 0. Since the test doesn't set its value in messageData. Why doesn't the serialization fail?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
Review Comment:
This is the default value. Do we need to set it explicitly?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageSerializationConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNotNullValue() {
Review Comment:
NotNullValue => NonNullValue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2570535329
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a byte value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.put((byte) -1);
+ return;
+ }
+
+ buffer.put((byte) 1);
+ super.write(buffer, o);
+ }
+
+ @Override
+ public Struct read(ByteBuffer buffer) {
+ byte nullIndicator = buffer.get();
+ if (nullIndicator < 0)
+ return null;
+
+ return super.read(buffer);
+ }
+
+ @Override
+ public int sizeOf(Object o) {
+ if (o == null)
+ return 1;
+
+ return 1 + super.sizeOf(o);
+ }
+
+ @Override
+ public Struct validate(Object item) {
+ if (item == null)
+ return null;
+
+ return super.validate(item);
+ }
+
+ @Override
+ public String typeName() {
+ return NULLABLE_STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String leftBracket() {
+ return "?{";
Review Comment:
ditto for `Schema`
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a byte value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.put((byte) -1);
+ return;
+ }
+
+ buffer.put((byte) 1);
+ super.write(buffer, o);
+ }
+
+ @Override
+ public Struct read(ByteBuffer buffer) {
+ byte nullIndicator = buffer.get();
+ if (nullIndicator < 0)
+ return null;
+
+ return super.read(buffer);
+ }
+
+ @Override
+ public int sizeOf(Object o) {
+ if (o == null)
+ return 1;
+
+ return 1 + super.sizeOf(o);
+ }
+
+ @Override
+ public Struct validate(Object item) {
+ if (item == null)
+ return null;
+
+ return super.validate(item);
+ }
+
+ @Override
+ public String typeName() {
+ return NULLABLE_STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String leftBracket() {
+ return "?{";
Review Comment:
Given that the `Array` types documentation include the symbol, should it also be included in the documentation for consistency?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2426979670
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
+ new Schema()};
Review Comment:
<img width="988" height="528" alt="20251014-015049-1(WeLinkPC)" src="https://github.com/user-attachments/assets/e13f6ba0-c363-4bfe-8ec6-9a76c7d3f04e" />
<img width="988" height="538" alt="20251014-015049-2(WeLinkPC)" src="https://github.com/user-attachments/assets/bdf578ee-4e68-4ebb-9d72-a3ec4f09089d" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2558144147
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
Review Comment:
Because the type of `myBytes` is `Type.BYTES`, which is non-nullable, while `myNullableBytes` is set to null default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #20614:
URL: https://github.com/apache/kafka/pull/20614#issuecomment-3431520353
@junrao : Thanks for your review.
> 1. 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.
> 2. 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:
<pre>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
</pre>
> 3. All static classes in Field except TaggedFieldsSection are not really being used. We can probably remove them.
Filed [KAFKA-19822](https://issues.apache.org/jira/browse/KAFKA-19822) to track this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2454763009
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -107,15 +107,15 @@ void generateSchemas(String className, StructSpec struct,
for (FieldSpec field : struct.fields()) {
if (field.type().isStructArray()) {
FieldType.ArrayType arrayType = (FieldType.ArrayType) field.type();
- generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions);
+ generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions, field.nullableVersions());
Review Comment:
I think you're right. It doesn't make sense to allow null elements in an array. If we need to represent null value, the NULLABLE_ARRAY should be used.
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -107,15 +107,15 @@ void generateSchemas(String className, StructSpec struct,
for (FieldSpec field : struct.fields()) {
if (field.type().isStructArray()) {
FieldType.ArrayType arrayType = (FieldType.ArrayType) field.type();
- generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions);
+ generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions, field.nullableVersions());
Review Comment:
I think you're right. It doesn't make sense to allow null elements in an array. If we need to represent null value, the `NULLABLE_ARRAY` should be used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #20614:
URL: https://github.com/apache/kafka/pull/20614#issuecomment-3446045043
@junrao : Thanks for your review.
Filed [KAFKA-19833](https://issues.apache.org/jira/browse/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:
<pre>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
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2512368296
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
@chia7712 : If a common struct X is referenced by two fields Y and Z, Y could be nullable and Z could be non-nullable, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2443606042
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
Review Comment:
`NullableSchema` is a subclass of `Schema` and overrides the `typeName()` method. When `typeName()` is called, the overridden version from `NullableSchema` will be invoked.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2566197344
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
+ assertEquals(serializedSchemaBuffer, serializedMessageAccessor.buffer(),
+ "Buffer contents should be identical between message and schema serialization");
+
+ serializedMessageAccessor.flip();
+ // Deserialize message
+ Schema schema = version == 0 ? AllTypeMessageData.SCHEMA_0 : AllTypeMessageData.SCHEMA_1;
+ Struct deserializedMessage = schema.read(serializedMessageAccessor.buffer());
+ assertEquals(struct, deserializedMessage, "Deserialized message struct should match original struct after round trip");
Review Comment:
Deserialized message struct => Deserialized struct ?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
+ assertEquals(serializedSchemaBuffer, serializedMessageAccessor.buffer(),
+ "Buffer contents should be identical between message and schema serialization");
+
+ serializedMessageAccessor.flip();
+ // Deserialize message
+ Schema schema = version == 0 ? AllTypeMessageData.SCHEMA_0 : AllTypeMessageData.SCHEMA_1;
+ Struct deserializedMessage = schema.read(serializedMessageAccessor.buffer());
Review Comment:
deserializedMessage => deserializedStruct?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
+ assertEquals(serializedSchemaBuffer, serializedMessageAccessor.buffer(),
+ "Buffer contents should be identical between message and schema serialization");
+
+ serializedMessageAccessor.flip();
+ // Deserialize message
+ Schema schema = version == 0 ? AllTypeMessageData.SCHEMA_0 : AllTypeMessageData.SCHEMA_1;
+ Struct deserializedMessage = schema.read(serializedMessageAccessor.buffer());
+ assertEquals(struct, deserializedMessage, "Deserialized message struct should match original struct after round trip");
+
+ serializedSchemaBuffer.flip();
+ // Deserialize schema
+ AllTypeMessageData deserializedSchema = new AllTypeMessageData(new ByteBufferAccessor(serializedSchemaBuffer), version);
Review Comment:
deserializedSchema => deserializedMessage?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
Review Comment:
Could we check the buffer is completely filled to struct.sizeOf()?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
Review Comment:
How about "Generated and non-generated schema serializer should serialize to the same length."?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
+ assertEquals(serializedSchemaBuffer, serializedMessageAccessor.buffer(),
+ "Buffer contents should be identical between message and schema serialization");
+
+ serializedMessageAccessor.flip();
+ // Deserialize message
+ Schema schema = version == 0 ? AllTypeMessageData.SCHEMA_0 : AllTypeMessageData.SCHEMA_1;
+ Struct deserializedMessage = schema.read(serializedMessageAccessor.buffer());
+ assertEquals(struct, deserializedMessage, "Deserialized message struct should match original struct after round trip");
+
+ serializedSchemaBuffer.flip();
+ // Deserialize schema
+ AllTypeMessageData deserializedSchema = new AllTypeMessageData(new ByteBufferAccessor(serializedSchemaBuffer), version);
+ assertEquals(message, deserializedSchema, "Deserialized schema message should match original message after round trip");
Review Comment:
Deserialized schema message => Deserialized message ?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1221,10 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- 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())};
Review Comment:
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?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolRoundTripConsistencyTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolRoundTripConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullDefault() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullDefault() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageRoundTripConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageRoundTripConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor serializedMessageAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(serializedMessageAccessor, cache, version);
+
+ ByteBuffer serializedSchemaBuffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(serializedSchemaBuffer);
+
+ assertEquals(serializedSchemaBuffer.position(), serializedMessageAccessor.buffer().position(),
+ "Buffer positions should match between message and schema serialization");
+ assertEquals(serializedSchemaBuffer, serializedMessageAccessor.buffer(),
+ "Buffer contents should be identical between message and schema serialization");
Review Comment:
How about "Generated and non-generated schema serializer should serialize to the same content."?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2393490168
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -24,7 +26,10 @@
/**
* The schema for a compound record definition
*/
-public final class Schema extends Type {
+public final class Schema extends DocumentedType {
+
+ private static final String STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
Review Comment:
> without the null indicator
The documentation for other nullable types also needs to be updated to mention the null indicator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2426970769
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
+ new Schema()};
Review Comment:
<img width="494" height="266" alt="20251014-014552(WeLinkPC)" src="https://github.com/user-attachments/assets/62f94bd5-7b9e-4d39-880d-2509bc509ab1" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2446176795
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -916,40 +916,71 @@ public String documentation() {
}
};
- public static final DocumentedType COMPACT_RECORDS = new DocumentedType() {
- @Override
- public boolean isNullable() {
- return true;
- }
+ public static final DocumentedType RECORDS = new DocumentedType() {
@Override
public void write(ByteBuffer buffer, Object o) {
- if (o == null) {
- COMPACT_NULLABLE_BYTES.write(buffer, null);
- } else if (o instanceof MemoryRecords) {
+ if (o instanceof MemoryRecords) {
MemoryRecords records = (MemoryRecords) o;
- COMPACT_NULLABLE_BYTES.write(buffer, records.buffer().duplicate());
+ BYTES.write(buffer, records.buffer().duplicate());
} else {
throw new IllegalArgumentException("Unexpected record type: " + o.getClass());
}
}
@Override
public MemoryRecords read(ByteBuffer buffer) {
- ByteBuffer recordsBuffer = (ByteBuffer) COMPACT_NULLABLE_BYTES.read(buffer);
- if (recordsBuffer == null) {
- return null;
- } else {
- return MemoryRecords.readableRecords(recordsBuffer);
- }
+ ByteBuffer recordsBuffer = (ByteBuffer) BYTES.read(buffer);
+ return MemoryRecords.readableRecords(recordsBuffer);
}
@Override
public int sizeOf(Object o) {
- if (o == null) {
- return 1;
+ BaseRecords records = (BaseRecords) o;
+ return 4 + records.sizeInBytes();
+ }
+
+ @Override
+ public String typeName() {
+ return "RECORDS";
+ }
+
+ @Override
+ public BaseRecords validate(Object item) {
+ if (item instanceof MemoryRecords)
+ return (BaseRecords) item;
+
+ throw new SchemaException(item + " is not an instance of " + BaseRecords.class.getName());
Review Comment:
We should use MemoryRecords instead BaseRecords.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Field... fs) {
+ this(false, fs);
+ }
+
+ public NullableSchema(boolean tolerateMissingFieldsWithDefaults, Field... fs) {
+ super(tolerateMissingFieldsWithDefaults, fs);
+ }
+
+ public NullableSchema(Schema schema) {
Review Comment:
This is unused.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
Review Comment:
STRUCT_TYPE_NAME => NULLABLE_STRUCT_TYPE_NAME
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
Should we add STRUCT to the top level? I guess the top level struct can never be null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2531330354
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
The current method parameter uses `Version.None` by default, indicating that the common struct only supports Schema.
Should we keep the existing logic unchanged?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2539989419
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
> 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.
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
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
```
Would it be better to use {} and ?{} to represent struct and nullable struct like the following? We will also want to change [T]? and (T)? to ?[T] and ?(T) to be consistent.
```
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 => STRUCT topic_id (partitions)
topic_id => UUID
partitions => INT32
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2505865860
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
@DL1231 : This one is important. So, I think we need to get this part right in this PR, instead of a followup one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2391990917
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
+ new Schema()};
Review Comment:
Can you generate the website with this change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2422446247
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -206,6 +211,19 @@ public Struct validate(Object item) {
}
}
+ @Override
+ public String typeName() {
+ return STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String documentation() {
+ return "Represents a composite object or null. " +
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -24,7 +26,10 @@
/**
* The schema for a compound record definition
*/
-public final class Schema extends Type {
+public final class Schema extends DocumentedType {
+
+ private static final String STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2566909710
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1221,10 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- 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())};
Review Comment:
Thanks very much for your detailed and patient review. I have updated the PR—please take another look when you have time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20614:
URL: https://github.com/apache/kafka/pull/20614
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20614:
URL: https://github.com/apache/kafka/pull/20614#issuecomment-3589404442
the flaky is already traced by https://issues.apache.org/jira/browse/KAFKA-18952
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2512368296
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
@chia7712 : If a field X references a common struct Y, X could be null, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2508426487
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
> So, we could punt on that until there is a need.
Sorry, I misunderstood your point earlier. I will address this issue asap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2529296703
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2543324218
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Field... fs) {
+ this(false, fs);
+ }
+
+ public NullableSchema(boolean tolerateMissingFieldsWithDefaults, Field... fs) {
+ super(tolerateMissingFieldsWithDefaults, fs);
+ }
+
+ public NullableSchema(Schema schema) {
Review Comment:
Could we only keep this constructor?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java:
##########
@@ -119,15 +132,25 @@ public Object[] validate(Object item) {
@Override
public String typeName() {
- return ARRAY_TYPE_NAME;
+ return nullable ? NULLABLE_ARRAY_TYPE_NAME : ARRAY_TYPE_NAME;
}
@Override
public String documentation() {
- return "Represents a sequence of objects of a given type T. " +
+ String doc;
+ if (nullable) {
+ doc = "Represents a sequence of objects of a given type T. " +
"Type T can be either a primitive type (e.g. " + STRING + ") or a structure. " +
"First, the length N is given as an " + INT32 + ". Then N instances of type T follow. " +
"A null array is represented with a length of -1. " +
+ "In protocol documentation a nullable array of T instances is referred to as [T]?.";
Review Comment:
[T]? => ?[T]
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/CompactArrayOf.java:
##########
@@ -125,15 +138,24 @@ public Object[] validate(Object item) {
@Override
public String typeName() {
- return COMPACT_ARRAY_TYPE_NAME;
+ return nullable ? COMPACT_NULLABLE_ARRAY_TYPE_NAME : COMPACT_ARRAY_TYPE_NAME;
}
@Override
public String documentation() {
- return "Represents a sequence of objects of a given type T. " +
+ String doc;
+ if (nullable) {
+ doc = "Represents a sequence of objects of a given type T. " +
"Type T can be either a primitive type (e.g. " + STRING + ") or a structure. " +
"First, the length N + 1 is given as an UNSIGNED_VARINT. Then N instances of type T follow. " +
"A null array is represented with a length of 0. " +
- "In protocol documentation an array of T instances is referred to as [T].";
+ "In protocol documentation a compact nullable array of T instances is referred to as (T)?.";
Review Comment:
[T]? => ?[T]
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -79,6 +79,14 @@ public final boolean isArray() {
return arrayElementType().isPresent();
}
+ public String leftBracket() {
Review Comment:
Could we add the javadoc for the new methods?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -53,6 +58,7 @@ public Schema(Field... fs) {
*
* @throws SchemaException If the given list have duplicate fields
*/
+ @SuppressWarnings("this-escape")
Review Comment:
Why is this needed?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/BoundField.java:
##########
@@ -30,6 +30,10 @@ public BoundField(Field def, Schema schema, int index) {
this.index = index;
}
+ public Field def() {
+ return this.def;
Review Comment:
Do we need this? def is a public field.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -173,6 +179,20 @@ public BoundField[] fields() {
return this.fields;
}
+ public boolean tolerateMissingFieldsWithDefaults() {
Review Comment:
Could this be protected instead of public?
##########
streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java:
##########
@@ -246,7 +246,7 @@ public static void verify(final String kafka, final boolean withRepartitioning,
}
// do not modify: required test output
- System.out.println("ALL-RECORDS-DELIVERED");
+ System.out.println("ALL-NULLABLE_RECORDS-DELIVERED");
Review Comment:
Hmm, why do we need to change this? Are records nullable?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -24,7 +26,10 @@
/**
* The schema for a compound record definition
*/
-public final class Schema extends Type {
+public class Schema extends DocumentedType {
+
Review Comment:
extra line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2544116035
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -53,6 +58,7 @@ public Schema(Field... fs) {
*
* @throws SchemaException If the given list have duplicate fields
*/
+ @SuppressWarnings("this-escape")
Review Comment:
This annotation is needed to suppress a new Java compiler warning about a potential `this` escape during object construction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2443682734
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
<pre>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] _tagged_fields
topic_partitions => STRUCT topic_id [partitions] _tagged_fields
topic_id => UUID
partitions => INT32
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2544144295
##########
streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java:
##########
@@ -246,7 +246,7 @@ public static void verify(final String kafka, final boolean withRepartitioning,
}
// do not modify: required test output
- System.out.println("ALL-RECORDS-DELIVERED");
+ System.out.println("ALL-NULLABLE_RECORDS-DELIVERED");
Review Comment:
You're right, my apologies. This was an inadvertent change from a previous global find/replace. I've reverted it to the original state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2454768605
##########
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java:
##########
@@ -342,9 +344,14 @@ private static boolean retainsBufferReference(Schema schema) {
Schema.Visitor detector = new Schema.Visitor() {
@Override
public void visit(Type field) {
- if (field == BYTES || field == NULLABLE_BYTES || field == RECORDS ||
- field == COMPACT_BYTES || field == COMPACT_NULLABLE_BYTES)
+ // avoid BooleanExpressionComplexity checkstyle warning
Review Comment:
I tried that, but the build failed.
https://github.com/apache/kafka/actions/runs/18641614486/job/53141605094
<pre>Caused by: java.lang.NullPointerException: Cannot invoke "java.util.Set.contains(Object)" because "org.apache.kafka.common.protocol.ApiKeys.BUFFER_FIELDS" is null
> Task :core:genProtocolApiKeyDocs FAILED
at org.apache.kafka.common.protocol.ApiKeys$1.visit(ApiKeys.java:354)</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2443689574
##########
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java:
##########
@@ -66,6 +66,8 @@ private static void schemaToBnfHtml(Schema schema, StringBuilder b, int indentSi
b.append(indentStr);
b.append(entry.getKey());
b.append(" => ");
+ b.append(((Schema) entry.getValue()).typeName());
+ b.append(" ");
Review Comment:
<pre>DescribeDelegationToken Request (Version: 3) => [owners] _tagged_fields
owners => NULLABLE_STRUCT principal_type principal_name _tagged_fields
principal_type => COMPACT_STRING
principal_name => COMPACT_STRING
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2550752381
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a short value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.putShort((short) -1);
+ return;
+ }
+
+ buffer.putShort((short) 1);
+ super.write(buffer, o);
+ }
+
+ @Override
+ public Struct read(ByteBuffer buffer) {
+ short nullIndicator = buffer.getShort();
+ if (nullIndicator < 0)
+ return null;
+
+ return super.read(buffer);
+ }
+
+ @Override
+ public int sizeOf(Object o) {
+ if (o == null)
+ return 2;
+
+ return 2 + super.sizeOf(o);
+ }
+
+ @Override
+ public String typeName() {
+ return NULLABLE_STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String leftBracket() {
+ return "?{";
+ }
+
+ @Override
+ public String rightBracket() {
+ return "}";
+ }
+
+ @Override
+ public String documentation() {
+ return "A struct is named by a string with a capitalized first letter and consists of one or more fields. " +
Review Comment:
struct => nullable struct
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a short value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.putShort((short) -1);
Review Comment:
This is incorrect. We should write a byte instead of a short. Could we write a unit test to cover this? I am thinking that we could create a schema json file with all existing types and verify that the serialized bytes are the same between Schema.write() and Message.write().
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -79,6 +79,20 @@ public final boolean isArray() {
return arrayElementType().isPresent();
}
+ /**
+ * Start flag for special data structure
Review Comment:
How about "For annotation in the generated html protocol doc"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20614:
URL: https://github.com/apache/kafka/pull/20614#issuecomment-3587008764
> Do you want to take another look at the PR?
yes, will take a look later!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2558135668
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
Review Comment:
<img width="568" height="37" alt="image" src="https://github.com/user-attachments/assets/a8f13059-516e-494d-a327-dbe3817d33da" />
Because in the generated code, `myIntArray` defaults to an empty list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2422448619
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.common.protocol.types;
+import org.apache.kafka.common.protocol.types.Type.DocumentedType;
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2392893626
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -206,6 +211,19 @@ public Struct validate(Object item) {
}
}
+ @Override
+ public String typeName() {
+ return STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String documentation() {
+ return "Represents a composite object or null. " +
Review Comment:
We should first say sth like "A struct is named by a string with a capitalized first letter and consists of one or more fields".
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -24,7 +26,10 @@
/**
* The schema for a compound record definition
*/
-public final class Schema extends Type {
+public final class Schema extends DocumentedType {
+
+ private static final String STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
Review Comment:
The current Schema class is written for a struct that's not nullable since the write() method just writes all fields without the null indicator. We should name this STRUCT and create a new NullableStruct version.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
Review Comment:
There are multiple of existing issues with the generated schemas in https://kafka.apache.org/protocol#protocol_api_keys.
1. The current RECORDS class is written as NULLABLE_RECORDS. So, we should rename it NULLABLE_RECORDS. We also need to create a new RECORDS class that's not nullable. In SchemaGenerator, we should distinguish between RECORDS and NULLABLE_RECORDS. Currently, the generated schema only has RECORDS for both nullable and non-nullable fields, which is misleading. Ditto for COMPACT_RECORDS.
2. For ArrayOf, it takes nullable as the input, but its name is always ARRAY. We need to add the nullable name. Ditto for CompactArrayOf. Currently, the generator schema only uses [] to represent an array. We need a way to denote whether it's nullable and compact.
3. When generating the schema for a struct, we just fold in the fields in the struct like the following without indicating whether that struct is nullable or not. We need a way to make that clear.
```
assignment => [topic_partitions] _tagged_fields
topic_partitions => topic_id [partitions] _tagged_fields
topic_id => UUID
partitions => INT32
```
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.common.protocol.types;
+import org.apache.kafka.common.protocol.types.Type.DocumentedType;
Review Comment:
There are a few things that we need to fix in clients/src/main/resources/common/message/README.md.
1. We need to add type struct in Field Types.
2. In Nullable Fields, it says uuid is nullable. This is incorrect (the generator throws an exception if a uuid filed is nullable) and needs to be removed.
3. We need to add struct as a nullable type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2508426487
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
Sorry, I misunderstood your point earlier. I will address this issue asap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2502201619
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
Thanks for pointing this out. Filed [KAFKA-19870](https://issues.apache.org/jira/browse/KAFKA-19870) to track it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #20614:
URL: https://github.com/apache/kafka/pull/20614#issuecomment-3550423393
@junrao Thanks for the review. I have updated the PT. The generated HTML looks like this:
<pre>
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2544116035
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -53,6 +58,7 @@ public Schema(Field... fs) {
*
* @throws SchemaException If the given list have duplicate fields
*/
+ @SuppressWarnings("this-escape")
Review Comment:
This annotation is needed to suppress a new Java compiler warning about a potential `this` escape during object construction.
https://github.com/apache/kafka/actions/runs/18640610101/job/53138690877
<pre>
Execution failed for task ':clients:compileJava'.
> Compilation failed; see the compiler output below.
error: warnings found and -Werror specified
/home/runner/work/kafka/kafka/clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:69: warning: [this-escape] possible 'this' escape before subclass is fully initialized
this.fields[i] = new BoundField(def, this, i);
^
</pre>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2426970769
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
+ new Schema()};
Review Comment:
<img width="494" height="266" alt="20251014-014552(WeLinkPC)" src="https://github.com/user-attachments/assets/62f94bd5-7b9e-4d39-880d-2509bc509ab1" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2451762083
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java:
##########
@@ -173,6 +179,10 @@ public BoundField[] fields() {
return this.fields;
}
+ public boolean tolerateMissingFieldsWithDefaults() {
Review Comment:
this method is unused
##########
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java:
##########
@@ -342,9 +344,14 @@ private static boolean retainsBufferReference(Schema schema) {
Schema.Visitor detector = new Schema.Visitor() {
@Override
public void visit(Type field) {
- if (field == BYTES || field == NULLABLE_BYTES || field == RECORDS ||
- field == COMPACT_BYTES || field == COMPACT_NULLABLE_BYTES)
+ // avoid BooleanExpressionComplexity checkstyle warning
Review Comment:
How about using a `Set` to keep track of those types
```java
private static final Set<FieldType> BUFFER_FIELD_TYPES = Set.of(
BYTES, NULLABLE_BYTES, COMPACT_BYTES, COMPACT_NULLABLE_BYTES,
RECORDS, NULLABLE_RECORDS, COMPACT_RECORDS, COMPACT_NULLABLE_RECORDS
);
if (BUFFER_FIELD_TYPES.contains(field)) {
hasBuffer.set(true);
}
```
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -107,15 +107,15 @@ void generateSchemas(String className, StructSpec struct,
for (FieldSpec field : struct.fields()) {
if (field.type().isStructArray()) {
FieldType.ArrayType arrayType = (FieldType.ArrayType) field.type();
- generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions);
+ generateSchemas(arrayType.elementType().toString(), structRegistry.findStruct(field), versions, field.nullableVersions());
Review Comment:
Does this mean the array can accept a null element? If so, that seems odd, since the protocol write does not check for null on each element in a `List`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Formalize nullable and non-nullable type distinctions in protocol specification [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2570670616
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a byte value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.put((byte) -1);
+ return;
+ }
+
+ buffer.put((byte) 1);
+ super.write(buffer, o);
+ }
+
+ @Override
+ public Struct read(ByteBuffer buffer) {
+ byte nullIndicator = buffer.get();
+ if (nullIndicator < 0)
+ return null;
+
+ return super.read(buffer);
+ }
+
+ @Override
+ public int sizeOf(Object o) {
+ if (o == null)
+ return 1;
+
+ return 1 + super.sizeOf(o);
+ }
+
+ @Override
+ public Struct validate(Object item) {
+ if (item == null)
+ return null;
+
+ return super.validate(item);
+ }
+
+ @Override
+ public String typeName() {
+ return NULLABLE_STRUCT_TYPE_NAME;
+ }
+
+ @Override
+ public String leftBracket() {
+ return "?{";
Review Comment:
Thanks for the review, I have updated the PR. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2561517694
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+
+ checkSchemaAndMessageSerializationConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testNonFlexibleWithNonNullValue() {
+ messageData.setMyBytes("bytes".getBytes());
+ messageData.setMyRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableString("nullable string");
+ messageData.setMyNullableBytes("nullable bytes".getBytes());
+ messageData.setMyNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ nonFlexibleStruct.set("my_nullable_string", "nullable string")
+ .set("my_nullable_bytes", ByteBuffer.wrap("nullable bytes".getBytes()))
+ .set("my_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageSerializationConsistency((short) 0, messageData, nonFlexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCommonStruct(null);
+
+ flexibleStruct.set("my_common_struct", null);
+
+ checkSchemaAndMessageSerializationConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ @Test
+ public void testFlexibleWithNonNullValue() {
+ messageData.setMyCompactBytes("compact bytes".getBytes());
+ messageData.setMyCompactRecords(MemoryRecords.EMPTY);
+ messageData.setMyCompactNullableString("compact nullable string");
+ messageData.setMyCompactNullableBytes("compact nullable bytes".getBytes());
+ messageData.setMyCompactNullableRecords(MemoryRecords.EMPTY);
+ messageData.setMyNullableIntArray(List.of(1, 2, 3));
+
+ flexibleStruct.set("my_compact_nullable_string", "compact nullable string")
+ .set("my_compact_nullable_bytes", ByteBuffer.wrap("compact nullable bytes".getBytes()))
+ .set("my_compact_nullable_records", MemoryRecords.EMPTY)
+ .set("my_nullable_int_array", new Object[] {1, 2, 3});
+
+ checkSchemaAndMessageSerializationConsistency((short) 1, messageData, flexibleStruct);
+ }
+
+ private void checkSchemaAndMessageSerializationConsistency(short version, AllTypeMessageData message, Struct struct) {
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ ByteBuffer buf = ByteBuffer.allocate(message.size(cache, version));
+ ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(buf);
+ // Serialize message
+ message.write(byteBufferAccessor, cache, version);
+
+ ByteBuffer buffer = ByteBuffer.allocate(struct.sizeOf());
+ // Serialize schema
+ struct.writeTo(buffer);
+
+ assertEquals(buffer.position(), byteBufferAccessor.buffer().position());
Review Comment:
Could we add the error message? Ditto below.
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/NullableSchema.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol.types;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * The nullable schema for a compound record definition
+ */
+public final class NullableSchema extends Schema {
+
+ private static final String NULLABLE_STRUCT_TYPE_NAME = "NULLABLE_STRUCT";
+
+ public NullableSchema(Schema schema) {
+ super(schema.tolerateMissingFieldsWithDefaults(), Arrays.stream(schema.fields()).map(field -> field.def).toArray(Field[]::new));
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Write a struct to the buffer with special handling for null values
+ * If the input object is null, writes a byte value of -1 to the buffer as a null indicator.
+ */
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ if (o == null) {
+ buffer.put((byte) -1);
+ return;
+ }
+
+ buffer.put((byte) 1);
+ super.write(buffer, o);
+ }
+
+ @Override
+ public Struct read(ByteBuffer buffer) {
+ short nullIndicator = buffer.getShort();
Review Comment:
This is incorrect. We should read a byte instead of a short. Could we improve the unit test to cover reads too?
##########
clients/src/test/java/org/apache/kafka/common/message/ProtocolSerializationConsistencyTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ProtocolSerializationConsistencyTest {
+
+ private Struct nonFlexibleStruct;
+
+ private Struct flexibleStruct;
+
+ private AllTypeMessageData messageData;
+
+ @BeforeEach
+ public void setup() {
+ nonFlexibleStruct = new Struct(AllTypeMessageData.SCHEMA_0)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_string", "string")
+ .set("my_nullable_string", null)
+ .set("my_bytes", ByteBuffer.wrap("bytes".getBytes()))
+ .set("my_nullable_bytes", null)
+ .set("my_records", MemoryRecords.EMPTY)
+ .set("my_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null);
+ nonFlexibleStruct.set("my_common_struct", nonFlexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123));
+
+ flexibleStruct = new Struct(AllTypeMessageData.SCHEMA_1)
+ .set("my_boolean", false)
+ .set("my_int8", (byte) 12)
+ .set("my_int16", (short) 123)
+ .set("my_uint16", 33000)
+ .set("my_int32", 1234)
+ .set("my_uint32", 1234567L)
+ .set("my_uint64", 0xcafcacafcacafcaL)
+ .set("my_uuid", Uuid.fromString("H3KKO4NTRPaCWtEmm3vW7A"))
+ .set("my_float64", 12.34D)
+ .set("my_compact_string", "compact string")
+ .set("my_compact_nullable_string", null)
+ .set("my_compact_bytes", ByteBuffer.wrap("compact bytes".getBytes()))
+ .set("my_compact_nullable_bytes", null)
+ .set("my_compact_records", MemoryRecords.EMPTY)
+ .set("my_compact_nullable_records", null)
+ .set("my_int_array", new Object[] {})
+ .set("my_nullable_int_array", null)
+ .set("_tagged_fields", new TreeMap<Integer, Field>());
+ flexibleStruct.set("my_common_struct", flexibleStruct.instance("my_common_struct")
+ .set("foo", 123)
+ .set("bar", 123)
+ .set("_tagged_fields", new TreeMap<Integer, Field>()));
+
+ messageData = new AllTypeMessageData();
+ }
+
+ @Test
+ public void testNonFlexibleWithNullValue() {
Review Comment:
WithNullValue => WithNullDefault ?
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java:
##########
@@ -119,15 +132,25 @@ public Object[] validate(Object item) {
@Override
public String typeName() {
- return ARRAY_TYPE_NAME;
+ return nullable ? NULLABLE_ARRAY_TYPE_NAME : ARRAY_TYPE_NAME;
}
@Override
public String documentation() {
- return "Represents a sequence of objects of a given type T. " +
+ String doc;
+ if (nullable) {
+ doc = "Represents a sequence of objects of a given type T. " +
"Type T can be either a primitive type (e.g. " + STRING + ") or a structure. " +
"First, the length N is given as an " + INT32 + ". Then N instances of type T follow. " +
"A null array is represented with a length of -1. " +
+ "In protocol documentation a nullable array of T instances is referred to as ?[T].";
Review Comment:
?[ => leftBracket()
] => rightBracket()
Ditto in a few other places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2525354397
##########
generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java:
##########
@@ -86,16 +86,16 @@ void generateSchemas(MessageSpec message) {
// available when we generate the inline structures
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);
Review Comment:
@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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19634: Document the encoding of nullable struct [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20614:
URL: https://github.com/apache/kafka/pull/20614#discussion_r2422448542
##########
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java:
##########
@@ -1116,7 +1116,8 @@ private static String toHtml() {
UINT16, UNSIGNED_INT32, VARINT, VARLONG, UUID, FLOAT64,
STRING, COMPACT_STRING, NULLABLE_STRING, COMPACT_NULLABLE_STRING,
BYTES, COMPACT_BYTES, NULLABLE_BYTES, COMPACT_NULLABLE_BYTES,
- RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING)};
+ RECORDS, COMPACT_RECORDS, new ArrayOf(STRING), new CompactArrayOf(COMPACT_STRING),
Review Comment:
> We need a way to make that clear.
Added a struct type to indecate whether the struct is nullable or not.
assignment => NULLABLE_STRUCT [topic_partitions] _tagged_fields
topic_partitions => STRUCT topic_id [partitions] _tagged_fields
topic_id => UUID
partitions => INT32
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org