forked from apache/kafka
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
KAFKA-12204; Implement DescribeCluster API in the broker (KIP-700) (a…
…pache#9903) This PR implements the DescribeCluster API in the broker. Reviewers: Rajini Sivaram <[email protected]>, Ismael Juma <[email protected]>
- Loading branch information
Showing
15 changed files
with
498 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
clients/src/main/java/org/apache/kafka/common/requests/DescribeClusterRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * 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.requests; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import org.apache.kafka.common.message.DescribeClusterRequestData; | ||
| import org.apache.kafka.common.message.DescribeClusterResponseData; | ||
| import org.apache.kafka.common.protocol.ApiKeys; | ||
| import org.apache.kafka.common.protocol.ByteBufferAccessor; | ||
|
|
||
| public class DescribeClusterRequest extends AbstractRequest { | ||
|
|
||
| public static class Builder extends AbstractRequest.Builder<DescribeClusterRequest> { | ||
|
|
||
| private final DescribeClusterRequestData data; | ||
|
|
||
| public Builder(DescribeClusterRequestData data) { | ||
| super(ApiKeys.DESCRIBE_CLUSTER); | ||
| this.data = data; | ||
| } | ||
|
|
||
| @Override | ||
| public DescribeClusterRequest build(final short version) { | ||
| return new DescribeClusterRequest(data, version); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return data.toString(); | ||
| } | ||
| } | ||
|
|
||
| private final DescribeClusterRequestData data; | ||
|
|
||
| public DescribeClusterRequest(DescribeClusterRequestData data, short version) { | ||
| super(ApiKeys.DESCRIBE_CLUSTER, version); | ||
| this.data = data; | ||
| } | ||
|
|
||
| @Override | ||
| public DescribeClusterRequestData data() { | ||
| return data; | ||
| } | ||
|
|
||
| @Override | ||
| public AbstractResponse getErrorResponse(final int throttleTimeMs, final Throwable e) { | ||
| ApiError apiError = ApiError.fromThrowable(e); | ||
| return new DescribeClusterResponse(new DescribeClusterResponseData() | ||
| .setErrorCode(apiError.error().code()) | ||
| .setErrorMessage(apiError.message())); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString(final boolean verbose) { | ||
| return data.toString(); | ||
| } | ||
|
|
||
| public static DescribeClusterRequest parse(ByteBuffer buffer, short version) { | ||
| return new DescribeClusterRequest(new DescribeClusterRequestData(new ByteBufferAccessor(buffer), version), version); | ||
| } | ||
| } |
63 changes: 63 additions & 0 deletions
63
clients/src/main/java/org/apache/kafka/common/requests/DescribeClusterResponse.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * 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.requests; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.Map; | ||
| import java.util.function.Function; | ||
| import java.util.stream.Collectors; | ||
| import org.apache.kafka.common.Node; | ||
| import org.apache.kafka.common.message.DescribeClusterResponseData; | ||
| import org.apache.kafka.common.protocol.ApiKeys; | ||
| import org.apache.kafka.common.protocol.ByteBufferAccessor; | ||
| import org.apache.kafka.common.protocol.Errors; | ||
|
|
||
| public class DescribeClusterResponse extends AbstractResponse { | ||
|
|
||
| private final DescribeClusterResponseData data; | ||
|
|
||
| public DescribeClusterResponse(DescribeClusterResponseData data) { | ||
| super(ApiKeys.DESCRIBE_CLUSTER); | ||
| this.data = data; | ||
| } | ||
|
|
||
| public Map<Integer, Node> nodes() { | ||
| return data.brokers().valuesList().stream() | ||
| .map(b -> new Node(b.brokerId(), b.host(), b.port(), b.rack())) | ||
| .collect(Collectors.toMap(Node::id, Function.identity())); | ||
| } | ||
|
|
||
| @Override | ||
| public Map<Errors, Integer> errorCounts() { | ||
| return errorCounts(Errors.forCode(data.errorCode())); | ||
| } | ||
|
|
||
| @Override | ||
| public int throttleTimeMs() { | ||
| return data.throttleTimeMs(); | ||
| } | ||
|
|
||
| @Override | ||
| public DescribeClusterResponseData data() { | ||
| return data; | ||
| } | ||
|
|
||
| public static DescribeClusterResponse parse(ByteBuffer buffer, short version) { | ||
| return new DescribeClusterResponse(new DescribeClusterResponseData(new ByteBufferAccessor(buffer), version)); | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
clients/src/main/resources/common/message/DescribeClusterRequest.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| // 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. | ||
|
|
||
| { | ||
| "apiKey": 60, | ||
| "type": "request", | ||
| "name": "DescribeClusterRequest", | ||
| "validVersions": "0", | ||
| "flexibleVersions": "0+", | ||
| "fields": [ | ||
| { "name": "IncludeClusterAuthorizedOperations", "type": "bool", "versions": "0+", | ||
| "about": "Whether to include cluster authorized operations." } | ||
| ] | ||
| } |
47 changes: 47 additions & 0 deletions
47
clients/src/main/resources/common/message/DescribeClusterResponse.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| // 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. | ||
|
|
||
| { | ||
| "apiKey": 60, | ||
| "type": "response", | ||
| "name": "DescribeClusterResponse", | ||
| "validVersions": "0", | ||
| "flexibleVersions": "0+", | ||
| "fields": [ | ||
| { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", | ||
| "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }, | ||
| { "name": "ErrorCode", "type": "int16", "versions": "0+", | ||
| "about": "The top-level error code, or 0 if there was no error" }, | ||
| { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", | ||
| "about": "The top-level error message, or null if there was no error." }, | ||
| { "name": "ClusterId", "type": "string", "versions": "0+", | ||
| "about": "The cluster ID that responding broker belongs to." }, | ||
| { "name": "ControllerId", "type": "int32", "versions": "0+", "default": "-1", "entityType": "brokerId", | ||
| "about": "The ID of the controller broker." }, | ||
| { "name": "Brokers", "type": "[]DescribeClusterBroker", "versions": "0+", | ||
| "about": "Each broker in the response.", "fields": [ | ||
| { "name": "BrokerId", "type": "int32", "versions": "0+", "mapKey": true, "entityType": "brokerId", | ||
| "about": "The broker ID." }, | ||
| { "name": "Host", "type": "string", "versions": "0+", | ||
| "about": "The broker hostname." }, | ||
| { "name": "Port", "type": "int32", "versions": "0+", | ||
| "about": "The broker port." }, | ||
| { "name": "Rack", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", | ||
| "about": "The rack of the broker, or null if it has not been assigned to a rack." } | ||
| ]}, | ||
| { "name": "ClusterAuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648", | ||
| "about": "32-bit bitfield to represent authorized operations for this cluster." } | ||
| ] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.