You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by di...@apache.org on 2024/01/02 14:58:26 UTC
(kafka) branch trunk updated: KAFKA-16059: Fix thread leak KafkaAPIsTest (#15093)
This is an automated email from the ASF dual-hosted git repository.
divijv pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 65b15585326 KAFKA-16059: Fix thread leak KafkaAPIsTest (#15093)
65b15585326 is described below
commit 65b155853269a8449f2c8a7779ed8ae5e4f26b4c
Author: Divij Vaidya <di...@amazon.com>
AuthorDate: Tue Jan 2 15:58:20 2024 +0100
KAFKA-16059: Fix thread leak KafkaAPIsTest (#15093)
Reviewers: Luke Chen <sh...@gmail.com>
---
.../scala/unit/kafka/server/KafkaApisTest.scala | 542 +++++++++++----------
1 file changed, 278 insertions(+), 264 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index d46508b93da..5a1a0cff7ed 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -31,7 +31,7 @@ import kafka.log.UnifiedLog
import kafka.network.{RequestChannel, RequestMetrics}
import kafka.server.QuotaFactory.QuotaManagers
import kafka.server.metadata.{ConfigRepository, KRaftMetadataCache, MockConfigRepository, ZkMetadataCache}
-import kafka.utils.{Log4jController, TestUtils}
+import kafka.utils.{CoreUtils, Log4jController, Logging, TestUtils}
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.admin.AlterConfigOp.OpType
import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry}
@@ -106,7 +106,7 @@ import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchParams, FetchP
import java.time.Duration
-class KafkaApisTest {
+class KafkaApisTest extends Logging {
private val requestChannel: RequestChannel = mock(classOf[RequestChannel])
private val requestChannelMetrics: RequestChannel.Metrics = mock(classOf[RequestChannel.Metrics])
private val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
@@ -139,10 +139,13 @@ class KafkaApisTest {
private val clusterId = "clusterId"
private val time = new MockTime
private val clientId = ""
+ private var kafkaApis: KafkaApis = _
@AfterEach
def tearDown(): Unit = {
- quotas.shutdown()
+ CoreUtils.swallow(quotas.shutdown(), this)
+ if (kafkaApis != null)
+ CoreUtils.swallow(kafkaApis.close(), this)
TestUtils.clearYammerMetrics()
metrics.close()
}
@@ -266,9 +269,8 @@ class KafkaApisTest {
requestHeader = Option(requestHeader))
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer), configRepository = configRepository)
- .handleDescribeConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer), configRepository = configRepository)
+ kafkaApis.handleDescribeConfigsRequest(request)
verify(authorizer).authorize(any(), ArgumentMatchers.eq(expectedActions.asJava))
val response = verifyNoThrottling[DescribeConfigsResponse](request)
@@ -335,8 +337,8 @@ class KafkaApisTest {
val capturedResponse: ArgumentCaptor[AlterConfigsResponse] = ArgumentCaptor.forClass(classOf[AlterConfigsResponse])
val capturedRequest: ArgumentCaptor[RequestChannel.Request] = ArgumentCaptor.forClass(classOf[RequestChannel.Request])
-
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer), enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching)
verify(requestChannel).sendResponse(
capturedRequest.capture(),
@@ -370,8 +372,8 @@ class KafkaApisTest {
leaveGroupRequest, kafkaPrincipalSerde, requestChannelMetrics, time.nanoseconds())
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[EnvelopeResponse](request)
assertEquals(Errors.INVALID_REQUEST, response.error())
@@ -424,7 +426,8 @@ class KafkaApisTest {
alterConfigsRequest, kafkaPrincipalSerde, requestChannelMetrics, time.nanoseconds(), fromPrivilegedListener = fromPrivilegedListener)
val capturedResponse: ArgumentCaptor[AbstractResponse] = ArgumentCaptor.forClass(classOf[AbstractResponse])
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer), enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching)
if (shouldCloseConnection) {
verify(requestChannel).closeConnection(
@@ -472,8 +475,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(adminManager.alterConfigs(any(), ArgumentMatchers.eq(false)))
.thenReturn(Map(authorizedResource -> ApiError.NONE))
-
- createKafkaApis(authorizer = Some(authorizer)).handleAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleAlterConfigsRequest(request)
val response = verifyNoThrottling[AlterConfigsResponse](request)
verifyAlterConfigResult(response, Map(authorizedTopic -> Errors.NONE,
@@ -513,8 +516,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(adminManager.alterConfigs(any(), ArgumentMatchers.eq(false)))
.thenReturn(Map(authorizedResource -> ApiError.NONE))
-
- createKafkaApis(authorizer = Some(authorizer)).handleAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleAlterConfigsRequest(request)
val response = verifyNoThrottling[AlterConfigsResponse](request)
verifyAlterConfigResult(response, Map(subscriptionName -> Errors.NONE))
verify(authorizer, times(1)).authorize(any(), any())
@@ -544,8 +547,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(adminManager.incrementalAlterConfigs(any(), ArgumentMatchers.eq(false)))
.thenReturn(Map(resource -> ApiError.NONE))
-
- createKafkaApis(authorizer = Some(authorizer)).handleIncrementalAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleIncrementalAlterConfigsRequest(request)
val response = verifyNoThrottling[IncrementalAlterConfigsResponse](request)
verifyIncrementalAlterConfigResult(response, Map(subscriptionName -> Errors.NONE ))
verify(authorizer, times(1)).authorize(any(), any())
@@ -594,9 +597,8 @@ class KafkaApisTest {
requestHeader = Option(requestHeader))
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer), configRepository = configRepository)
- .handleDescribeConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer), configRepository = configRepository)
+ kafkaApis.handleDescribeConfigsRequest(request)
val response = verifyNoThrottling[DescribeConfigsResponse](request)
// Verify that authorize is only called once
@@ -619,7 +621,8 @@ class KafkaApisTest {
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis(enableForwarding = true).handle(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[DescribeQuorumResponse](request)
assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
@@ -631,8 +634,7 @@ class KafkaApisTest {
val requestBuilder = new DescribeQuorumRequest.Builder(requestData)
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- testForwardableApi(
- createKafkaApis(raftSupport = true),
+ testForwardableApi(kafkaApis = createKafkaApis(raftSupport = true),
ApiKeys.DESCRIBE_QUORUM,
requestBuilder
)
@@ -643,16 +645,14 @@ class KafkaApisTest {
requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]
): Unit = {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- testForwardableApi(
- createKafkaApis(enableForwarding = true, raftSupport = true),
+ testForwardableApi(kafkaApis = createKafkaApis(enableForwarding = true, raftSupport = true),
apiKey,
requestBuilder
)
}
private def testForwardableApi(apiKey: ApiKeys, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Unit = {
- testForwardableApi(
- createKafkaApis(enableForwarding = true),
+ testForwardableApi(kafkaApis = createKafkaApis(enableForwarding = true),
apiKey,
requestBuilder
)
@@ -760,8 +760,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(adminManager.incrementalAlterConfigs(any(), ArgumentMatchers.eq(false)))
.thenReturn(Map(authorizedResource -> ApiError.NONE))
-
- createKafkaApis(authorizer = Some(authorizer)).handleIncrementalAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleIncrementalAlterConfigsRequest(request)
val capturedResponse = verifyNoThrottling[IncrementalAlterConfigsResponse](request)
verifyIncrementalAlterConfigResult(capturedResponse, Map(
@@ -811,8 +811,8 @@ class KafkaApisTest {
when(controller.isActive).thenReturn(true)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
anyLong)).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer)).handleAlterClientQuotasRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleAlterClientQuotasRequest(request)
val capturedResponse = verifyNoThrottling[AlterClientQuotasResponse](request)
verifyAlterClientQuotaResult(capturedResponse, Map(quotaEntity -> Errors.CLUSTER_AUTHORIZATION_FAILED))
@@ -882,8 +882,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(clientControllerQuotaManager.newQuotaFor(
ArgumentMatchers.eq(request), ArgumentMatchers.eq(6))).thenReturn(UnboundedControllerMutationQuota)
-
- createKafkaApis(authorizer = Some(authorizer)).handleCreateTopicsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleCreateTopicsRequest(request)
val capturedCallback: ArgumentCaptor[Map[String, ApiError] => Unit] = ArgumentCaptor.forClass(classOf[Map[String, ApiError] => Unit])
@@ -930,7 +930,7 @@ class KafkaApisTest {
val requestBuilder = new CreateTopicsRequest.Builder(requestData).build()
val request = buildRequest(requestBuilder)
- val kafkaApis = createKafkaApis(enableForwarding = true, raftSupport = true)
+ kafkaApis = createKafkaApis(enableForwarding = true, raftSupport = true)
val forwardCallback: ArgumentCaptor[Option[AbstractResponse] => Unit] =
ArgumentCaptor.forClass(classOf[Option[AbstractResponse] => Unit])
@@ -965,7 +965,7 @@ class KafkaApisTest {
@Test
def testCreatePartitionsAuthorization(): Unit = {
val authorizer: Authorizer = mock(classOf[Authorizer])
- val kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
val timeoutMs = 35000
val requestData = new CreatePartitionsRequestData()
@@ -1221,9 +1221,9 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
val capturedRequest = verifyTopicCreation(topicName, true, true, request)
-
- createKafkaApis(authorizer = Some(authorizer),
- overrideProperties = topicConfigOverride).handleFindCoordinatorRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer),
+ overrideProperties = topicConfigOverride)
+ kafkaApis.handleFindCoordinatorRequest(request)
val response = verifyNoThrottling[FindCoordinatorResponse](request)
if (version >= 4) {
@@ -1324,9 +1324,9 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
val capturedRequest = verifyTopicCreation(topicName, enableAutoTopicCreation, isInternal, request)
-
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = enableAutoTopicCreation,
- overrideProperties = topicConfigOverride).handleTopicMetadataRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer), enableForwarding = enableAutoTopicCreation,
+ overrideProperties = topicConfigOverride)
+ kafkaApis.handleTopicMetadataRequest(request)
val response = verifyNoThrottling[MetadataResponse](request)
val expectedMetadataResponse = util.Collections.singletonList(new TopicMetadata(
@@ -1388,7 +1388,7 @@ class KafkaApisTest {
topics.foreach(topic => {
val metadataRequestData = new MetadataRequestData().setTopics(Collections.singletonList(topic))
val request = buildRequest(new MetadataRequest(metadataRequestData, version.toShort))
- val kafkaApis = createKafkaApis()
+ kafkaApis = createKafkaApis()
val capturedResponse: ArgumentCaptor[AbstractResponse] = ArgumentCaptor.forClass(classOf[AbstractResponse])
kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching)
@@ -1429,8 +1429,8 @@ class KafkaApisTest {
offsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1473,8 +1473,9 @@ class KafkaApisTest {
offsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1569,8 +1570,8 @@ class KafkaApisTest {
expectedOffsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1665,7 +1666,8 @@ class KafkaApisTest {
val request = buildRequest(offsetCommitRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis().handleOffsetCommitRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetCommitRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[OffsetCommitResponse](request)
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION,
@@ -1696,7 +1698,8 @@ class KafkaApisTest {
val request = buildRequest(offsetCommitRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis().handleTxnOffsetCommitRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleTxnOffsetCommitRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[TxnOffsetCommitResponse](request)
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, response.errors().get(invalidTopicPartition))
@@ -1735,8 +1738,8 @@ class KafkaApisTest {
txnOffsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1780,8 +1783,8 @@ class KafkaApisTest {
txnOffsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1877,8 +1880,8 @@ class KafkaApisTest {
expectedTxnOffsetCommitRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -1990,8 +1993,8 @@ class KafkaApisTest {
.setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code)
).asJava)
).asJava))
-
- createKafkaApis().handleTxnOffsetCommitRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleTxnOffsetCommitRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2055,8 +2058,8 @@ class KafkaApisTest {
responseCallback.capture(),
ArgumentMatchers.eq(requestLocal)
)).thenAnswer(_ => responseCallback.getValue.apply(InitProducerIdResult(producerId, epoch, Errors.PRODUCER_FENCED)))
-
- createKafkaApis().handleInitProducerIdRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleInitProducerIdRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2113,8 +2116,8 @@ class KafkaApisTest {
responseCallback.capture(),
ArgumentMatchers.eq(requestLocal)
)).thenAnswer(_ => responseCallback.getValue.apply(Errors.PRODUCER_FENCED))
-
- createKafkaApis().handleAddOffsetsToTxnRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAddOffsetsToTxnRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2167,8 +2170,8 @@ class KafkaApisTest {
responseCallback.capture(),
ArgumentMatchers.eq(requestLocal)
)).thenAnswer(_ => responseCallback.getValue.apply(Errors.PRODUCER_FENCED))
-
- createKafkaApis().handleAddPartitionsToTxnRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAddPartitionsToTxnRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2246,8 +2249,8 @@ class KafkaApisTest {
ArgumentMatchers.eq(Set(tp1)),
verifyPartitionsCallback.capture(),
)).thenAnswer(_ => verifyPartitionsCallback.getValue.apply(AddPartitionsToTxnResponse.resultForTransaction(transactionalId2, Map(tp1 -> Errors.PRODUCER_FENCED).asJava)))
-
- createKafkaApis().handleAddPartitionsToTxnRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAddPartitionsToTxnRequest(request, requestLocal)
val response = verifyNoThrottling[AddPartitionsToTxnResponse](request)
@@ -2299,8 +2302,8 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handle(
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -2356,8 +2359,8 @@ class KafkaApisTest {
).asJava.iterator())).build(version)
val requestChannelRequest = buildRequest(addPartitionsToTxnRequest)
-
- createKafkaApis().handleAddPartitionsToTxnRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAddPartitionsToTxnRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -2410,8 +2413,8 @@ class KafkaApisTest {
responseCallback.capture(),
ArgumentMatchers.eq(requestLocal)
)).thenAnswer(_ => responseCallback.getValue.apply(Errors.PRODUCER_FENCED))
-
- createKafkaApis().handleEndTxnRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleEndTxnRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2471,8 +2474,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(clientQuotaManager.maybeRecordAndGetThrottleTimeMs(
any[RequestChannel.Request](), anyDouble, anyLong)).thenReturn(0)
-
- createKafkaApis().handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[ProduceResponse](request)
@@ -2534,8 +2537,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(clientQuotaManager.maybeRecordAndGetThrottleTimeMs(
any[RequestChannel.Request](), anyDouble, anyLong)).thenReturn(0)
-
- createKafkaApis().handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[ProduceResponse](request)
@@ -2598,8 +2601,8 @@ class KafkaApisTest {
any[Long])).thenReturn(0)
when(clientQuotaManager.maybeRecordAndGetThrottleTimeMs(
any[RequestChannel.Request](), anyDouble, anyLong)).thenReturn(0)
-
- createKafkaApis().handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[ProduceResponse](request)
@@ -2666,8 +2669,8 @@ class KafkaApisTest {
when(metadataCache.contains(tp)).thenAnswer(_ => true)
when(metadataCache.getPartitionInfo(tp.topic(), tp.partition())).thenAnswer(_ => Option.empty)
when(metadataCache.getAliveBrokerNode(any(), any())).thenReturn(Option.empty)
-
- createKafkaApis().handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[ProduceResponse](request)
@@ -2711,8 +2714,7 @@ class KafkaApisTest {
.build(version.toShort)
val request = buildRequest(produceRequest)
- val kafkaApis = createKafkaApis()
-
+ kafkaApis = createKafkaApis()
kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching)
verify(replicaManager).appendRecords(anyLong,
@@ -2745,7 +2747,8 @@ class KafkaApisTest {
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis().handleAddPartitionsToTxnRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAddPartitionsToTxnRequest(request, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[AddPartitionsToTxnResponse](request)
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, response.errors().get(AddPartitionsToTxnResponse.V3_AND_BELOW_TXN_ID).get(invalidTopicPartition))
@@ -2794,8 +2797,8 @@ class KafkaApisTest {
when(replicaManager.getMagic(topicPartition))
.thenReturn(Some(RecordBatch.MAGIC_VALUE_V1))
-
- createKafkaApis().handleWriteTxnMarkersRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleWriteTxnMarkersRequest(request, RequestLocal.withThreadConfinedCaching)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2815,7 +2818,8 @@ class KafkaApisTest {
when(replicaManager.getMagic(topicPartition))
.thenReturn(None)
- createKafkaApis().handleWriteTxnMarkersRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleWriteTxnMarkersRequest(request, RequestLocal.withThreadConfinedCaching)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2854,8 +2858,8 @@ class KafkaApisTest {
any(),
any()
)).thenAnswer(_ => responseCallback.getValue.apply(Map(tp2 -> new PartitionResponse(Errors.NONE))))
-
- createKafkaApis().handleWriteTxnMarkersRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleWriteTxnMarkersRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
@@ -2944,8 +2948,8 @@ class KafkaApisTest {
), Errors.NONE)
)
when(controller.brokerEpoch).thenReturn(brokerEpoch)
-
- createKafkaApis().handleStopReplicaRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleStopReplicaRequest(request)
if (deletePartition) {
if (leaderEpoch >= 0) {
@@ -2986,8 +2990,8 @@ class KafkaApisTest {
any(),
any()
)).thenAnswer(_ => responseCallback.getValue.apply(Map(tp2 -> new PartitionResponse(Errors.NONE))))
-
- createKafkaApis().handleWriteTxnMarkersRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleWriteTxnMarkersRequest(request, requestLocal)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
capturedResponse.capture(),
@@ -3006,8 +3010,8 @@ class KafkaApisTest {
.thenReturn(Some(RecordBatch.MAGIC_VALUE_V2))
val requestLocal = RequestLocal.withThreadConfinedCaching
-
- createKafkaApis().handleWriteTxnMarkersRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleWriteTxnMarkersRequest(request, requestLocal)
verify(replicaManager).appendRecords(anyLong,
anyShort,
ArgumentMatchers.eq(true),
@@ -3104,10 +3108,10 @@ class KafkaApisTest {
}.toMap
)
}
-
- createKafkaApis(overrideProperties = Map(
+ kafkaApis = createKafkaApis(overrideProperties = Map(
KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
- )).handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
+ ))
+ kafkaApis.handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
val expectedResponse = new WriteTxnMarkersResponseData()
.setMarkers(List(
@@ -3189,10 +3193,10 @@ class KafkaApisTest {
ArgumentMatchers.eq(TransactionResult.COMMIT),
ArgumentMatchers.eq(Duration.ofMillis(Defaults.RequestTimeoutMs))
)).thenReturn(FutureUtils.failedFuture[Void](error.exception()))
-
- createKafkaApis(overrideProperties = Map(
+ kafkaApis = createKafkaApis(overrideProperties = Map(
KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
- )).handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
+ ))
+ kafkaApis.handleWriteTxnMarkersRequest(requestChannelRequest, RequestLocal.NoCaching)
val expectedError = error match {
case Errors.COORDINATOR_NOT_AVAILABLE | Errors.COORDINATOR_LOAD_IN_PROGRESS | Errors.NOT_COORDINATOR =>
@@ -3274,8 +3278,8 @@ class KafkaApisTest {
List("group-1", "group-2", "group-3").asJava,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleDeleteGroupsRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDeleteGroupsRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -3317,8 +3321,8 @@ class KafkaApisTest {
List("group-1", "group-2", "group-3").asJava,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleDeleteGroupsRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDeleteGroupsRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -3376,8 +3380,8 @@ class KafkaApisTest {
List("group-2", "group-3").asJava,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDeleteGroupsRequest(
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDeleteGroupsRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -3422,8 +3426,8 @@ class KafkaApisTest {
requestChannelRequest.context,
describeGroupsRequest.groups
)).thenReturn(future)
-
- createKafkaApis().handleDescribeGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDescribeGroupsRequest(requestChannelRequest)
val groupResults = List(
new DescribeGroupsResponseData.DescribedGroup()
@@ -3464,8 +3468,8 @@ class KafkaApisTest {
requestChannelRequest.context,
describeGroupsRequest.groups
)).thenReturn(future)
-
- createKafkaApis().handleDescribeGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDescribeGroupsRequest(requestChannelRequest)
val expectedDescribeGroupsResponse = new DescribeGroupsResponseData().setGroups(List(
new DescribeGroupsResponseData.DescribedGroup()
@@ -3518,8 +3522,8 @@ class KafkaApisTest {
requestChannelRequest.context,
List("group-2").asJava
)).thenReturn(future)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDescribeGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDescribeGroupsRequest(requestChannelRequest)
future.complete(List(
new DescribeGroupsResponseData.DescribedGroup()
@@ -3576,8 +3580,8 @@ class KafkaApisTest {
offsetDeleteRequest.data,
requestLocal.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleOffsetDeleteRequest(request, requestLocal)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetDeleteRequest(request, requestLocal)
val offsetDeleteResponseData = new OffsetDeleteResponseData()
.setTopics(new OffsetDeleteResponseData.OffsetDeleteResponseTopicCollection(List(
@@ -3669,8 +3673,8 @@ class KafkaApisTest {
expectedOffsetDeleteRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -3776,8 +3780,8 @@ class KafkaApisTest {
)).thenReturn(CompletableFuture.completedFuture(
new OffsetDeleteResponseData()
))
-
- createKafkaApis().handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[OffsetDeleteResponse](request)
@@ -3806,8 +3810,8 @@ class KafkaApisTest {
offsetDeleteRequest.data,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
future.completeExceptionally(Errors.GROUP_ID_NOT_FOUND.exception)
@@ -3840,8 +3844,8 @@ class KafkaApisTest {
new OffsetDeleteRequestData().setGroupId(group), // Nonexistent topics won't be passed to groupCoordinator.
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetDeleteRequest(request, RequestLocal.NoCaching)
future.complete(new OffsetDeleteResponseData()
.setErrorCode(Errors.GROUP_ID_NOT_FOUND.code())
@@ -3876,8 +3880,8 @@ class KafkaApisTest {
val request = buildRequest(listOffsetRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis().handleListOffsetRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleListOffsetRequest(request)
val response = verifyNoThrottling[ListOffsetsResponse](request)
val partitionDataOptional = response.topics.asScala.find(_.name == tp.topic).get
@@ -4051,8 +4055,8 @@ class KafkaApisTest {
val repByTopicId = buildRequest(metadataReqByTopicId, plaintextListener)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer)).handleTopicMetadataRequest(repByTopicId)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleTopicMetadataRequest(repByTopicId)
val metadataByTopicIdResp = verifyNoThrottling[MetadataResponse](repByTopicId)
val metadataByTopicId = metadataByTopicIdResp.data().topics().asScala.groupBy(_.topicId()).map(kv => (kv._1, kv._2.head))
@@ -4073,8 +4077,8 @@ class KafkaApisTest {
reset(clientRequestQuotaManager, requestChannel)
val metadataReqByTopicName = new MetadataRequest.Builder(util.Arrays.asList(authorizedTopic, unauthorizedTopic), false).build()
val repByTopicName = buildRequest(metadataReqByTopicName, plaintextListener)
-
- createKafkaApis(authorizer = Some(authorizer)).handleTopicMetadataRequest(repByTopicName)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleTopicMetadataRequest(repByTopicName)
val metadataByTopicNameResp = verifyNoThrottling[MetadataResponse](repByTopicName)
val metadataByTopicName = metadataByTopicNameResp.data().topics().asScala.groupBy(_.name()).map(kv => (kv._1, kv._2.head))
@@ -4139,8 +4143,8 @@ class KafkaApisTest {
val fetchRequest = new FetchRequest.Builder(9, 9, -1, -1, 100, 0, fetchDataBuilder)
.build()
val request = buildRequest(fetchRequest)
-
- createKafkaApis().handleFetchRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleFetchRequest(request)
val response = verifyNoThrottling[FetchResponse](request)
val responseData = response.responseData(metadataCache.topicIdsToNames(), 9)
@@ -4195,8 +4199,8 @@ class KafkaApisTest {
val fetchRequest = new FetchRequest.Builder(ApiKeys.FETCH.latestVersion, ApiKeys.FETCH.latestVersion,
replicaId, replicaEpoch, 100, 0, fetchDataBuilder).metadata(fetchMetadata).build()
val request = buildRequest(fetchRequest)
-
- createKafkaApis().handleFetchRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleFetchRequest(request)
val response = verifyNoThrottling[FetchResponse](request)
val responseData = response.responseData(metadataCache.topicIdsToNames(), ApiKeys.FETCH.latestVersion)
@@ -4262,8 +4266,8 @@ class KafkaApisTest {
val fetchRequest = new FetchRequest.Builder(16, 16, -1, -1, 100, 0, fetchDataBuilder)
.build()
val request = buildRequest(fetchRequest)
-
- createKafkaApis().handleFetchRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleFetchRequest(request)
val response = verifyNoThrottling[FetchResponse](request)
val responseData = response.responseData(metadataCache.topicIdsToNames(), 16)
@@ -4302,8 +4306,8 @@ class KafkaApisTest {
expectedJoinGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleJoinGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleJoinGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4345,8 +4349,8 @@ class KafkaApisTest {
expectedJoinGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleJoinGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleJoinGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4383,8 +4387,8 @@ class KafkaApisTest {
joinGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleJoinGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleJoinGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4408,8 +4412,8 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleJoinGroupRequest(
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleJoinGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4442,8 +4446,8 @@ class KafkaApisTest {
}.thenAnswer { invocation =>
response = invocation.getArgument(1, classOf[JoinGroupResponse])
}
-
- createKafkaApis().handle(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4478,8 +4482,8 @@ class KafkaApisTest {
expectedSyncGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleSyncGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleSyncGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4515,8 +4519,8 @@ class KafkaApisTest {
expectedSyncGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleSyncGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleSyncGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4539,8 +4543,8 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleSyncGroupRequest(
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleSyncGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4568,8 +4572,8 @@ class KafkaApisTest {
expectedSyncGroupRequest,
RequestLocal.NoCaching.bufferSupplier
)).thenReturn(future)
-
- createKafkaApis().handleSyncGroupRequest(
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleSyncGroupRequest(
requestChannelRequest,
RequestLocal.NoCaching
)
@@ -4609,8 +4613,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedHeartbeatRequest
)).thenReturn(future)
-
- createKafkaApis().handleHeartbeatRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleHeartbeatRequest(requestChannelRequest)
val expectedHeartbeatResponse = new HeartbeatResponseData()
future.complete(expectedHeartbeatResponse)
@@ -4637,8 +4641,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedHeartbeatRequest
)).thenReturn(future)
-
- createKafkaApis().handleHeartbeatRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleHeartbeatRequest(requestChannelRequest)
future.completeExceptionally(Errors.UNKNOWN_SERVER_ERROR.exception)
val response = verifyNoThrottling[HeartbeatResponse](requestChannelRequest)
@@ -4657,8 +4661,8 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleHeartbeatRequest(
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleHeartbeatRequest(
requestChannelRequest
)
@@ -4678,8 +4682,8 @@ class KafkaApisTest {
).build()
val requestChannelRequest = buildRequest(joinGroupRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleJoinGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleJoinGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[JoinGroupResponse](requestChannelRequest)
assertEquals(Errors.UNSUPPORTED_VERSION, response.error())
@@ -4696,8 +4700,8 @@ class KafkaApisTest {
).build()
val requestChannelRequest = buildRequest(syncGroupRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleSyncGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleSyncGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[SyncGroupResponse](requestChannelRequest)
assertEquals(Errors.UNSUPPORTED_VERSION, response.error)
@@ -4713,8 +4717,8 @@ class KafkaApisTest {
.setGenerationId(1)
).build()
val requestChannelRequest = buildRequest(heartbeatRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleHeartbeatRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleHeartbeatRequest(requestChannelRequest)
val response = verifyNoThrottling[HeartbeatResponse](requestChannelRequest)
assertEquals(Errors.UNSUPPORTED_VERSION, response.error())
@@ -4742,8 +4746,8 @@ class KafkaApisTest {
).build()
val requestChannelRequest = buildRequest(offsetCommitRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleOffsetCommitRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleOffsetCommitRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
val expectedTopicErrors = Collections.singletonList(
new OffsetCommitResponseData.OffsetCommitResponseTopic()
@@ -4797,8 +4801,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedLeaveGroupRequest
)).thenReturn(future)
-
- createKafkaApis().handleLeaveGroupRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleLeaveGroupRequest(requestChannelRequest)
val expectedLeaveResponse = new LeaveGroupResponseData()
.setErrorCode(Errors.NONE.code)
@@ -4842,8 +4846,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedLeaveGroupRequest
)).thenReturn(future)
-
- createKafkaApis().handleLeaveGroupRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleLeaveGroupRequest(requestChannelRequest)
val leaveGroupResponse = new LeaveGroupResponseData()
.setErrorCode(Errors.NONE.code)
@@ -4895,8 +4899,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedLeaveGroupRequest
)).thenReturn(future)
-
- createKafkaApis().handleLeaveGroupRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleLeaveGroupRequest(requestChannelRequest)
future.completeExceptionally(Errors.UNKNOWN_SERVER_ERROR.exception)
val response = verifyNoThrottling[LeaveGroupResponse](requestChannelRequest)
@@ -4931,8 +4935,8 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleLeaveGroupRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleLeaveGroupRequest(requestChannelRequest)
val response = verifyNoThrottling[LeaveGroupResponse](requestChannelRequest)
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED, response.error)
@@ -5002,8 +5006,8 @@ class KafkaApisTest {
.setTopics(null),
false
)).thenReturn(group4Future)
-
- createKafkaApis().handleOffsetFetchRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetFetchRequest(requestChannelRequest)
val group1Response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
.setGroupId("group-1")
@@ -5094,8 +5098,8 @@ class KafkaApisTest {
.setPartitionIndexes(List[Integer](0, 1).asJava)).asJava),
false
)).thenReturn(future)
-
- createKafkaApis().handleOffsetFetchRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetFetchRequest(requestChannelRequest)
val group1Response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
.setGroupId("group-1")
@@ -5167,8 +5171,8 @@ class KafkaApisTest {
.setTopics(null),
false
)).thenReturn(future)
-
- createKafkaApis().handleOffsetFetchRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleOffsetFetchRequest(requestChannelRequest)
val group1Response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
.setGroupId("group-1")
@@ -5276,8 +5280,8 @@ class KafkaApisTest {
.setTopics(null),
false
)).thenReturn(group3Future)
-
- createKafkaApis(authorizer = Some(authorizer)).handle(requestChannelRequest, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val group1ResponseFromCoordinator = new OffsetFetchResponseData.OffsetFetchResponseGroup()
.setGroupId("group-1")
@@ -5426,8 +5430,8 @@ class KafkaApisTest {
.setPartitionIndexes(List[Integer](0).asJava)).asJava),
false
)).thenReturn(group1Future)
-
- createKafkaApis(authorizer = Some(authorizer)).handle(requestChannelRequest, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
// group-2 mocks using the new group coordinator.
// When the coordinator is not active, a response with top-level error code is returned
@@ -5501,8 +5505,8 @@ class KafkaApisTest {
when(replicaManager.getLogConfig(ArgumentMatchers.eq(tp0))).thenReturn(None)
when(replicaManager.isAddingReplica(any(), anyInt)).thenReturn(isReassigning)
-
- createKafkaApis().handle(fetchFromFollower, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(fetchFromFollower, RequestLocal.withThreadConfinedCaching)
verify(replicaQuotaManager).record(anyLong)
if (isReassigning)
@@ -5523,8 +5527,8 @@ class KafkaApisTest {
).build()
val requestChannelRequest = buildRequest(initProducerIdRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[InitProducerIdResponse](requestChannelRequest)
assertEquals(Errors.INVALID_REQUEST, response.error)
@@ -5540,8 +5544,8 @@ class KafkaApisTest {
.setProducerEpoch(2)
).build()
val requestChannelRequest = buildRequest(initProducerIdRequest)
-
- createKafkaApis(IBP_2_2_IV1).handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis(IBP_2_2_IV1)
+ kafkaApis.handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
val response = verifyNoThrottling[InitProducerIdResponse](requestChannelRequest)
assertEquals(Errors.INVALID_REQUEST, response.error)
@@ -5578,8 +5582,8 @@ class KafkaApisTest {
)).thenReturn(
Seq()
)
-
- createKafkaApis().handleUpdateMetadataRequest(request, RequestLocal.withThreadConfinedCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleUpdateMetadataRequest(request, RequestLocal.withThreadConfinedCaching)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
capturedResponse.capture(),
@@ -5651,8 +5655,8 @@ class KafkaApisTest {
)).thenReturn(
response
)
-
- createKafkaApis().handleLeaderAndIsrRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleLeaderAndIsrRequest(request)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
capturedResponse.capture(),
@@ -5714,8 +5718,8 @@ class KafkaApisTest {
fooPartition -> Errors.NONE
), Errors.NONE)
)
-
- createKafkaApis().handleStopReplicaRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleStopReplicaRequest(request)
verify(requestChannel).sendResponse(
ArgumentMatchers.eq(request),
capturedResponse.capture(),
@@ -5749,8 +5753,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedListGroupsRequest
)).thenReturn(future)
-
- createKafkaApis().handleListGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleListGroupsRequest(requestChannelRequest)
val expectedListGroupsResponse = new ListGroupsResponseData()
.setGroups(List(
@@ -5784,8 +5788,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedListGroupsRequest
)).thenReturn(future)
-
- createKafkaApis().handleListGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleListGroupsRequest(requestChannelRequest)
future.completeExceptionally(Errors.UNKNOWN_SERVER_ERROR.exception)
val response = verifyNoThrottling[ListGroupsResponse](requestChannelRequest)
@@ -5880,8 +5884,8 @@ class KafkaApisTest {
requestChannelRequest.context,
expectedListGroupsRequest
)).thenReturn(future)
-
- createKafkaApis(authorizer = Some(authorizer)).handleListGroupsRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleListGroupsRequest(requestChannelRequest)
val listGroupsResponse = new ListGroupsResponseData()
groups.foreach { groupId =>
@@ -5934,8 +5938,8 @@ class KafkaApisTest {
.setIncludeClusterAuthorizedOperations(true)).build()
val request = buildRequest(describeClusterRequest, plaintextListener)
-
- createKafkaApis().handleDescribeCluster(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleDescribeCluster(request)
val describeClusterResponse = verifyNoThrottling[DescribeClusterResponse](request)
@@ -5987,8 +5991,8 @@ class KafkaApisTest {
private def sendMetadataRequestWithInconsistentListeners(requestListener: ListenerName): MetadataResponse = {
val metadataRequest = MetadataRequest.Builder.allTopics.build()
val requestChannelRequest = buildRequest(metadataRequest, requestListener)
-
- createKafkaApis().handleTopicMetadataRequest(requestChannelRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleTopicMetadataRequest(requestChannelRequest)
verifyNoThrottling[MetadataResponse](requestChannelRequest)
}
@@ -6014,8 +6018,8 @@ class KafkaApisTest {
val listOffsetRequest = ListOffsetsRequest.Builder.forConsumer(true, isolationLevel, false)
.setTargetTimes(targetTimes).build()
val request = buildRequest(listOffsetRequest)
-
- createKafkaApis().handleListOffsetRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleListOffsetRequest(request)
val response = verifyNoThrottling[ListOffsetsResponse](request)
val partitionDataOptional = response.topics.asScala.find(_.name == tp.topic).get
@@ -6182,8 +6186,8 @@ class KafkaApisTest {
t0p1 -> "/foo",
t0p2 -> "/foo"))))
.thenReturn(partitionResults)
-
- createKafkaApis().handleAlterReplicaLogDirsRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleAlterReplicaLogDirsRequest(request)
val response = verifyNoThrottling[AlterReplicaLogDirsResponse](request)
assertEquals(partitionResults, response.data.results.asScala.flatMap { tr =>
@@ -6293,8 +6297,8 @@ class KafkaApisTest {
val request = buildRequest(describeProducersRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDescribeProducersRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDescribeProducersRequest(request)
val response = verifyNoThrottling[DescribeProducersResponse](request)
assertEquals(Set("foo", "bar", "baz", "invalid;topic"), response.data.topics.asScala.map(_.name).toSet)
@@ -6356,8 +6360,8 @@ class KafkaApisTest {
when(authorizer.authorize(any[RequestContext], ArgumentMatchers.eq(buildExpectedActions("bar"))))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDescribeTransactionsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDescribeTransactionsRequest(request)
val response = verifyNoThrottling[DescribeTransactionsResponse](request)
assertEquals(2, response.data.transactionStates.size)
@@ -6428,8 +6432,8 @@ class KafkaApisTest {
when(txnCoordinator.handleDescribeTransactions("foo"))
.thenReturn(describeTransactionsResponse)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDescribeTransactionsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDescribeTransactionsRequest(request)
val response = verifyNoThrottling[DescribeTransactionsResponse](request)
assertEquals(1, response.data.transactionStates.size)
@@ -6455,8 +6459,8 @@ class KafkaApisTest {
when(txnCoordinator.handleListTransactions(Set.empty[Long], Set.empty[String]))
.thenReturn(new ListTransactionsResponseData()
.setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code))
-
- createKafkaApis().handleListTransactionsRequest(request)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleListTransactionsRequest(request)
val response = verifyNoThrottling[ListTransactionsResponse](request)
assertEquals(0, response.data.transactionStates.size)
@@ -6498,8 +6502,8 @@ class KafkaApisTest {
when(authorizer.authorize(any[RequestContext], ArgumentMatchers.eq(buildExpectedActions("bar"))))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(authorizer = Some(authorizer)).handleListTransactionsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleListTransactionsRequest(request)
val response = verifyNoThrottling[ListTransactionsResponse](request)
assertEquals(1, response.data.transactionStates.size())
@@ -6564,8 +6568,8 @@ class KafkaApisTest {
val request = buildRequest(deleteRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDeleteTopicsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDeleteTopicsRequest(request)
verify(authorizer, times(2)).authorize(any(), any())
val deleteResponse = verifyNoThrottling[DeleteTopicsResponse](request)
@@ -6643,8 +6647,8 @@ class KafkaApisTest {
val request = buildRequest(deleteRequest)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(authorizer = Some(authorizer)).handleDeleteTopicsRequest(request)
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handleDeleteTopicsRequest(request)
verify(authorizer, times(2)).authorize(any(), any())
val deleteResponse = verifyNoThrottling[DeleteTopicsResponse](request)
@@ -6751,8 +6755,8 @@ class KafkaApisTest {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
-
- createKafkaApis(raftSupport = true).handleAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handleAlterConfigsRequest(request)
val response = verifyNoThrottling[AlterConfigsResponse](request)
assertEquals(new AlterConfigsResponseData(), response.data())
}
@@ -6771,7 +6775,8 @@ class KafkaApisTest {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis(raftSupport = true).handleAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handleAlterConfigsRequest(request)
val response = verifyNoThrottling[AlterConfigsResponse](request)
assertEquals(new AlterConfigsResponseData().setResponses(asList(
new LAlterConfigsResourceResponse().
@@ -6794,7 +6799,8 @@ class KafkaApisTest {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis(raftSupport = true).handleIncrementalAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handleIncrementalAlterConfigsRequest(request)
val response = verifyNoThrottling[IncrementalAlterConfigsResponse](request)
assertEquals(new IncrementalAlterConfigsResponseData(), response.data())
}
@@ -6813,7 +6819,8 @@ class KafkaApisTest {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](),
any[Long])).thenReturn(0)
- createKafkaApis(raftSupport = true).handleIncrementalAlterConfigsRequest(request)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handleIncrementalAlterConfigsRequest(request)
val response = verifyNoThrottling[IncrementalAlterConfigsResponse](request)
assertEquals(new IncrementalAlterConfigsResponseData().setResponses(asList(
new IAlterConfigsResourceResponse().
@@ -6883,8 +6890,8 @@ class KafkaApisTest {
val consumerGroupHeartbeatRequest = new ConsumerGroupHeartbeatRequestData().setGroupId("group")
val requestChannelRequest = buildRequest(new ConsumerGroupHeartbeatRequest.Builder(consumerGroupHeartbeatRequest, true).build())
-
- createKafkaApis().handle(requestChannelRequest, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val expectedHeartbeatResponse = new ConsumerGroupHeartbeatResponseData()
.setErrorCode(Errors.UNSUPPORTED_VERSION.code)
@@ -6903,10 +6910,10 @@ class KafkaApisTest {
requestChannelRequest.context,
consumerGroupHeartbeatRequest
)).thenReturn(future)
-
- createKafkaApis(overrideProperties = Map(
+ kafkaApis = createKafkaApis(overrideProperties = Map(
KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
- )).handle(requestChannelRequest, RequestLocal.NoCaching)
+ ))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val consumerGroupHeartbeatResponse = new ConsumerGroupHeartbeatResponseData()
.setMemberId("member")
@@ -6927,10 +6934,10 @@ class KafkaApisTest {
requestChannelRequest.context,
consumerGroupHeartbeatRequest
)).thenReturn(future)
-
- createKafkaApis(overrideProperties = Map(
+ kafkaApis = createKafkaApis(overrideProperties = Map(
KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
- )).handle(requestChannelRequest, RequestLocal.NoCaching)
+ ))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
val response = verifyNoThrottling[ConsumerGroupHeartbeatResponse](requestChannelRequest)
@@ -6946,11 +6953,11 @@ class KafkaApisTest {
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
-
- createKafkaApis(
+ kafkaApis = createKafkaApis(
authorizer = Some(authorizer),
overrideProperties = Map(KafkaConfig.NewGroupCoordinatorEnableProp -> "true")
- ).handle(requestChannelRequest, RequestLocal.NoCaching)
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val response = verifyNoThrottling[ConsumerGroupHeartbeatResponse](requestChannelRequest)
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.errorCode)
@@ -6968,10 +6975,10 @@ class KafkaApisTest {
any[RequestContext],
any[util.List[String]]
)).thenReturn(future)
-
- createKafkaApis(
+ kafkaApis = createKafkaApis(
overrideProperties = Map(KafkaConfig.NewGroupCoordinatorEnableProp -> "true")
- ).handle(requestChannelRequest, RequestLocal.NoCaching)
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val describedGroups = List(
new DescribedGroup().setGroupId(groupIds.get(0)),
@@ -6999,8 +7006,8 @@ class KafkaApisTest {
val expectedDescribedGroup = new DescribedGroup().setGroupId(groupId).setErrorCode(errorCode)
val expectedResponse = new ConsumerGroupDescribeResponseData()
expectedResponse.groups.add(expectedDescribedGroup)
-
- createKafkaApis().handle(requestChannelRequest, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
assertEquals(expectedResponse, response.data)
@@ -7022,11 +7029,11 @@ class KafkaApisTest {
any[util.List[String]]
)).thenReturn(future)
future.complete(List().asJava)
-
- createKafkaApis(
+ kafkaApis = createKafkaApis(
authorizer = Some(authorizer),
overrideProperties = Map(KafkaConfig.NewGroupCoordinatorEnableProp -> "true")
- ).handle(requestChannelRequest, RequestLocal.NoCaching)
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.groups.get(0).errorCode)
@@ -7043,10 +7050,10 @@ class KafkaApisTest {
any[RequestContext],
any[util.List[String]]
)).thenReturn(future)
-
- createKafkaApis(overrideProperties = Map(
+ kafkaApis = createKafkaApis(overrideProperties = Map(
KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
- )).handle(requestChannelRequest, RequestLocal.NoCaching)
+ ))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.NoCaching)
future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
@@ -7058,7 +7065,8 @@ class KafkaApisTest {
val data = new GetTelemetrySubscriptionsRequestData()
val request = buildRequest(new GetTelemetrySubscriptionsRequest.Builder(data, true).build())
- createKafkaApis(enableForwarding = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[GetTelemetrySubscriptionsResponse](request)
assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
@@ -7075,7 +7083,8 @@ class KafkaApisTest {
new GetTelemetrySubscriptionsResponseData()))
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[GetTelemetrySubscriptionsResponse](request)
@@ -7093,7 +7102,8 @@ class KafkaApisTest {
any[RequestContext]())).thenThrow(new RuntimeException("test"))
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[GetTelemetrySubscriptionsResponse](request)
@@ -7106,7 +7116,8 @@ class KafkaApisTest {
val data = new PushTelemetryRequestData()
val request = buildRequest(new PushTelemetryRequest.Builder(data, true).build())
- createKafkaApis(enableForwarding = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[PushTelemetryResponse](request)
assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
@@ -7121,7 +7132,8 @@ class KafkaApisTest {
.thenReturn(new PushTelemetryResponse(new PushTelemetryResponseData()))
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[PushTelemetryResponse](request)
val expectedResponse = new PushTelemetryResponseData().setErrorCode(Errors.NONE.code)
@@ -7137,7 +7149,8 @@ class KafkaApisTest {
.thenThrow(new RuntimeException("test"))
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[PushTelemetryResponse](request)
val expectedResponse = new PushTelemetryResponseData().setErrorCode(Errors.INVALID_REQUEST.code)
@@ -7147,7 +7160,8 @@ class KafkaApisTest {
@Test
def testListClientMetricsResourcesNotAllowedForZkClusters(): Unit = {
val request = buildRequest(new ListClientMetricsResourcesRequest.Builder(new ListClientMetricsResourcesRequestData()).build())
- createKafkaApis(enableForwarding = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(enableForwarding = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[ListClientMetricsResourcesResponse](request)
assertEquals(Errors.UNKNOWN_SERVER_ERROR, Errors.forCode(response.data.errorCode))
@@ -7162,8 +7176,8 @@ class KafkaApisTest {
resources.add("test1")
resources.add("test2")
when(clientMetricsManager.listClientMetricsResources).thenReturn(resources.asJava)
-
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[ListClientMetricsResourcesResponse](request)
val expectedResponse = new ListClientMetricsResourcesResponseData().setClientMetricsResources(
resources.map(resource => new ClientMetricsResource().setName(resource)).toBuffer.asJava)
@@ -7177,8 +7191,8 @@ class KafkaApisTest {
val resources = new mutable.HashSet[String]
when(clientMetricsManager.listClientMetricsResources).thenReturn(resources.asJava)
-
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[ListClientMetricsResourcesResponse](request)
val expectedResponse = new ListClientMetricsResourcesResponseData()
assertEquals(expectedResponse, response.data)
@@ -7190,8 +7204,8 @@ class KafkaApisTest {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
when(clientMetricsManager.listClientMetricsResources).thenThrow(new RuntimeException("test"))
-
- createKafkaApis(raftSupport = true).handle(request, RequestLocal.NoCaching)
+ kafkaApis = createKafkaApis(raftSupport = true)
+ kafkaApis.handle(request, RequestLocal.NoCaching)
val response = verifyNoThrottling[ListClientMetricsResourcesResponse](request)
val expectedResponse = new ListClientMetricsResourcesResponseData().setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code)