[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r881292400 ## clients/src/main/java/org/apache/kafka/clients/admin/DescribeMetadataQuorumResult.java: ## @@ -0,0 +1,40 @@ +/* + * 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.clients.admin; + +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.QuorumInfo; + +/** + * The result of {@link Admin#describeMetadataQuorum(DescribeMetadataQuorumOptions)} + * + */ +public class DescribeMetadataQuorumResult { + +private final KafkaFuture quorumInfo; + +public DescribeMetadataQuorumResult(KafkaFuture quorumInfo) { Review Comment: nit: Could we keep it package private? ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4330,61 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponse response) { +Integer partition = 0; +String topicName = response.getTopicNameByIndex(partition); +Integer leaderId = response.getPartitionLeaderId(topicName, partition); +List voters = new ArrayList<>(); +for (Map.Entry entry: response.getVoterOffsets(topicName, partition).entrySet()) { Review Comment: nit: We could use `response.getVoterOffsets(topicName, partition).forEach((replicaId, logEndOffset) ->`. That makes the code a bit more readable in my opinion. ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4330,61 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponse response) { +Integer partition = 0; +String topicName = response.getTopicNameByIndex(partition); +Integer leaderId = response.getPartitionLeaderId(topicName, partition); +List voters = new ArrayList<>(); +for (Map.Entry entry: response.getVoterOffsets(topicName, partition).entrySet()) { +voters.add(new ReplicaState(entry.getKey(), entry.getValue(), OptionalLong.empty(), OptionalLong.empty())); Review Comment: hm.. `OptionalLong.empty()` seems wrong here. Don't you need to put the last fetch time and the last caughtup time? ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4330,61 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponse response) { +Intege
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r888913198 ## clients/src/main/java/org/apache/kafka/clients/admin/Admin.java: ## @@ -1446,6 +1446,35 @@ default DescribeFeaturesResult describeFeatures() { */ UpdateFeaturesResult updateFeatures(Map featureUpdates, UpdateFeaturesOptions options); +/** + * Describes the state of the metadata quorum. + * + * This is a convenience method for {@link #describeMetadataQuorum(DescribeMetadataQuorumOptions)} with default options. Review Comment: nit: There is an extra space before `with`. ## clients/src/main/java/org/apache/kafka/clients/admin/DescribeMetadataQuorumResult.java: ## @@ -0,0 +1,39 @@ +/* + * 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.clients.admin; + +import org.apache.kafka.common.KafkaFuture; + +/** + * The result of {@link Admin#describeMetadataQuorum(DescribeMetadataQuorumOptions)} + * + */ +public class DescribeMetadataQuorumResult { + +private final KafkaFuture quorumInfo; + +DescribeMetadataQuorumResult(KafkaFuture quorumInfo) { +this.quorumInfo = quorumInfo; +} + +/** + * Returns a future QuorumInfo Review Comment: nit: Should we say `Returns a future containing the quorum info.`? ## clients/src/main/java/org/apache/kafka/clients/admin/DescribeMetadataQuorumOptions.java: ## @@ -0,0 +1,25 @@ +/* + * 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.clients.admin; + +/** + * Options for {@link ConfluentAdmin#describeQuorum(DescribeMetadataQuorumOptions)}. + * Review Comment: nit: We could remove this empty line. ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4328,91 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeMetadataQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponse response) { +Integer partition = 0; +String topicName = response.getTopicNameByIndex(0); +Integer leaderId = response.getPartitionLeaderId(topicName, partition); +List voters = new ArrayList<>(); +List observers = new ArrayList<>(); +response.getVoterInfo(topicName, partition).forEach(v -> { +voters.add(new QuorumInfo.ReplicaState(v.replicaId(), +v.logEndOffset(), +OptionalLong.of(v.lastFetchTimestamp()), +OptionalLong.of(v.lastCaughtUpTimestamp(; +}); +response.getObserverInfo(topicName, partition).forEach(o -> { +observers.add(new QuorumInfo.ReplicaState(o.replicaId(), +o.logEndOffset(), +OptionalLong.of(o.lastFetchTimestamp()), +
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r890984307 ## clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java: ## @@ -4833,6 +4879,92 @@ public void testDescribeFeaturesFailure() { } } +@Test +public void testDescribeMetadataQuorumSuccess() throws Exception { +try (final AdminClientUnitTestEnv env = mockClientEnv()) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create(ApiKeys.DESCRIBE_QUORUM.id, Review Comment: Ack. I was not aware of this bug. -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r890986175 ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4337,25 +4337,22 @@ public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuoru final Call call = new Call( "describeMetadataQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { -private QuorumInfo createQuorumResult(final DescribeQuorumResponse response) { -Integer partition = 0; -String topicName = response.getTopicNameByIndex(0); -Integer leaderId = response.getPartitionLeaderId(topicName, partition); +private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.PartitionData partition) { List voters = new ArrayList<>(); List observers = new ArrayList<>(); -response.getVoterInfo(topicName, partition).forEach(v -> { +partition.currentVoters().forEach(v -> { voters.add(new QuorumInfo.ReplicaState(v.replicaId(), v.logEndOffset(), OptionalLong.of(v.lastFetchTimestamp()), OptionalLong.of(v.lastCaughtUpTimestamp(; Review Comment: When `lastFetchTimestamp` or `lastCaughtUpTimestamp` are not provided (equals to -1), don't we want to return an empty option instead of returning an option containing -1? -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r892495891 ## core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala: ## @@ -778,4 +778,35 @@ class KRaftClusterTest { cluster.close() } } + + @Test + def testDescribeQuorumRequestToBrokers() : Unit = { +val cluster = new KafkaClusterTestKit.Builder( + new TestKitNodes.Builder(). +setNumBrokerNodes(4). +setNumControllerNodes(3).build()).build() +try { + cluster.format + cluster.startup + for (i <- 0 to 3) { +TestUtils.waitUntilTrue(() => cluster.brokers.get(i).brokerState == BrokerState.RUNNING, + "Broker Never started up") + } + val props = cluster.clientProperties() + props.put(AdminClientConfig.CLIENT_ID_CONFIG, this.getClass.getName) + val admin = Admin.create(props) + try { +val quorumState = admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions) +val quorumInfo = quorumState.quorumInfo().get() + +assertEquals(0, quorumInfo.leaderId()) Review Comment: Should we check all the data in quorum info in this integration test? Just checking the leader and the number of voters/observers seems weak to me. ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4328,88 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeMetadataQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.PartitionData partition) { +List voters = new ArrayList<>(); +List observers = new ArrayList<>(); +partition.currentVoters().forEach(v -> { +voters.add(new QuorumInfo.ReplicaState(v.replicaId(), +v.logEndOffset(), +v.lastFetchTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(v.lastFetchTimestamp()), +v.lastCaughtUpTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(v.lastCaughtUpTimestamp(; +}); +partition.observers().forEach(o -> { +observers.add(new QuorumInfo.ReplicaState(o.replicaId(), +o.logEndOffset(), +o.lastFetchTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(o.lastFetchTimestamp()), +o.lastCaughtUpTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(o.lastCaughtUpTimestamp(; Review Comment: Could we extract this bloc into an helper method and share it with the voters part? ## core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala: ## @@ -54,30 +54,46 @@ class DescribeQuorumRequestTest(cluster: ClusterInstance) { @ClusterTest def testDescribeQuorum(): Unit = { -val request = new DescribeQuorumRequest.Builder( - singletonRequest(KafkaRaftServer.MetadataPartition) -).build() - -val response = connectAndReceive[DescribeQuorumResponse](request) - -assertEquals(Errors.NONE, Errors.forCode(response.data.errorCode)) -assertEquals(1, response.data.topics.size) - -val topicData = response.data.topics.get(0) -assertEquals(KafkaRaftServer.MetadataTopic, topicData.topicName) -assertEquals(1, topicData.partitions.size) - -val partitionData = topicData.partitions.get(0) -assertEquals(KafkaRaftServer.MetadataPartition.partition, partitionData.partitionIndex) -assertEquals(Errors.NONE, Errors.forCode(partitionData.errorCode)) -assertTrue(partitionData.leaderEpoch > 0) - -val leaderId = partitionData.leaderId -assertTrue(leaderId > 0) - -val leaderState = partitionData.currentVoters.asScala.find(_.replicaId == leaderId) - .getOrElse(throw new AssertionError("Failed to find leader among current voter states")) -assertTrue(leaderState.logEndOffset > 0) +for (version <- ApiKeys.DESCRIBE_QUORUM.allVersions.asScala) { + val request = new DescribeQuorumRequest.Builder( +singletonRequest(KafkaRaftServer.MetadataPartition) + ).build(version.toShort) + val response = connectAndReceive[DescribeQuorumResponse](request) + + assertEquals(Errors.NONE, Errors.forCode(response.data.errorCode)) + assertEquals(1, response.data.topics.size) + + val topicData = response.data.topics.get(0) + assertEquals(KafkaRaftServer.Met
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r893659548 ## clients/src/main/java/org/apache/kafka/common/requests/DescribeQuorumResponse.java: ## @@ -93,4 +93,5 @@ public static DescribeQuorumResponseData singletonResponse(TopicPartition topicP public static DescribeQuorumResponse parse(ByteBuffer buffer, short version) { return new DescribeQuorumResponse(new DescribeQuorumResponseData(new ByteBufferAccessor(buffer), version)); } + Review Comment: nit: This empty line could be removed. ## clients/src/main/resources/common/message/DescribeQuorumResponse.json: ## @@ -17,7 +17,8 @@ "apiKey": 55, "type": "response", "name": "DescribeQuorumResponse", - "validVersions": "0", + // Version 1 adds LastFetchTimeStamp and LastCaughtUpTimestamp in ReplicaState (KIP-836) Review Comment: nit: Can we add a dot at the end of the sentence? ## clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java: ## @@ -0,0 +1,149 @@ +/* + * 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.clients.admin; + +import java.util.List; +import java.util.Objects; +import java.util.OptionalLong; + +/** + * This is used to describe per-partition state in the DescribeQuorumResponse. Review Comment: This comment may have been missed. ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -4321,6 +4328,88 @@ void handleFailure(Throwable throwable) { return new UpdateFeaturesResult(new HashMap<>(updateFutures)); } +@Override +public DescribeMetadataQuorumResult describeMetadataQuorum(DescribeMetadataQuorumOptions options) { +NodeProvider provider = new LeastLoadedNodeProvider(); + +final KafkaFutureImpl future = new KafkaFutureImpl<>(); +final long now = time.milliseconds(); +final Call call = new Call( +"describeMetadataQuorum", calcDeadlineMs(now, options.timeoutMs()), provider) { + +private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.PartitionData partition) { +List voters = new ArrayList<>(); +List observers = new ArrayList<>(); +partition.currentVoters().forEach(v -> { +voters.add(new QuorumInfo.ReplicaState(v.replicaId(), +v.logEndOffset(), +v.lastFetchTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(v.lastFetchTimestamp()), +v.lastCaughtUpTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(v.lastCaughtUpTimestamp(; +}); +partition.observers().forEach(o -> { +observers.add(new QuorumInfo.ReplicaState(o.replicaId(), +o.logEndOffset(), +o.lastFetchTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(o.lastFetchTimestamp()), +o.lastCaughtUpTimestamp() == -1 ? OptionalLong.empty() : OptionalLong.of(o.lastCaughtUpTimestamp(; +}); +QuorumInfo info = new QuorumInfo(partition.leaderId(), voters, observers); +return info; +} + +@Override +DescribeQuorumRequest.Builder createRequest(int timeoutMs) { +return new Builder(DescribeQuorumRequest.singletonRequest( +new TopicPartition(METADATA_TOPIC_NAME, METADATA_TOPIC_PARTITION.partition(; +} + +@Override +void handleResponse(AbstractResponse response) { +final DescribeQuorumResponse quorumResponse = (DescribeQuorumResponse) response; +if (quorumResponse.data().errorCode() != Errors.NONE.code()) { +throw Errors.forCode(quorumResponse.data().errorCode()).exception(); Review Comment: Great, thanks. ## clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java: ## @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * c
[GitHub] [kafka] dajac commented on a diff in pull request #12206: KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag
dajac commented on code in PR #12206: URL: https://github.com/apache/kafka/pull/12206#discussion_r895790891 ## clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java: ## @@ -0,0 +1,153 @@ +/* + * 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.clients.admin; + +import java.util.List; +import java.util.Objects; +import java.util.OptionalLong; + +/** + * This class is used to describe the state of the quorum received in DescribeQuorumResponse. + */ +public class QuorumInfo { +private final Integer leaderId; +private final List voters; +private final List observers; + +QuorumInfo(Integer leaderId, List voters, List observers) { +this.leaderId = leaderId; +this.voters = voters; +this.observers = observers; +} + +public Integer leaderId() { +return leaderId; +} + +public List voters() { +return voters; +} + +public List observers() { +return observers; +} + +@Override +public boolean equals(Object o) { +if (this == o) return true; +if (o == null || getClass() != o.getClass()) return false; +QuorumInfo that = (QuorumInfo) o; +return leaderId.equals(that.leaderId) +&& voters.equals(that.voters) +&& observers.equals(that.observers); +} + +@Override +public int hashCode() { +return Objects.hash(leaderId, voters, observers); +} + +@Override +public String toString() { +return "QuorumInfo(" + +"leaderId=" + leaderId + +", voters=" + voters.toString() + +", observers=" + observers.toString() + +')'; +} + +public static class ReplicaState { +private final int replicaId; +private final long logEndOffset; +private final OptionalLong lastFetchTimeMs; +private final OptionalLong lastCaughtUpTimeMs; + +ReplicaState() { +this(0, 0, OptionalLong.empty(), OptionalLong.empty()); +} + +ReplicaState( +int replicaId, +long logEndOffset, +OptionalLong lastFetchTimeMs, +OptionalLong lastCaughtUpTimeMs Review Comment: nit: We usually use 4 spaces indentation here. ## core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala: ## @@ -778,4 +778,44 @@ class KRaftClusterTest { cluster.close() } } + + @Test + def testDescribeQuorumRequestToBrokers() : Unit = { +val cluster = new KafkaClusterTestKit.Builder( + new TestKitNodes.Builder(). +setNumBrokerNodes(4). +setNumControllerNodes(3).build()).build() +try { + cluster.format + cluster.startup + for (i <- 0 to 3) { +TestUtils.waitUntilTrue(() => cluster.brokers.get(i).brokerState == BrokerState.RUNNING, + "Broker Never started up") + } + val props = cluster.clientProperties() + props.put(AdminClientConfig.CLIENT_ID_CONFIG, this.getClass.getName) + val admin = Admin.create(props) + try { +val quorumState = admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions) +val quorumInfo = quorumState.quorumInfo.get() + +assertEquals(0, quorumInfo.observers.size()) +assertEquals(3, quorumInfo.voters.size()) Review Comment: nit: We could omit parenthesis after `size`. The same for the previous line. ## core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala: ## @@ -54,30 +54,48 @@ class DescribeQuorumRequestTest(cluster: ClusterInstance) { @ClusterTest def testDescribeQuorum(): Unit = { -val request = new DescribeQuorumRequest.Builder( - singletonRequest(KafkaRaftServer.MetadataPartition) -).build() - -val response = connectAndReceive[DescribeQuorumResponse](request) - -assertEquals(Errors.NONE, Errors.forCode(response.data.errorCode)) -assertEquals(1, response.data.topics.size) - -val topicData = response.data.topics.get(0) -assertEquals(KafkaRaftServer.MetadataTopic, topicData.topicName) -assertEquals(1, topicData.partitions.size) - -val partitionData = topicD