Skip to content

feat(inkless): Parse request and response of InitDisklessLog #545

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeqo merged 2 commits into from
Mar 20, 2026
Merged

feat(inkless): Parse request and response of InitDisklessLog #545

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e8b9947
feat(inkless): Parse request and response of InitDisklessLog
giuseppelillo Mar 20, 2026
493c288
fixup! Add test in RequestResponseTest
giuseppelillo Mar 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
return AlterShareGroupOffsetsRequest.parse(readable, apiVersion);
case DELETE_SHARE_GROUP_OFFSETS:
return DeleteShareGroupOffsetsRequest.parse(readable, apiVersion);
case INIT_DISKLESS_LOG:
return InitDisklessLogRequest.parse(readable, apiVersion);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
"code should be updated to do so.", apiKey));
Expand Down
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Readable readable,
return AlterShareGroupOffsetsResponse.parse(readable, version);
case DELETE_SHARE_GROUP_OFFSETS:
return DeleteShareGroupOffsetsResponse.parse(readable, version);
case INIT_DISKLESS_LOG:
return InitDisklessLogResponse.parse(readable, version);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
"code should be updated to do so.", apiKey));
Expand Down
45 changes: 45 additions & 0 deletions clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,8 @@
import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfig;
import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResponse;
import org.apache.kafka.common.message.InitDisklessLogRequestData;
import org.apache.kafka.common.message.InitDisklessLogResponseData;
import org.apache.kafka.common.message.InitProducerIdRequestData;
import org.apache.kafka.common.message.InitProducerIdResponseData;
import org.apache.kafka.common.message.InitializeShareGroupStateRequestData;
Expand Down Expand Up @@ -1075,6 +1077,7 @@ private AbstractRequest getRequest(ApiKeys apikey, short version) {
case DESCRIBE_SHARE_GROUP_OFFSETS: return createDescribeShareGroupOffsetsRequest(version);
case ALTER_SHARE_GROUP_OFFSETS: return createAlterShareGroupOffsetsRequest(version);
case DELETE_SHARE_GROUP_OFFSETS: return createDeleteShareGroupOffsetsRequest(version);
case INIT_DISKLESS_LOG: return createInitDisklessLogRequest(version);
default: throw new IllegalArgumentException("Unknown API key " + apikey);
}
}
Expand Down Expand Up @@ -1170,6 +1173,7 @@ private AbstractResponse getResponse(ApiKeys apikey, short version) {
case DESCRIBE_SHARE_GROUP_OFFSETS: return createDescribeShareGroupOffsetsResponse();
case ALTER_SHARE_GROUP_OFFSETS: return createAlterShareGroupOffsetsResponse();
case DELETE_SHARE_GROUP_OFFSETS: return createDeleteShareGroupOffsetsResponse();
case INIT_DISKLESS_LOG: return createInitDisklessLogResponse();
default: throw new IllegalArgumentException("Unknown API key " + apikey);
}
}
Expand Down Expand Up @@ -1263,6 +1267,47 @@ private AssignReplicasToDirsResponse createAssignReplicasToDirsResponse() {
return new AssignReplicasToDirsResponse(data);
}

private InitDisklessLogRequest createInitDisklessLogRequest(short version) {
InitDisklessLogRequestData data = new InitDisklessLogRequestData()
.setBrokerId(1)
.setBrokerEpoch(123L)
.setTopics(singletonList(
new InitDisklessLogRequestData.TopicData()
.setTopicId(Uuid.fromString("qo0Pcp70TdGnAa7YKMKCqw"))
.setPartitions(singletonList(
new InitDisklessLogRequestData.PartitionData()
.setPartitionId(0)
.setDisklessStartOffset(100L)
.setLeaderEpoch(5)
.setProducerStates(singletonList(
new InitDisklessLogRequestData.ProducerState()
.setProducerId(1000L)
.setProducerEpoch((short) 0)
.setBaseSequence(0)
.setLastSequence(10)
.setAssignedOffset(100L)
.setBatchMaxTimestamp(System.currentTimeMillis())
))
))
));
return new InitDisklessLogRequest.Builder(data).build(version);
}

private InitDisklessLogResponse createInitDisklessLogResponse() {
InitDisklessLogResponseData data = new InitDisklessLogResponseData()
.setThrottleTimeMs(123)
.setTopics(singletonList(
new InitDisklessLogResponseData.TopicResponse()
.setTopicId(Uuid.fromString("qo0Pcp70TdGnAa7YKMKCqw"))
.setPartitions(singletonList(
new InitDisklessLogResponseData.PartitionResponse()
.setPartitionId(0)
.setErrorCode(Errors.NONE.code())
))
));
return new InitDisklessLogResponse(data);
}

private DescribeTopicPartitionsRequest createDescribeTopicPartitionsRequest(short version) {
DescribeTopicPartitionsRequestData data = new DescribeTopicPartitionsRequestData()
.setTopics(singletonList(new DescribeTopicPartitionsRequestData.TopicRequest().setName("foo")))
Expand Down
Loading