Skip to content

KIP-569-KAFKA-9494: DescribeConfigsResponse - include additional metadata information #8723

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 1 commit into from
May 29, 2020
Merged

KIP-569-KAFKA-9494: DescribeConfigsResponse - include additional metadata information #8723

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
39 changes: 37 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@ public class ConfigEntry {
private final boolean isSensitive;
private final boolean isReadOnly;
private final List<ConfigSynonym> synonyms;
private final ConfigType type;
private final String documentation;

/**
* Create a configuration entry with the provided values.
Expand Down Expand Up @@ -65,7 +67,9 @@ public ConfigEntry(String name, String value, boolean isDefault, boolean isSensi
isDefault ? ConfigSource.DEFAULT_CONFIG : ConfigSource.UNKNOWN,
isSensitive,
isReadOnly,
Collections.<ConfigSynonym>emptyList());
Collections.<ConfigSynonym>emptyList(),
ConfigType.UNKNOWN,
null);
}

/**
Expand All @@ -79,14 +83,16 @@ public ConfigEntry(String name, String value, boolean isDefault, boolean isSensi
* @param synonyms Synonym configs in order of precedence
*/
ConfigEntry(String name, String value, ConfigSource source, boolean isSensitive, boolean isReadOnly,
List<ConfigSynonym> synonyms) {
List<ConfigSynonym> synonyms, ConfigType type, String documentation) {
Objects.requireNonNull(name, "name should not be null");
this.name = name;
this.value = value;
this.source = source;
this.isSensitive = isSensitive;
this.isReadOnly = isReadOnly;
this.synonyms = synonyms;
this.type = type;
this.documentation = documentation;
}

/**
Expand Down Expand Up @@ -141,6 +147,20 @@ public List<ConfigSynonym> synonyms() {
return synonyms;
}

/**
* Return the config data type.
*/
public ConfigType type() {
return type;
}

/**
* Return the config documentation.
*/
public String documentation() {
return documentation;
}

@Override
public boolean equals(Object o) {
if (this == o)
Expand Down Expand Up @@ -183,6 +203,21 @@ public String toString() {
")";
}

/**
* Data type of configuration entry.
*/
public enum ConfigType {
UNKNOWN,
BOOLEAN,
STRING,
INT,
SHORT,
LONG,
DOUBLE,
LIST,
CLASS,
PASSWORD
}

/**
* Source of configuration entries.
Expand Down
15 changes: 15 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/admin/DescribeConfigsOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
public class DescribeConfigsOptions extends AbstractOptions<DescribeConfigsOptions> {

private boolean includeSynonyms = false;
private boolean includeDocumentation = false;

/**
* Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the
Expand All @@ -49,6 +50,13 @@ public boolean includeSynonyms() {
return includeSynonyms;
}

/**
* Return true if config documentation should be returned in the response.
*/
public boolean includeDocumentation() {
return includeDocumentation;
}

/**
* Set to true if synonym configs should be returned in the response.
*/
Expand All @@ -57,4 +65,11 @@ public DescribeConfigsOptions includeSynonyms(boolean includeSynonyms) {
return this;
}

/**
* Set to true if config documentation should be returned in the response.
*/
public DescribeConfigsOptions includeDocumentation(boolean includeDocumentation) {
this.includeDocumentation = includeDocumentation;
return this;
}
}
55 changes: 50 additions & 5 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 @@ -1468,7 +1468,9 @@ public void handleResponse(AbstractResponse abstractResponse) {
configSource(DescribeConfigsResponse.ConfigSource.forId(config.configSource())),
config.isSensitive(),
config.readOnly(),
Collections.emptyList()))
Collections.emptyList(),
null,
null))
.collect(Collectors.toSet()));
topicMetadataAndConfig = new TopicMetadataAndConfig(result.numPartitions(),
result.replicationFactor(),
Expand Down Expand Up @@ -1932,7 +1934,8 @@ public DescribeConfigsResult describeConfigs(Collection<ConfigResource> configRe
@Override
DescribeConfigsRequest.Builder createRequest(int timeoutMs) {
return new DescribeConfigsRequest.Builder(unifiedRequestResources)
.includeSynonyms(options.includeSynonyms());
.includeSynonyms(options.includeSynonyms())
.includeDocumentation(options.includeDocumentation());
}

@Override
Expand All @@ -1956,7 +1959,8 @@ void handleResponse(AbstractResponse abstractResponse) {
configEntries.add(new ConfigEntry(configEntry.name(),
configEntry.value(), configSource(configEntry.source()),
configEntry.isSensitive(), configEntry.isReadOnly(),
configSynonyms(configEntry)));
configSynonyms(configEntry), configType(configEntry.type()),
configEntry.documentation()));
}
future.complete(new Config(configEntries));
}
Expand All @@ -1979,7 +1983,8 @@ void handleFailure(Throwable throwable) {
@Override
DescribeConfigsRequest.Builder createRequest(int timeoutMs) {
return new DescribeConfigsRequest.Builder(Collections.singleton(resource))
.includeSynonyms(options.includeSynonyms());
.includeSynonyms(options.includeSynonyms())
.includeDocumentation(options.includeDocumentation());
}

@Override
Expand All @@ -1999,7 +2004,7 @@ void handleResponse(AbstractResponse abstractResponse) {
for (DescribeConfigsResponse.ConfigEntry configEntry : config.entries()) {
configEntries.add(new ConfigEntry(configEntry.name(), configEntry.value(),
configSource(configEntry.source()), configEntry.isSensitive(), configEntry.isReadOnly(),
configSynonyms(configEntry)));
configSynonyms(configEntry), configType(configEntry.type()), configEntry.documentation()));
}
brokerFuture.complete(new Config(configEntries));
}
Expand Down Expand Up @@ -2052,6 +2057,46 @@ private ConfigEntry.ConfigSource configSource(DescribeConfigsResponse.ConfigSour
return configSource;
}

private ConfigEntry.ConfigType configType(DescribeConfigsResponse.ConfigType type) {
if (type == null) {
return ConfigEntry.ConfigType.UNKNOWN;
}

ConfigEntry.ConfigType configType;
switch (type) {
case BOOLEAN:
configType = ConfigEntry.ConfigType.BOOLEAN;
break;
case CLASS:
configType = ConfigEntry.ConfigType.CLASS;
break;
case DOUBLE:
configType = ConfigEntry.ConfigType.DOUBLE;
break;
case INT:
configType = ConfigEntry.ConfigType.INT;
break;
case LIST:
configType = ConfigEntry.ConfigType.LIST;
break;
case LONG:
configType = ConfigEntry.ConfigType.LONG;
break;
case PASSWORD:
configType = ConfigEntry.ConfigType.PASSWORD;
break;
case SHORT:
configType = ConfigEntry.ConfigType.SHORT;
break;
case STRING:
configType = ConfigEntry.ConfigType.STRING;
break;
default:
configType = ConfigEntry.ConfigType.UNKNOWN;
}
return configType;
}

@Override
@Deprecated
public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options) {
Expand Down
7 changes: 7 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,13 @@ public ConfigDef.Type typeOf(String key) {
return configKey.type;
}

public String documentationOf(String key) {
Comment thread
srpanwar-confluent marked this conversation as resolved.
Outdated
ConfigDef.ConfigKey configKey = definition.configKeys().get(key);
if (configKey == null)
return null;
return configKey.documentation;
}

public Password getPassword(String key) {
return (Password) get(key);
}
Expand Down
39 changes: 36 additions & 3 deletions clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ public class DescribeConfigsRequest extends AbstractRequest {
private static final String RESOURCE_TYPE_KEY_NAME = "resource_type";
private static final String RESOURCE_NAME_KEY_NAME = "resource_name";
private static final String CONFIG_NAMES_KEY_NAME = "config_names";
private static final String INCLUDE_DOCUMENTATION = "include_documentation";

@rajinisivaram rajinisivaram May 28, 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.

Can we switch over to using the generated DescribeConfigsRequestData class?


private static final Schema DESCRIBE_CONFIGS_REQUEST_RESOURCE_V0 = new Schema(
new Field(RESOURCE_TYPE_KEY_NAME, INT8),
Expand All @@ -61,13 +62,24 @@ public class DescribeConfigsRequest extends AbstractRequest {
*/
private static final Schema DESCRIBE_CONFIGS_REQUEST_V2 = DESCRIBE_CONFIGS_REQUEST_V1;

private static final Schema DESCRIBE_CONFIGS_REQUEST_V3 = new Schema(
new Field(RESOURCES_KEY_NAME, new ArrayOf(DESCRIBE_CONFIGS_REQUEST_RESOURCE_V0), "An array of config resources to be returned."),
new Field(INCLUDE_SYNONYMS, BOOLEAN),
new Field(INCLUDE_DOCUMENTATION, BOOLEAN));

public static Schema[] schemaVersions() {
return new Schema[]{DESCRIBE_CONFIGS_REQUEST_V0, DESCRIBE_CONFIGS_REQUEST_V1, DESCRIBE_CONFIGS_REQUEST_V2};
return new Schema[] {
DESCRIBE_CONFIGS_REQUEST_V0,
DESCRIBE_CONFIGS_REQUEST_V1,
DESCRIBE_CONFIGS_REQUEST_V2,
DESCRIBE_CONFIGS_REQUEST_V3
};
}

public static class Builder extends AbstractRequest.Builder<DescribeConfigsRequest> {
private final Map<ConfigResource, Collection<String>> resourceToConfigNames;
private boolean includeSynonyms;
private boolean includeDocumentation;

public Builder(Map<ConfigResource, Collection<String>> resourceToConfigNames) {
super(ApiKeys.DESCRIBE_CONFIGS);
Expand All @@ -79,6 +91,11 @@ public Builder includeSynonyms(boolean includeSynonyms) {
return this;
}

public Builder includeDocumentation(boolean includeDocumentation) {
this.includeDocumentation = includeDocumentation;
return this;
}

public Builder(Collection<ConfigResource> resources) {
this(toResourceToConfigNames(resources));
}
Expand All @@ -92,17 +109,27 @@ private static Map<ConfigResource, Collection<String>> toResourceToConfigNames(C

@Override
public DescribeConfigsRequest build(short version) {
return new DescribeConfigsRequest(version, resourceToConfigNames, includeSynonyms);
return new DescribeConfigsRequest(
version, resourceToConfigNames, includeSynonyms, includeDocumentation);
}
}

private final Map<ConfigResource, Collection<String>> resourceToConfigNames;
private final boolean includeSynonyms;
private final boolean includeDocumentation;

public DescribeConfigsRequest(short version, Map<ConfigResource, Collection<String>> resourceToConfigNames, boolean includeSynonyms) {
public DescribeConfigsRequest(
short version, Map<ConfigResource, Collection<String>> resourceToConfigNames,
boolean includeSynonyms) {
this(version, resourceToConfigNames, includeSynonyms, false);
}
public DescribeConfigsRequest(
short version, Map<ConfigResource, Collection<String>> resourceToConfigNames,
boolean includeSynonyms, boolean includeDocumentation) {
super(ApiKeys.DESCRIBE_CONFIGS, version);
this.resourceToConfigNames = Objects.requireNonNull(resourceToConfigNames, "resourceToConfigNames");
this.includeSynonyms = includeSynonyms;
this.includeDocumentation = includeDocumentation;
}

public DescribeConfigsRequest(Struct struct, short version) {
Expand All @@ -125,6 +152,7 @@ public DescribeConfigsRequest(Struct struct, short version) {
resourceToConfigNames.put(new ConfigResource(resourceType, resourceName), configNames);
}
this.includeSynonyms = struct.hasField(INCLUDE_SYNONYMS) ? struct.getBoolean(INCLUDE_SYNONYMS) : false;
this.includeDocumentation = struct.hasField(INCLUDE_DOCUMENTATION) ? struct.getBoolean(INCLUDE_DOCUMENTATION) : false;
}

public Collection<ConfigResource> resources() {
Expand All @@ -142,6 +170,10 @@ public boolean includeSynonyms() {
return includeSynonyms;
}

public boolean includeDocumentation() {
return includeDocumentation;
}

@Override
protected Struct toStruct() {
Struct struct = new Struct(ApiKeys.DESCRIBE_CONFIGS.requestSchema(version()));
Expand All @@ -159,6 +191,7 @@ protected Struct toStruct() {
}
struct.set(RESOURCES_KEY_NAME, resourceStructs.toArray(new Struct[0]));
struct.setIfExists(INCLUDE_SYNONYMS, includeSynonyms);
struct.setIfExists(INCLUDE_DOCUMENTATION, includeDocumentation);
return struct;
}

Expand Down
Loading