You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mingyen066 (via GitHub)" <gi...@apache.org> on 2026/02/10 02:24:42 UTC
[PR] KAFKA-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
mingyen066 opened a new pull request, #21439:
URL: https://github.com/apache/kafka/pull/21439
Modified From #19110
1. Convert ControlRecordType schema to use auto-generated protocol `ControlRecordTypeSchema`.
2. Substitute `CURRENT_CONTROL_RECORD_KEY_SIZE` with `controlRecordKeySize()` method since the size is accumulated from `ControlRecordTypeSchema`.
3. Add buffer to `ControlRecordType` to avoid twice compute from `recordKey()` and `controlRecordKeySize()`.
4. flexibleVersions is set to none.
5. Make `parseTypeId` private. Callers use `ControlRecordType.parse(ByteBuffer)` instead.
Co-Authored-By: @dengziming
--
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-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #21439:
URL: https://github.com/apache/kafka/pull/21439#discussion_r2785543423
##########
core/src/main/scala/kafka/tools/DumpLogSegments.scala:
##########
@@ -355,7 +354,7 @@ object DumpLogSegments {
val voters = ControlRecordUtils.deserializeVotersRecord(record)
print(s" KRaftVoters ${VotersRecordJsonConverter.write(voters, voters.version())}")
case controlType =>
- print(s" controlType: $controlType($controlTypeId)")
+ print(s" controlType: $controlType")
Review Comment:
Since `parseTypeId` is now private and callers use `parse()` instead, `DumpLogSegments.printControlRecord` and `MemoryRecordsBuilder.appendControlRecordWithOffset` no longer have access to the raw `typeId`. This means the numeric
`typeId` is no longer included in the output/exception message (e.g., `UNKNOWN(337)` becomes `UNKNOWN`). Would it be acceptable to lose this debug info, or should we keep `parseTypeId` accessible?
--
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-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #21439:
URL: https://github.com/apache/kafka/pull/21439#discussion_r2804735091
##########
clients/src/main/java/org/apache/kafka/common/record/internal/ControlRecordType.java:
##########
@@ -59,46 +58,45 @@ public enum ControlRecordType {
private static final Logger log = LoggerFactory.getLogger(ControlRecordType.class);
- static final short CURRENT_CONTROL_RECORD_KEY_VERSION = 0;
- static final int CURRENT_CONTROL_RECORD_KEY_SIZE = 4;
- private static final Schema CONTROL_RECORD_KEY_SCHEMA_VERSION_V0 = new Schema(
- new Field("version", Type.INT16),
- new Field("type", Type.INT16));
-
private final short type;
+ private final ByteBuffer buffer;
ControlRecordType(short type) {
this.type = type;
+ ControlRecordTypeSchema schema = new ControlRecordTypeSchema().setType(type);
+ buffer = MessageUtil.toVersionPrefixedByteBuffer(ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION, schema);
}
public short type() {
return type;
}
- public Struct recordKey() {
+ public ByteBuffer recordKey() {
if (this == UNKNOWN)
throw new IllegalArgumentException("Cannot serialize UNKNOWN control record type");
+ return buffer.duplicate();
+ }
- Struct struct = new Struct(CONTROL_RECORD_KEY_SCHEMA_VERSION_V0);
- struct.set("version", CURRENT_CONTROL_RECORD_KEY_VERSION);
- struct.set("type", type);
- return struct;
+ public int controlRecordKeySize() {
+ return buffer.remaining();
}
public static short parseTypeId(ByteBuffer key) {
- if (key.remaining() < CURRENT_CONTROL_RECORD_KEY_SIZE)
- throw new InvalidRecordException("Invalid value size found for end control record key. Must have " +
- "at least " + CURRENT_CONTROL_RECORD_KEY_SIZE + " bytes, but found only " + key.remaining());
-
- short version = key.getShort(0);
- if (version < 0)
+ // We should duplicate the original buffer since it will be read again in some cases, for example,
+ // read by KafkaRaftClient and RaftClient.Listener
+ ByteBuffer buffer = key.duplicate();
+ short version = buffer.getShort();
Review Comment:
Sure
--
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-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #21439:
URL: https://github.com/apache/kafka/pull/21439
--
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-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
Posted by "mingyen066 (via GitHub)" <gi...@apache.org>.
mingyen066 commented on code in PR #21439:
URL: https://github.com/apache/kafka/pull/21439#discussion_r2785634360
##########
core/src/main/scala/kafka/tools/DumpLogSegments.scala:
##########
@@ -355,7 +354,7 @@ object DumpLogSegments {
val voters = ControlRecordUtils.deserializeVotersRecord(record)
print(s" KRaftVoters ${VotersRecordJsonConverter.write(voters, voters.version())}")
case controlType =>
- print(s" controlType: $controlType($controlTypeId)")
+ print(s" controlType: $controlType")
Review Comment:
Keep parseTypeId public first.
--
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-10863: Convert ControRecordType schema to use auto-generated protocol [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21439:
URL: https://github.com/apache/kafka/pull/21439#discussion_r2797833306
##########
clients/src/main/java/org/apache/kafka/common/record/internal/ControlRecordType.java:
##########
@@ -59,46 +58,45 @@ public enum ControlRecordType {
private static final Logger log = LoggerFactory.getLogger(ControlRecordType.class);
- static final short CURRENT_CONTROL_RECORD_KEY_VERSION = 0;
- static final int CURRENT_CONTROL_RECORD_KEY_SIZE = 4;
- private static final Schema CONTROL_RECORD_KEY_SCHEMA_VERSION_V0 = new Schema(
- new Field("version", Type.INT16),
- new Field("type", Type.INT16));
-
private final short type;
+ private final ByteBuffer buffer;
ControlRecordType(short type) {
this.type = type;
+ ControlRecordTypeSchema schema = new ControlRecordTypeSchema().setType(type);
+ buffer = MessageUtil.toVersionPrefixedByteBuffer(ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION, schema);
}
public short type() {
return type;
}
- public Struct recordKey() {
+ public ByteBuffer recordKey() {
if (this == UNKNOWN)
throw new IllegalArgumentException("Cannot serialize UNKNOWN control record type");
+ return buffer.duplicate();
+ }
- Struct struct = new Struct(CONTROL_RECORD_KEY_SCHEMA_VERSION_V0);
- struct.set("version", CURRENT_CONTROL_RECORD_KEY_VERSION);
- struct.set("type", type);
- return struct;
+ public int controlRecordKeySize() {
+ return buffer.remaining();
}
public static short parseTypeId(ByteBuffer key) {
- if (key.remaining() < CURRENT_CONTROL_RECORD_KEY_SIZE)
- throw new InvalidRecordException("Invalid value size found for end control record key. Must have " +
- "at least " + CURRENT_CONTROL_RECORD_KEY_SIZE + " bytes, but found only " + key.remaining());
-
- short version = key.getShort(0);
- if (version < 0)
+ // We should duplicate the original buffer since it will be read again in some cases, for example,
+ // read by KafkaRaftClient and RaftClient.Listener
+ ByteBuffer buffer = key.duplicate();
+ short version = buffer.getShort();
Review Comment:
Could you consider wrapping the error from `BufferUnderflowException` to `InvalidRecordException` for compatibility?
--
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