You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/06/18 21:01:55 UTC
[GitHub] [kafka] hachikuji opened a new pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
hachikuji opened a new pull request #10900:
URL: https://github.com/apache/kafka/pull/10900
We added the DescribeQuorum API in KIP-595. This patch adds the logic to forward DescribeQuorum requests to the controller when KRaft is enabled.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r655556345
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Not needed for correctness of course, but KRaft seems to be the preferred term these days for the Kafka+Raft clusters.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r657576935
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -217,6 +227,7 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.DESCRIBE_TRANSACTIONS => handleDescribeTransactionsRequest(request)
case ApiKeys.LIST_TRANSACTIONS => handleListTransactionsRequest(request)
case ApiKeys.ALLOCATE_PRODUCER_IDS => maybeForwardToController(request, handleAllocateProducerIdsRequest)
+ case ApiKeys.DESCRIBE_QUORUM => forwardToControllerOrFail(request)
Review comment:
Yeah, because DescribeQuorum does not make sense for zk clusters. As far as I know, this is the only case of a client api which is only exposed under kraft. However, we do have some internal controller apis that are only used by kraft (e.g. for broker registration and heartbeating).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r655556345
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Not needed for correctness of course, but KRaft seems to be the preferred term these days for the Kafka+Raft clusters.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
+
+ private def testForwardableApi(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ testForwardableApi(
+ createKafkaApis(enableForwarding = true),
+ apiKey,
+ requestBuilder
+ )
}
- private def testForwardableAPI(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ private def testForwardableApi(
+ kafkaApis: KafkaApis,
+ apiKey: ApiKeys,
+ requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest],
+ ): Unit = {
val topicHeader = new RequestHeader(apiKey, apiKey.latestVersion,
clientId, 0)
val request = buildRequest(requestBuilder.build(topicHeader.apiVersion))
- EasyMock.expect(controller.isActive).andReturn(false)
+ EasyMock.expect(controller.isActive).andStubReturn(false)
Review comment:
I was being a little lazy. For `DescribeQuorum` in KRaft clusters, we forward it to the controller blindly because the broker cannot be a controller. Let me see if I can make this conditional.
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Not needed for correctness of course, but KRaft seems to be the preferred term these days for the Kafka+Raft clusters. I thought I would go ahead and change here since there are only a few uses so far.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r654069402
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
Review comment:
Fair question. Let me see if I can add a request test.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
Review comment:
Added a basic integration test case. Good thing too because it uncovered the need to set the `forwardable` flag in `ApiKeys`.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r657234456
##########
File path: core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala
##########
@@ -0,0 +1,67 @@
+/*
+ * 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 kafka.server
+
+import kafka.server.IntegrationTestUtils.connectAndReceive
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.DescribeQuorumRequest.singletonRequest
+import org.apache.kafka.common.requests.{DescribeQuorumRequest, DescribeQuorumResponse}
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.extension.ExtendWith
+
+import scala.jdk.CollectionConverters._
+
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.KRAFT)
+class DescribeQuorumRequestTest(cluster: ClusterInstance) {
Review comment:
Is it worth adding a test for ZK mode here? Just to see that it fails
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -217,6 +227,7 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.DESCRIBE_TRANSACTIONS => handleDescribeTransactionsRequest(request)
case ApiKeys.LIST_TRANSACTIONS => handleListTransactionsRequest(request)
case ApiKeys.ALLOCATE_PRODUCER_IDS => maybeForwardToController(request, handleAllocateProducerIdsRequest)
+ case ApiKeys.DESCRIBE_QUORUM => forwardToControllerOrFail(request)
Review comment:
Hm, so we are _only_ allowing forwarding with this `forwardToControllerOrFail` since this RPC did not exist before? Are there other recently added controller-only RPCs that need this treatment?
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Seems fine to change this here. I've been meaning to do this anyways.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r655019794
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Is this change needed?
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
+
+ private def testForwardableApi(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ testForwardableApi(
+ createKafkaApis(enableForwarding = true),
+ apiKey,
+ requestBuilder
+ )
}
- private def testForwardableAPI(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ private def testForwardableApi(
+ kafkaApis: KafkaApis,
+ apiKey: ApiKeys,
+ requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest],
+ ): Unit = {
val topicHeader = new RequestHeader(apiKey, apiKey.latestVersion,
clientId, 0)
val request = buildRequest(requestBuilder.build(topicHeader.apiVersion))
- EasyMock.expect(controller.isActive).andReturn(false)
+ EasyMock.expect(controller.isActive).andStubReturn(false)
Review comment:
What's the purpose of this switch?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r654055214
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
Review comment:
Sounds good. How do we check that the `ControlerApi` is doing the right thing :tm: ? Should we add a test to `RaftClusterTest` or do we need to add this to the `AdminClient` first to do that?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #10900:
URL: https://github.com/apache/kafka/pull/10900
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r655591961
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -462,16 +461,55 @@ class KafkaApisTest {
@Test
def testAlterConfigsWithForwarding(): Unit = {
val requestBuilder = new AlterConfigsRequest.Builder(Collections.emptyMap(), false)
- testForwardableAPI(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ testForwardableApi(ApiKeys.ALTER_CONFIGS, requestBuilder)
+ }
+
+ @Test
+ def testDescribeQuorumNotAllowedForZkClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ val request = buildRequest(requestBuilder.build())
+
+ val capturedResponse = expectNoThrottling(request)
+ EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, adminManager, controller)
+ createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+
+ val response = capturedResponse.getValue.asInstanceOf[DescribeQuorumResponse]
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
+ }
+
+ @Test
+ def testDescribeQuorumForwardedForKRaftClusters(): Unit = {
+ val requestData = DescribeQuorumRequest.singletonRequest(KafkaRaftServer.MetadataPartition)
+ val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
+ metadataCache = MetadataCache.raftMetadataCache(brokerId)
+
+ testForwardableApi(
+ createKafkaApis(raftSupport = true),
+ ApiKeys.DESCRIBE_QUORUM,
+ requestBuilder
+ )
+ }
+
+ private def testForwardableApi(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ testForwardableApi(
+ createKafkaApis(enableForwarding = true),
+ apiKey,
+ requestBuilder
+ )
}
- private def testForwardableAPI(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
+ private def testForwardableApi(
+ kafkaApis: KafkaApis,
+ apiKey: ApiKeys,
+ requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest],
+ ): Unit = {
val topicHeader = new RequestHeader(apiKey, apiKey.latestVersion,
clientId, 0)
val request = buildRequest(requestBuilder.build(topicHeader.apiVersion))
- EasyMock.expect(controller.isActive).andReturn(false)
+ EasyMock.expect(controller.isActive).andStubReturn(false)
Review comment:
I was being a little lazy. For `DescribeQuorum` in KRaft clusters, we forward it to the controller blindly because the broker cannot be a controller. Let me see if I can make this conditional.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r657803664
##########
File path: core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala
##########
@@ -0,0 +1,67 @@
+/*
+ * 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 kafka.server
+
+import kafka.server.IntegrationTestUtils.connectAndReceive
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.DescribeQuorumRequest.singletonRequest
+import org.apache.kafka.common.requests.{DescribeQuorumRequest, DescribeQuorumResponse}
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.extension.ExtendWith
+
+import scala.jdk.CollectionConverters._
+
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.KRAFT)
+class DescribeQuorumRequestTest(cluster: ClusterInstance) {
Review comment:
Yeah, good idea. Added in the latest commit.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10900: KAFKA-12967; KRaft broker should forward DescribeQuorum to controller
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10900:
URL: https://github.com/apache/kafka/pull/10900#discussion_r655556345
##########
File path: core/src/test/java/kafka/test/annotation/Type.java
##########
@@ -28,7 +28,7 @@
* The type of cluster config being requested. Used by {@link kafka.test.ClusterConfig} and the test annotations.
*/
public enum Type {
- RAFT {
+ KRAFT {
Review comment:
Not needed for correctness of course, but KRaft seems to be the preferred term these days for the Kafka+Raft clusters. I thought I would go ahead and change here since there are only a few uses so far.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org