You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/29 05:59:06 UTC
[GitHub] [kafka] hachikuji opened a new pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
hachikuji opened a new pull request #9998:
URL: https://github.com/apache/kafka/pull/9998
This patch adds a `RecordSerde` implementation for the metadata record format expected by KIP-631.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569643702
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
Review comment:
The type here leaks over into `RaftClient` and into `RaftManager`, which is really annoying. Having `RaftClient<RecordType>` seems reasonable. An alternative is to change the type to always be `ObjectSerializationCache` since there isn't actually any other use 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569643702
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
Review comment:
The types here leak into `RaftClient` and `RaftManager`, which is really annoying. Having `RaftClient<ApiMessageAndVersion>` seems reasonable, but `RaftClient<ApiMessageAndVersion, ObjectSerializationCache>` is cumbersome. An alternative is to change the type to always be `ObjectSerializationCache` since there isn't actually any other use 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569647498
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
I don't feel too strongly about the use of the unsigned type. I have seen the varint size functions pop up on some flame graphs, so the concern is valid. If it becomes a real issue, one idea would be to precompute the size during the code generation. Really we could even hard-code the value 1 here and then write a test which fails if there are any api keys/versions which do not fit in a single byte.
@cmccabe Yeah, I still think it's probably worth having a version byte or something in case we need to change this in the future. We have regretted not having a header version in the RPC schema.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569643182
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
@cmccabe That really depends on the typical record size. And hence why I asked the question.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#issuecomment-772365541
Is this ready for review @hachikuji?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569704505
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
Review comment:
I pushed a commit to do this. Let me know if that addresses the concern.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569805968
##########
File path: raft/src/test/java/org/apache/kafka/raft/metadata/MetadataRecordSerdeTest.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.raft.metadata;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.metadata.TopicRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class MetadataRecordSerdeTest {
+
+ @Test
+ public void testSerde() {
+ TopicRecord topicRecord = new TopicRecord()
+ .setName("foo")
+ .setTopicId(Uuid.randomUuid());
+
+ MetadataRecordSerde serde = new MetadataRecordSerde();
+
+ for (short version = TopicRecord.LOWEST_SUPPORTED_VERSION; version <= TopicRecord.HIGHEST_SUPPORTED_VERSION; version++) {
+ ApiMessageAndVersion messageAndVersion = new ApiMessageAndVersion(topicRecord, version);
+
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ int size = serde.recordSize(messageAndVersion, cache);
+
+ ByteBuffer buffer = ByteBuffer.allocate(size);
+ ByteBufferAccessor bufferAccessor = new ByteBufferAccessor(buffer);
+
+ serde.write(messageAndVersion, cache, bufferAccessor);
+ buffer.flip();
+
+ assertEquals(size, buffer.remaining());
+ ApiMessageAndVersion readMessageAndVersion = serde.read(bufferAccessor, size);
+ assertEquals(messageAndVersion, readMessageAndVersion);
+ }
+ }
+
+ @Test
+ public void testDeserializeWithUnhandledFrameVersion() {
+ ByteBuffer buffer = ByteBuffer.allocate(16);
+ ByteUtils.writeUnsignedVarint(15, buffer);
+ buffer.flip();
+
+ MetadataRecordSerde serde = new MetadataRecordSerde();
+ assertThrows(SerializationException.class,
+ () -> serde.read(new ByteBufferAccessor(buffer), 16));
+ }
+
+}
Review comment:
Nod. Will fix in the follow-up mentioned above. Did not want to trigger another build.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569278561
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
Review comment:
Nit: seems a bit redundant. Can we not assign the size in the line below?
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
Review comment:
Why do we cast instead of having the context type as a type parameter?
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
I know this is in the KIP but is it really worth deviating from the request format (two shorts) to save two bytes for low api key/version? We end up with additional complexity on writes/reads (see size code above for example).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569754497
##########
File path: raft/src/main/java/org/apache/kafka/raft/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.raft.metadata;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+ private static final short DEFAULT_FRAME_VERSION = 0;
+ private static final int DEFAULT_FRAME_VERSION_SIZE = ByteUtils.sizeOfUnsignedVarint(DEFAULT_FRAME_VERSION);
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, ObjectSerializationCache serializationCache) {
+ int size = DEFAULT_FRAME_VERSION_SIZE;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, ObjectSerializationCache serializationCache, Writable out) {
+ out.writeUnsignedVarint(DEFAULT_FRAME_VERSION);
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short frameVersion = (short) input.readUnsignedVarint();
Review comment:
Do we need to verify that the cast is safe?
##########
File path: raft/src/test/java/org/apache/kafka/raft/metadata/MetadataRecordSerdeTest.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.raft.metadata;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.metadata.TopicRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class MetadataRecordSerdeTest {
+
+ @Test
+ public void testSerde() {
+ TopicRecord topicRecord = new TopicRecord()
+ .setName("foo")
+ .setTopicId(Uuid.randomUuid());
+
+ MetadataRecordSerde serde = new MetadataRecordSerde();
+
+ for (short version = TopicRecord.LOWEST_SUPPORTED_VERSION; version <= TopicRecord.HIGHEST_SUPPORTED_VERSION; version++) {
+ ApiMessageAndVersion messageAndVersion = new ApiMessageAndVersion(topicRecord, version);
+
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ int size = serde.recordSize(messageAndVersion, cache);
+
+ ByteBuffer buffer = ByteBuffer.allocate(size);
+ ByteBufferAccessor bufferAccessor = new ByteBufferAccessor(buffer);
+
+ serde.write(messageAndVersion, cache, bufferAccessor);
+ buffer.flip();
+
+ assertEquals(size, buffer.remaining());
+ ApiMessageAndVersion readMessageAndVersion = serde.read(bufferAccessor, size);
+ Assertions.assertEquals(messageAndVersion, readMessageAndVersion);
Review comment:
Nit: remove `Assertions.` prefix.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569633608
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
It is a big savings overall, right? 2 bytes per record, for every record. We use varints in the record format for the same reasons.
Actually the question that @hachikuji raised earlier was whether we should have a third varint which comes first for message frame version, which is not unreasonable.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
cmccabe commented on pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#issuecomment-772836974
LGTM
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569802761
##########
File path: raft/src/test/java/org/apache/kafka/raft/metadata/MetadataRecordSerdeTest.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.raft.metadata;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.metadata.TopicRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class MetadataRecordSerdeTest {
+
+ @Test
+ public void testSerde() {
+ TopicRecord topicRecord = new TopicRecord()
+ .setName("foo")
+ .setTopicId(Uuid.randomUuid());
+
+ MetadataRecordSerde serde = new MetadataRecordSerde();
+
+ for (short version = TopicRecord.LOWEST_SUPPORTED_VERSION; version <= TopicRecord.HIGHEST_SUPPORTED_VERSION; version++) {
+ ApiMessageAndVersion messageAndVersion = new ApiMessageAndVersion(topicRecord, version);
+
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ int size = serde.recordSize(messageAndVersion, cache);
+
+ ByteBuffer buffer = ByteBuffer.allocate(size);
+ ByteBufferAccessor bufferAccessor = new ByteBufferAccessor(buffer);
+
+ serde.write(messageAndVersion, cache, bufferAccessor);
+ buffer.flip();
+
+ assertEquals(size, buffer.remaining());
+ ApiMessageAndVersion readMessageAndVersion = serde.read(bufferAccessor, size);
+ assertEquals(messageAndVersion, readMessageAndVersion);
+ }
+ }
+
+ @Test
+ public void testDeserializeWithUnhandledFrameVersion() {
+ ByteBuffer buffer = ByteBuffer.allocate(16);
+ ByteUtils.writeUnsignedVarint(15, buffer);
+ buffer.flip();
+
+ MetadataRecordSerde serde = new MetadataRecordSerde();
+ assertThrows(SerializationException.class,
+ () -> serde.read(new ByteBufferAccessor(buffer), 16));
+ }
+
+}
Review comment:
Missing newline character.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569765201
##########
File path: raft/src/main/java/org/apache/kafka/raft/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.raft.metadata;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+ private static final short DEFAULT_FRAME_VERSION = 0;
+ private static final int DEFAULT_FRAME_VERSION_SIZE = ByteUtils.sizeOfUnsignedVarint(DEFAULT_FRAME_VERSION);
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, ObjectSerializationCache serializationCache) {
+ int size = DEFAULT_FRAME_VERSION_SIZE;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, ObjectSerializationCache serializationCache, Writable out) {
+ out.writeUnsignedVarint(DEFAULT_FRAME_VERSION);
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short frameVersion = (short) input.readUnsignedVarint();
Review comment:
Yeah, we should have a helper here. I will create a follow-up.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569743611
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
I filed this JIRA: https://issues.apache.org/jira/browse/KAFKA-12286.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569643702
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
Review comment:
The type here leaks over into `RaftClient` and into `RaftManager`, which is really annoying. Having `RaftClient<ApiMessageAndVersion>` seems reasonable, but `RaftClient<ApiMessageAndVersion, ObjectSerializationCache>` is cumbersome. An alternative is to change the type to always be `ObjectSerializationCache` since there isn't actually any other use 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569704286
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
Review comment:
Fixed. I did it that way because the asymmetry in the following calls kind of annoyed me 😆 .
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9998:
URL: https://github.com/apache/kafka/pull/9998#discussion_r569707559
##########
File path: metadata/src/main/java/org/apache/kafka/metadata/MetadataRecordSerde.java
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ObjectSerializationCache;
+import org.apache.kafka.common.protocol.Readable;
+import org.apache.kafka.common.protocol.Writable;
+import org.apache.kafka.common.utils.ByteUtils;
+import org.apache.kafka.raft.RecordSerde;
+
+public class MetadataRecordSerde implements RecordSerde<ApiMessageAndVersion> {
+
+ @Override
+ public ObjectSerializationCache newWriteContext() {
+ return new ObjectSerializationCache();
+ }
+
+ @Override
+ public int recordSize(ApiMessageAndVersion data, Object context) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ int size = 0;
+ size += ByteUtils.sizeOfUnsignedVarint(data.message().apiKey());
+ size += ByteUtils.sizeOfUnsignedVarint(data.version());
+ size += data.message().size(serializationCache, data.version());
+ return size;
+ }
+
+ @Override
+ public void write(ApiMessageAndVersion data, Object context, Writable out) {
+ ObjectSerializationCache serializationCache = (ObjectSerializationCache) context;
+ out.writeUnsignedVarint(data.message().apiKey());
+ out.writeUnsignedVarint(data.version());
+ data.message().write(out, serializationCache, data.version());
+ }
+
+ @Override
+ public ApiMessageAndVersion read(Readable input, int size) {
+ short apiKey = (short) input.readUnsignedVarint();
+ short version = (short) input.readUnsignedVarint();
Review comment:
OK, so with the frame version, we're now talking about 6 bytes. I'm happy to go with this and we can re-evaluate before 3.0 once we have clear data on storage savings vs cpu impact. It may be worth filing a JIRA so that we don't forget.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9998: KAFKA-12250; Add metadata record serde for KIP-631
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9998:
URL: https://github.com/apache/kafka/pull/9998
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org