Skip to content

KAFKA-10852: AlterIsr should not be throttled #9747

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 4 commits into from
Dec 15, 2020
Merged

KAFKA-10852: AlterIsr should not be throttled #9747

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6cae9e7
KAFKA-10852: AlterIsr should be a cluster action
ijuma Dec 14, 2020
75343d9
Don't throttle alter isr for now
ijuma Dec 14, 2020
e49d24b
Fix failing test
ijuma Dec 15, 2020
c48d020
Clean-ups
ijuma Dec 15, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,7 +251,7 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
EndQuorumEpochRequestData.SCHEMAS, EndQuorumEpochResponseData.SCHEMAS),
DESCRIBE_QUORUM(55, "DescribeQuorum", true, false,
DescribeQuorumRequestData.SCHEMAS, DescribeQuorumResponseData.SCHEMAS),
ALTER_ISR(56, "AlterIsr", AlterIsrRequestData.SCHEMAS, AlterIsrResponseData.SCHEMAS),
ALTER_ISR(56, "AlterIsr", true, AlterIsrRequestData.SCHEMAS, AlterIsrResponseData.SCHEMAS),
UPDATE_FEATURES(57, "UpdateFeatures",
UpdateFeaturesRequestData.SCHEMAS, UpdateFeaturesResponseData.SCHEMAS, true),
ENVELOPE(58, "Envelope", true, false, EnvelopeRequestData.SCHEMAS, EnvelopeResponseData.SCHEMAS);
Expand Down
17 changes: 13 additions & 4 deletions clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,12 @@
import org.apache.kafka.common.protocol.types.Schema;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.EnumSet;
import java.util.Set;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class ApiKeysTest {

Expand All @@ -43,6 +44,11 @@ public void schemaVersionOutOfRange() {
ApiKeys.PRODUCE.requestSchema((short) ApiKeys.PRODUCE.requestSchemas.length);
}

@Test
public void testAlterIsrIsClusterAction() {
assertTrue(ApiKeys.ALTER_ISR.clusterAction);
}

/**
* All valid client responses which may be throttled should have a field named
* 'throttle_time_ms' to return the throttle time to the client. Exclusions are
Expand All @@ -55,11 +61,14 @@ public void schemaVersionOutOfRange() {
*/
@Test
public void testResponseThrottleTime() {
List<ApiKeys> authenticationKeys = Arrays.asList(ApiKeys.SASL_HANDSHAKE, ApiKeys.SASL_AUTHENTICATE);
Set<ApiKeys> authenticationKeys = EnumSet.of(ApiKeys.SASL_HANDSHAKE, ApiKeys.SASL_AUTHENTICATE);
// Newer protocol apis include throttle time ms even for cluster actions
Set<ApiKeys> clusterActionsWithThrottleTimeMs = EnumSet.of(ApiKeys.ALTER_ISR);

@ijuma ijuma Dec 14, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have to decide whether we want to do this or remove the throttle time ms from AlterIsr before 2.7.0.

@ijuma ijuma Dec 14, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc @bbejeck

@ijuma ijuma Dec 14, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We decided this is not a blocker.

for (ApiKeys apiKey: ApiKeys.values()) {
Schema responseSchema = apiKey.responseSchema(apiKey.latestVersion());
BoundField throttleTimeField = responseSchema.get(CommonFields.THROTTLE_TIME_MS.name);
if (apiKey.clusterAction || authenticationKeys.contains(apiKey))
if ((apiKey.clusterAction && !clusterActionsWithThrottleTimeMs.contains(apiKey))
|| authenticationKeys.contains(apiKey))
assertNull("Unexpected throttle time field: " + apiKey, throttleTimeField);
else
assertNotNull("Throttle time field missing: " + apiKey, throttleTimeField);
Expand Down
19 changes: 7 additions & 12 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3238,20 +3238,15 @@ class KafkaApis(val requestChannel: RequestChannel,

def handleAlterIsrRequest(request: RequestChannel.Request): Unit = {
val alterIsrRequest = request.body[AlterIsrRequest]
authorizeClusterOperation(request, CLUSTER_ACTION)

if (!authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) {
sendResponseMaybeThrottle(request, requestThrottleMs =>
alterIsrRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
} else if (!controller.isActive) {
sendResponseMaybeThrottle(request, requestThrottleMs =>
alterIsrRequest.getErrorResponse(requestThrottleMs, Errors.NOT_CONTROLLER.exception()))
} else {
controller.alterIsrs(alterIsrRequest.data,
alterIsrResp => sendResponseMaybeThrottle(request, requestThrottleMs =>
new AlterIsrResponse(alterIsrResp.setThrottleTimeMs(requestThrottleMs))
)
if (!controller.isActive)
sendResponseExemptThrottle(request, alterIsrRequest.getErrorResponse(
AbstractResponse.DEFAULT_THROTTLE_TIME, Errors.NOT_CONTROLLER.exception))
else
controller.alterIsrs(alterIsrRequest.data, alterIsrResp =>
sendResponseExemptThrottle(request, new AlterIsrResponse(alterIsrResp))
)
}
}

def handleUpdateFeatures(request: RequestChannel.Request): Unit = {
Expand Down
6 changes: 3 additions & 3 deletions core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -718,7 +718,7 @@ class RequestQuotaTest extends BaseRequestTest {
val exemptTarget = exemptRequestMetricValue + 0.02
val clientId = apiKey.toString
val client = Client(clientId, apiKey)
val updated = client.runUntil(response => exemptRequestMetricValue > exemptTarget)
val updated = client.runUntil(_ => exemptRequestMetricValue > exemptTarget)

assertTrue(s"Exempt-request-time metric not updated: $client", updated)
assertTrue(s"Client should not have been throttled: $client", throttleTimeMetricValue(clientId).isNaN)
Expand All @@ -727,13 +727,13 @@ class RequestQuotaTest extends BaseRequestTest {
private def checkUnauthorizedRequestThrottle(apiKey: ApiKeys): Unit = {
val clientId = "unauthorized-" + apiKey.toString
val client = Client(clientId, apiKey)
val throttled = client.runUntil(response => throttleTimeMetricValue(clientId) > 0.0)
val throttled = client.runUntil(_ => throttleTimeMetricValue(clientId) > 0.0)
assertTrue(s"Unauthorized client should have been throttled: $client", throttled)
}
}

object RequestQuotaTest {
val ClusterActions = ApiKeys.enabledApis.asScala.toSet.filter(apiKey => apiKey.clusterAction)
val ClusterActions = ApiKeys.enabledApis.asScala.filter(_.clusterAction).toSet
val SaslActions = Set(ApiKeys.SASL_HANDSHAKE, ApiKeys.SASL_AUTHENTICATE)
val ClientActions = ApiKeys.enabledApis.asScala.toSet -- ClusterActions -- SaslActions

Expand Down