Skip to content

KAFKA-9432 automated protocol for DescribeConfigs #8312

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
mimaison merged 7 commits into from
Jun 12, 2020
Merged

KAFKA-9432 automated protocol for DescribeConfigs #8312

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
141ca72
KAFKA-9432 automated protocol for DescribeConfigs
tombentley Jan 17, 2020
ca2e5e8
rebase
tombentley May 12, 2020
2962bf7
Review comments
tombentley May 27, 2020
55783db
Fix rebase
tombentley Jun 8, 2020
c89ec96
Better test
tombentley Jun 8, 2020
b5b3006
Use single request when getting BROKER & BROKER_LOGGER from same broker
tombentley Jun 8, 2020
5a9c778
Review comments
tombentley Jun 11, 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
185 changes: 80 additions & 105 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@
import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.ElectionType;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.Node;
Expand Down Expand Up @@ -102,6 +101,8 @@
import org.apache.kafka.common.message.DeleteRecordsResponseData.DeleteRecordsTopicResult;
import org.apache.kafka.common.message.DeleteTopicsRequestData;
import org.apache.kafka.common.message.DeleteTopicsResponseData.DeletableTopicResult;
import org.apache.kafka.common.message.DescribeConfigsRequestData;
import org.apache.kafka.common.message.DescribeConfigsResponseData;
import org.apache.kafka.common.message.DescribeGroupsRequestData;
import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroup;
import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroupMember;
Expand Down Expand Up @@ -1916,129 +1917,96 @@ void handleFailure(Throwable throwable) {

@Override
public DescribeConfigsResult describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options) {
final Map<ConfigResource, KafkaFutureImpl<Config>> unifiedRequestFutures = new HashMap<>();
final Map<ConfigResource, KafkaFutureImpl<Config>> brokerFutures = new HashMap<>(configResources.size());

// The BROKER resources which we want to describe. We must make a separate DescribeConfigs
// request for every BROKER resource we want to describe.
final Collection<ConfigResource> brokerResources = new ArrayList<>();

// The non-BROKER resources which we want to describe. These resources can be described by a
// single, unified DescribeConfigs request.
final Collection<ConfigResource> unifiedRequestResources = new ArrayList<>(configResources.size());
// Partition the requested config resources based on which broker they must be sent to with the
// null broker being used for config resources which can be obtained from any broker
final Map<Integer, Map<ConfigResource, KafkaFutureImpl<Config>>> brokerFutures = new HashMap<>(configResources.size());

for (ConfigResource resource : configResources) {
if (dependsOnSpecificNode(resource)) {
brokerFutures.put(resource, new KafkaFutureImpl<>());
brokerResources.add(resource);
} else {
unifiedRequestFutures.put(resource, new KafkaFutureImpl<>());
unifiedRequestResources.add(resource);
}
Integer broker = nodeFor(resource);
brokerFutures.compute(broker, (key, value) -> {
if (value == null) {
// Only BROKER and BROKER_LOGGER configs are broker-specific
value = new HashMap<>(broker != null ? 2 : 16);
Comment thread
tombentley marked this conversation as resolved.
Outdated
}
value.put(resource, new KafkaFutureImpl<>());
return value;
});
}

final long now = time.milliseconds();
if (!unifiedRequestResources.isEmpty()) {
for (Map.Entry<Integer, Map<ConfigResource, KafkaFutureImpl<Config>>> entry : brokerFutures.entrySet()) {
Integer broker = entry.getKey();
Map<ConfigResource, KafkaFutureImpl<Config>> unified = entry.getValue();

runnable.call(new Call("describeConfigs", calcDeadlineMs(now, options.timeoutMs()),
new LeastLoadedNodeProvider()) {
broker != null ? new ConstantNodeIdProvider(broker) : new LeastLoadedNodeProvider()) {

@Override
DescribeConfigsRequest.Builder createRequest(int timeoutMs) {
return new DescribeConfigsRequest.Builder(unifiedRequestResources)
.includeSynonyms(options.includeSynonyms())
.includeDocumentation(options.includeDocumentation());
return new DescribeConfigsRequest.Builder(new DescribeConfigsRequestData()
.setResources(unified.keySet().stream()
.map(config ->
new DescribeConfigsRequestData.DescribeConfigsResource()
.setResourceName(config.name())
.setResourceType(config.type().id()))
.collect(Collectors.toList()))
.setIncludeSynonyms(options.includeSynonyms())
.setIncludeDocumentation(options.includeDocumentation()));
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
DescribeConfigsResponse response = (DescribeConfigsResponse) abstractResponse;
for (Map.Entry<ConfigResource, KafkaFutureImpl<Config>> entry : unifiedRequestFutures.entrySet()) {
for (Map.Entry<ConfigResource, DescribeConfigsResponseData.DescribeConfigsResult> entry : response.resultMap().entrySet()) {
ConfigResource configResource = entry.getKey();
KafkaFutureImpl<Config> future = entry.getValue();
DescribeConfigsResponse.Config config = response.config(configResource);
if (config == null) {
future.completeExceptionally(new UnknownServerException(
"Malformed broker response: missing config for " + configResource));
continue;
}
if (config.error().isFailure()) {
future.completeExceptionally(config.error().exception());
continue;
}
List<ConfigEntry> configEntries = new ArrayList<>();
for (DescribeConfigsResponse.ConfigEntry configEntry : config.entries()) {
configEntries.add(new ConfigEntry(configEntry.name(),
configEntry.value(), configSource(configEntry.source()),
configEntry.isSensitive(), configEntry.isReadOnly(),
configSynonyms(configEntry), configType(configEntry.type()),
configEntry.documentation()));
DescribeConfigsResponseData.DescribeConfigsResult describeConfigsResult = entry.getValue();
KafkaFutureImpl<Config> future = unified.get(configResource);
if (future == null) {
if (broker != null) {
log.warn("The config {} in the response from broker {} is not in the request",
configResource, broker);
} else {
log.warn("The config {} in the response from the least loaded broker is not in the request",
configResource);
}
} else {
if (describeConfigsResult.errorCode() != Errors.NONE.code()) {
future.completeExceptionally(Errors.forCode(describeConfigsResult.errorCode())
.exception(describeConfigsResult.errorMessage()));
} else {
future.complete(describeConfigResult(describeConfigsResult));
}
}
future.complete(new Config(configEntries));
}
completeUnrealizedFutures(
unified.entrySet().stream(),
configResource -> "The broker response did not contain a result for config resource " + configResource);
}

@Override
void handleFailure(Throwable throwable) {
completeAllExceptionally(unifiedRequestFutures.values(), throwable);
completeAllExceptionally(unified.values(), throwable);
}
}, now);
}

for (Map.Entry<ConfigResource, KafkaFutureImpl<Config>> entry : brokerFutures.entrySet()) {
final KafkaFutureImpl<Config> brokerFuture = entry.getValue();
final ConfigResource resource = entry.getKey();
final int nodeId = Integer.parseInt(resource.name());
runnable.call(new Call("describeBrokerConfigs", calcDeadlineMs(now, options.timeoutMs()),
new ConstantNodeIdProvider(nodeId)) {

@Override
DescribeConfigsRequest.Builder createRequest(int timeoutMs) {
return new DescribeConfigsRequest.Builder(Collections.singleton(resource))
.includeSynonyms(options.includeSynonyms())
.includeDocumentation(options.includeDocumentation());
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
DescribeConfigsResponse response = (DescribeConfigsResponse) abstractResponse;
DescribeConfigsResponse.Config config = response.configs().get(resource);

if (config == null) {
brokerFuture.completeExceptionally(new UnknownServerException(
"Malformed broker response: missing config for " + resource));
return;
}
if (config.error().isFailure())
brokerFuture.completeExceptionally(config.error().exception());
else {
List<ConfigEntry> configEntries = new ArrayList<>();
for (DescribeConfigsResponse.ConfigEntry configEntry : config.entries()) {
configEntries.add(new ConfigEntry(configEntry.name(), configEntry.value(),
configSource(configEntry.source()), configEntry.isSensitive(), configEntry.isReadOnly(),
configSynonyms(configEntry), configType(configEntry.type()), configEntry.documentation()));
Comment thread
tombentley marked this conversation as resolved.
}
brokerFuture.complete(new Config(configEntries));
}
}

@Override
void handleFailure(Throwable throwable) {
brokerFuture.completeExceptionally(throwable);
}
}, now);
}
final Map<ConfigResource, KafkaFuture<Config>> allFutures = new HashMap<>();
allFutures.putAll(brokerFutures);
allFutures.putAll(unifiedRequestFutures);
return new DescribeConfigsResult(allFutures);
return new DescribeConfigsResult(new HashMap<>(brokerFutures.entrySet().stream()
.flatMap(x -> x.getValue().entrySet().stream())
.collect(Collectors.toMap(Entry::getKey, Entry::getValue))));
}

private List<ConfigEntry.ConfigSynonym> configSynonyms(DescribeConfigsResponse.ConfigEntry configEntry) {
List<ConfigEntry.ConfigSynonym> synonyms = new ArrayList<>(configEntry.synonyms().size());
for (DescribeConfigsResponse.ConfigSynonym synonym : configEntry.synonyms()) {
synonyms.add(new ConfigEntry.ConfigSynonym(synonym.name(), synonym.value(), configSource(synonym.source())));
}
return synonyms;
private Config describeConfigResult(DescribeConfigsResponseData.DescribeConfigsResult describeConfigsResult) {
Comment thread
tombentley marked this conversation as resolved.
Outdated
return new Config(describeConfigsResult.configs().stream().map(config -> new ConfigEntry(
config.name(),
config.value(),
DescribeConfigsResponse.ConfigSource.forId(config.configSource()).source(),
config.isSensitive(),
config.readOnly(),
(config.synonyms().stream().map(synonym -> new ConfigEntry.ConfigSynonym(synonym.name(), synonym.value(),
DescribeConfigsResponse.ConfigSource.forId(synonym.source()).source()))).collect(Collectors.toList()),
DescribeConfigsResponse.ConfigType.forId(config.configType()).type(),
config.documentation()
)).collect(Collectors.toList()));
}

private ConfigEntry.ConfigSource configSource(DescribeConfigsResponse.ConfigSource source) {
Expand Down Expand Up @@ -2118,8 +2086,9 @@ public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, fina
final Collection<ConfigResource> unifiedRequestResources = new ArrayList<>();

for (ConfigResource resource : configs.keySet()) {
if (dependsOnSpecificNode(resource)) {
NodeProvider nodeProvider = new ConstantNodeIdProvider(Integer.parseInt(resource.name()));
Integer node = nodeFor(resource);
if (node != null) {
NodeProvider nodeProvider = new ConstantNodeIdProvider(node);
allFutures.putAll(alterConfigs(configs, options, Collections.singleton(resource), nodeProvider));
} else
unifiedRequestResources.add(resource);
Expand Down Expand Up @@ -2183,8 +2152,9 @@ public AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection
final Collection<ConfigResource> unifiedRequestResources = new ArrayList<>();

for (ConfigResource resource : configs.keySet()) {
if (dependsOnSpecificNode(resource)) {
NodeProvider nodeProvider = new ConstantNodeIdProvider(Integer.parseInt(resource.name()));
Integer node = nodeFor(resource);
if (node != null) {
NodeProvider nodeProvider = new ConstantNodeIdProvider(node);
allFutures.putAll(incrementalAlterConfigs(configs, options, Collections.singleton(resource), nodeProvider));
} else
unifiedRequestResources.add(resource);
Expand Down Expand Up @@ -3685,11 +3655,16 @@ private void handleNotControllerError(Errors error) throws ApiException {
}

/**
* Returns a boolean indicating whether the resource needs to go to a specific node
* Returns the broker id pertaining to the given resource, or null if the resource is not associated
* with a particular broker.
*/
private boolean dependsOnSpecificNode(ConfigResource resource) {
return (resource.type() == ConfigResource.Type.BROKER && !resource.isDefault())
|| resource.type() == ConfigResource.Type.BROKER_LOGGER;
private Integer nodeFor(ConfigResource resource) {
if ((resource.type() == ConfigResource.Type.BROKER && !resource.isDefault())
|| resource.type() == ConfigResource.Type.BROKER_LOGGER) {
return Integer.valueOf(resource.name());
} else {
return null;
}
}

private List<MemberIdentity> getMembersFromGroup(String groupId) {
Expand Down
8 changes: 4 additions & 4 deletions 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 @@ -53,6 +53,8 @@
import org.apache.kafka.common.message.DescribeAclsResponseData;
import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
import org.apache.kafka.common.message.DescribeConfigsRequestData;
import org.apache.kafka.common.message.DescribeConfigsResponseData;
import org.apache.kafka.common.message.DescribeDelegationTokenRequestData;
import org.apache.kafka.common.message.DescribeDelegationTokenResponseData;
import org.apache.kafka.common.message.DescribeGroupsRequestData;
Expand Down Expand Up @@ -112,8 +114,6 @@
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.protocol.types.Type;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.DescribeConfigsRequest;
import org.apache.kafka.common.requests.DescribeConfigsResponse;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.common.requests.FetchResponse;
import org.apache.kafka.common.requests.ListOffsetRequest;
Expand Down Expand Up @@ -184,8 +184,8 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
DESCRIBE_ACLS(29, "DescribeAcls", DescribeAclsRequestData.SCHEMAS, DescribeAclsResponseData.SCHEMAS),
CREATE_ACLS(30, "CreateAcls", CreateAclsRequestData.SCHEMAS, CreateAclsResponseData.SCHEMAS),
DELETE_ACLS(31, "DeleteAcls", DeleteAclsRequestData.SCHEMAS, DeleteAclsResponseData.SCHEMAS),
DESCRIBE_CONFIGS(32, "DescribeConfigs", DescribeConfigsRequest.schemaVersions(),
DescribeConfigsResponse.schemaVersions()),
DESCRIBE_CONFIGS(32, "DescribeConfigs", DescribeConfigsRequestData.SCHEMAS,
DescribeConfigsResponseData.SCHEMAS),
ALTER_CONFIGS(33, "AlterConfigs", AlterConfigsRequestData.SCHEMAS,
AlterConfigsResponseData.SCHEMAS),
ALTER_REPLICA_LOG_DIRS(34, "AlterReplicaLogDirs", AlterReplicaLogDirsRequestData.SCHEMAS,
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct, shor
case DELETE_ACLS:
return new DeleteAclsResponse(struct, version);
case DESCRIBE_CONFIGS:
return new DescribeConfigsResponse(struct);
return new DescribeConfigsResponse(struct, version);
case ALTER_CONFIGS:
return new AlterConfigsResponse(struct, version);
case ALTER_REPLICA_LOG_DIRS:
Expand Down
Loading