Skip to content

feat(s3): support get streams' offset range #38

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
merged 2 commits into from
Aug 31, 2023
Merged

feat(s3): support get streams' offset range #38

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
0624640
feat(s3): support get streams' offset range
TheR1sing3un Aug 31, 2023
5e26339
style(s3): replace array with list
TheR1sing3un Aug 31, 2023
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
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,8 +118,8 @@ public enum ApiKeys {
CLOSE_STREAM(ApiMessageType.CLOSE_STREAM, false, true),
PREPARE_S3_OBJECT(ApiMessageType.PREPARE_S3_OBJECT, false, true),
COMMIT_WALOBJECT(ApiMessageType.COMMIT_WALOBJECT, false, true),
COMMIT_COMPACT_OBJECT(ApiMessageType.COMMIT_COMPACT_OBJECT, false, true),
COMMIT_STREAM_OBJECT(ApiMessageType.COMMIT_STREAM_OBJECT, false, true);
COMMIT_STREAM_OBJECT(ApiMessageType.COMMIT_STREAM_OBJECT, false, true),
GET_STREAMS_OFFSET(ApiMessageType.GET_STREAMS_OFFSET, false, true);
// Kafka on S3 inject end

private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
Expand Down
4 changes: 3 additions & 1 deletion ...a/common/requests/CloseStreamRequest.java → ...ommon/requests/s3/CloseStreamRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.CloseStreamRequestData;
import org.apache.kafka.common.message.CloseStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class CloseStreamRequest extends AbstractRequest {

Expand Down
3 changes: 2 additions & 1 deletion .../common/requests/CloseStreamResponse.java → ...mmon/requests/s3/CloseStreamResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.CloseStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class CloseStreamResponse extends AbstractResponse {

Expand Down
4 changes: 3 additions & 1 deletion ...n/requests/CommitStreamObjectRequest.java → ...equests/s3/CommitStreamObjectRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.CommitStreamObjectRequestData;
import org.apache.kafka.common.message.CommitStreamObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class CommitStreamObjectRequest extends AbstractRequest {

Expand Down
3 changes: 2 additions & 1 deletion .../requests/CommitStreamObjectResponse.java → ...quests/s3/CommitStreamObjectResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.CommitStreamObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class CommitStreamObjectResponse extends AbstractResponse {

Expand Down
4 changes: 3 additions & 1 deletion ...mmon/requests/CommitWALObjectRequest.java → ...n/requests/s3/CommitWALObjectRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.CommitWALObjectRequestData;
import org.apache.kafka.common.message.CommitWALObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class CommitWALObjectRequest extends AbstractRequest {

Expand Down
3 changes: 2 additions & 1 deletion ...mon/requests/CommitWALObjectResponse.java → .../requests/s3/CommitWALObjectResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.CommitWALObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class CommitWALObjectResponse extends AbstractResponse {

Expand Down
4 changes: 3 additions & 1 deletion .../common/requests/DeleteStreamRequest.java → ...mmon/requests/s3/DeleteStreamRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.DeleteStreamRequestData;
import org.apache.kafka.common.message.DeleteStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class DeleteStreamRequest extends AbstractRequest {
public static class Builder extends AbstractRequest.Builder<DeleteStreamRequest> {
Expand Down
3 changes: 2 additions & 1 deletion ...common/requests/DeleteStreamResponse.java → ...mon/requests/s3/DeleteStreamResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.DeleteStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class DeleteStreamResponse extends AbstractResponse {

Expand Down
39 changes: 22 additions & 17 deletions .../requests/CommitCompactObjectRequest.java → .../requests/s3/GetStreamsOffsetRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,49 +15,54 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.CommitCompactObjectRequestData;
import org.apache.kafka.common.message.CommitCompactObjectResponseData;
import org.apache.kafka.common.message.GetStreamsOffsetRequestData;
import org.apache.kafka.common.message.CreateStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class CommitCompactObjectRequest extends AbstractRequest {
public static class Builder extends AbstractRequest.Builder<CommitCompactObjectRequest> {
public class GetStreamsOffsetRequest extends AbstractRequest {

private final CommitCompactObjectRequestData data;
public Builder(CommitCompactObjectRequestData data) {
super(ApiKeys.COMMIT_COMPACT_OBJECT);
public static class Builder extends AbstractRequest.Builder<GetStreamsOffsetRequest> {

private final GetStreamsOffsetRequestData data;
public Builder(GetStreamsOffsetRequestData data) {
super(ApiKeys.GET_STREAMS_OFFSET);
this.data = data;
}

@Override
public CommitCompactObjectRequest build(short version) {
return new CommitCompactObjectRequest(data, version);
public GetStreamsOffsetRequest build(short version) {
return new GetStreamsOffsetRequest(data, version);
}

@Override
public String toString() {
return data.toString();
}
}
private final CommitCompactObjectRequestData data;

public CommitCompactObjectRequest(CommitCompactObjectRequestData data, short version) {
super(ApiKeys.DELETE_STREAM, version);
private final GetStreamsOffsetRequestData data;

public GetStreamsOffsetRequest(GetStreamsOffsetRequestData data, short version) {
super(ApiKeys.GET_STREAMS_OFFSET, version);
this.data = data;
}

@Override
public CommitCompactObjectResponse getErrorResponse(int throttleTimeMs, Throwable e) {
public CreateStreamResponse getErrorResponse(int throttleTimeMs, Throwable e) {
ApiError apiError = ApiError.fromThrowable(e);
CommitCompactObjectResponseData response = new CommitCompactObjectResponseData()
CreateStreamResponseData response = new CreateStreamResponseData()
.setErrorCode(apiError.error().code())
.setThrottleTimeMs(throttleTimeMs);
return new CommitCompactObjectResponse(response);
return new CreateStreamResponse(response);
}

@Override
public CommitCompactObjectRequestData data() {
public GetStreamsOffsetRequestData data() {
return data;
}

}
17 changes: 8 additions & 9 deletions ...requests/CommitCompactObjectResponse.java → ...requests/s3/GetStreamsOffsetResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,24 +15,24 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.CommitCompactObjectResponseData;
import org.apache.kafka.common.message.GetStreamsOffsetResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class CommitCompactObjectResponse extends AbstractResponse {
public class GetStreamsOffsetResponse extends AbstractResponse {
private final GetStreamsOffsetResponseData data;

private final CommitCompactObjectResponseData data;

public CommitCompactObjectResponse(CommitCompactObjectResponseData data) {
super(ApiKeys.COMMIT_COMPACT_OBJECT);
public GetStreamsOffsetResponse(GetStreamsOffsetResponseData data) {
super(ApiKeys.GET_STREAMS_OFFSET);
this.data = data;
}

@Override
public CommitCompactObjectResponseData data() {
public GetStreamsOffsetResponseData data() {
return data;
}

Expand All @@ -50,5 +50,4 @@ public int throttleTimeMs() {
public void maybeSetThrottleTimeMs(int throttleTimeMs) {
data.setThrottleTimeMs(throttleTimeMs);
}

}
4 changes: 3 additions & 1 deletion ...ka/common/requests/OpenStreamRequest.java → ...common/requests/s3/OpenStreamRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.OpenStreamRequestData;
import org.apache.kafka.common.message.OpenStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class OpenStreamRequest extends AbstractRequest {

Expand Down
3 changes: 2 additions & 1 deletion ...a/common/requests/OpenStreamResponse.java → ...ommon/requests/s3/OpenStreamResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.OpenStreamResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class OpenStreamResponse extends AbstractResponse {

Expand Down
4 changes: 3 additions & 1 deletion ...mmon/requests/PrepareS3ObjectRequest.java → ...n/requests/s3/PrepareS3ObjectRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import org.apache.kafka.common.message.PrepareS3ObjectRequestData;
import org.apache.kafka.common.message.PrepareS3ObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;

public class PrepareS3ObjectRequest extends AbstractRequest {

Expand Down
3 changes: 2 additions & 1 deletion ...mon/requests/PrepareS3ObjectResponse.java → .../requests/s3/PrepareS3ObjectResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,13 @@
* limitations under the License.
*/

package org.apache.kafka.common.requests;
package org.apache.kafka.common.requests.s3;

import java.util.Map;
import org.apache.kafka.common.message.PrepareS3ObjectResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;

public class PrepareS3ObjectResponse extends AbstractResponse {
private final PrepareS3ObjectResponseData data;
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/CloseStreamRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
// limitations under the License.

{
"apiKey": 70,
"apiKey": 503,
"type": "request",
"listeners": [
"controller",
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/CloseStreamResponse.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
// limitations under the License.

{
"apiKey": 70,
"apiKey": 503,
"type": "response",
"name": "CloseStreamResponse",
"validVersions": "0",
Expand Down
Loading