Skip to content

MINOR: Filter out quota configs for ConfigCommand using --bootstrap-server #9030

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
rajinisivaram merged 2 commits into from
Jul 17, 2020
Merged

MINOR: Filter out quota configs for ConfigCommand using --bootstrap-server #9030

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1acc25d
MINOR: Filter out quota configs for ConfigCommand using --bootstrap-s…
rajinisivaram Jul 16, 2020
c150a16
Address review comments
rajinisivaram Jul 16, 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
13 changes: 9 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/requests/AlterClientQuotasResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,12 @@
import org.apache.kafka.common.message.AlterClientQuotasResponseData;
import org.apache.kafka.common.message.AlterClientQuotasResponseData.EntityData;
import org.apache.kafka.common.message.AlterClientQuotasResponseData.EntryData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.quota.ClientQuotaEntity;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
Expand All @@ -50,14 +52,13 @@ public AlterClientQuotasResponse(Map<ClientQuotaEntity, ApiError> result, int th
}

public AlterClientQuotasResponse(Collection<ClientQuotaEntity> entities, int throttleTimeMs, Throwable e) {
short errorCode = Errors.forException(e).code();
String errorMessage = e.getMessage();
ApiError apiError = ApiError.fromThrowable(e);

List<EntryData> entries = new ArrayList<>(entities.size());
for (ClientQuotaEntity entity : entities) {
entries.add(new EntryData()
.setErrorCode(errorCode)
.setErrorMessage(errorMessage)
.setErrorCode(apiError.error().code())
.setErrorMessage(apiError.message())
.setEntity(toEntityData(entity)));
}

Expand Down Expand Up @@ -120,4 +121,8 @@ private static List<EntityData> toEntityData(ClientQuotaEntity entity) {
}
return entityData;
}

public static AlterClientQuotasResponse parse(ByteBuffer buffer, short version) {
return new AlterClientQuotasResponse(ApiKeys.ALTER_CLIENT_QUOTAS.parseResponse(version, buffer), version);
}
}
5 changes: 3 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/ApiError.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,9 +38,10 @@ public class ApiError {
private final String message;

public static ApiError fromThrowable(Throwable t) {
// Avoid populating the error message if it's a generic one
// Avoid populating the error message if it's a generic one. Also don't populate error
// message for UNKNOWN_SERVER_ERROR to ensure we don't leak sensitive information.
Errors error = Errors.forException(t);
String message = error.message().equals(t.getMessage()) ? null : t.getMessage();
String message = error == Errors.UNKNOWN_SERVER_ERROR || error.message().equals(t.getMessage()) ? null : t.getMessage();
return new ApiError(error, message);
}

Expand Down
11 changes: 9 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/DescribeClientQuotasResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,12 @@
import org.apache.kafka.common.message.DescribeClientQuotasResponseData.EntityData;
import org.apache.kafka.common.message.DescribeClientQuotasResponseData.EntryData;
import org.apache.kafka.common.message.DescribeClientQuotasResponseData.ValueData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.quota.ClientQuotaEntity;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -66,10 +68,11 @@ public DescribeClientQuotasResponse(Map<ClientQuotaEntity, Map<String, Double>>
}

public DescribeClientQuotasResponse(int throttleTimeMs, Throwable e) {
ApiError apiError = ApiError.fromThrowable(e);

@rondagostino rondagostino Jul 16, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Should the same thing happen in public AlterClientQuotasResponse(Collection<ClientQuotaEntity> entities, int throttleTimeMs, Throwable e) for consistency? And also maybe in IncrementalAlterConfigsResponse (public static IncrementalAlterConfigsResponseData toResponseData)?

@rajinisivaram rajinisivaram Jul 16, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@rondagostino Thanks for the review. Updated AlterClientQuotasResponse. I think IncrementalAlterConfigsResponse is ok since it uses ApiError, we create the error with the right message when converting from Throwable. Can you just verify that the method with ApiError was the one you meant? Thanks,

this.data = new DescribeClientQuotasResponseData()
.setThrottleTimeMs(throttleTimeMs)
.setErrorCode(Errors.forException(e).code())
.setErrorMessage(e.getMessage())
.setErrorCode(apiError.error().code())
.setErrorMessage(apiError.message())
.setEntries(null);
}

Expand Down Expand Up @@ -115,4 +118,8 @@ public Map<Errors, Integer> errorCounts() {
protected Struct toStruct(short version) {
return data.toStruct(version);
}

public static DescribeClientQuotasResponse parse(ByteBuffer buffer, short version) {
return new DescribeClientQuotasResponse(ApiKeys.DESCRIBE_CLIENT_QUOTAS.parseResponse(version, buffer), version);
}
}
5 changes: 3 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/SaslAuthenticateRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,10 @@ public SaslAuthenticateRequestData data() {

@Override
public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
ApiError apiError = ApiError.fromThrowable(e);
SaslAuthenticateResponseData response = new SaslAuthenticateResponseData()
.setErrorCode(ApiError.fromThrowable(e).error().code())
.setErrorMessage(e.getMessage());
.setErrorCode(apiError.error().code())
.setErrorMessage(apiError.message());
return new SaslAuthenticateResponse(response);
}

Expand Down
Loading