Skip to content

KAFKA-17821: the set of configs displayed by logAll could be invalid #17993

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

Open
wants to merge 4 commits into
base: trunk
Choose a base branch
from
Open

KAFKA-17821: the set of configs displayed by logAll could be invalid #17993

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
8e05dfd
draft a version to remove logging log
m1a2st Nov 30, 2024
d5af8f9
addressed comments and add test
m1a2st Dec 14, 2024
8ddd273
revert unused change
m1a2st Dec 14, 2024
b61b9bd
Merge branch 'trunk' into KAFKA-17821
m1a2st Mar 16, 2025
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
25 changes: 24 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,8 @@ public class ConsumerConfig extends AbstractConfig {
/**
* A list of configuration keys not supported for CONSUMER protocol.
*/
private static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
// visible for testing
static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
HEARTBEAT_INTERVAL_MS_CONFIG,
SESSION_TIMEOUT_MS_CONFIG,
Expand Down Expand Up @@ -782,6 +783,28 @@ private void checkUnsupportedConfigsPostProcess(GroupProtocol groupProtocol, Lis
}
}

@Override
protected Map<String, Object> clearUnsupportedConfigsForLogging(Map<String, Object> values) {
String groupProtocol = (String) values.get(GROUP_PROTOCOL_CONFIG);
if (groupProtocol != null) {
if (GroupProtocol.CLASSIC.name().equalsIgnoreCase(groupProtocol)) {
values = clearUnsupportedConfigsForLogging(values, CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS);
} else if (GroupProtocol.CONSUMER.name().equalsIgnoreCase(groupProtocol)) {
values = clearUnsupportedConfigsForLogging(values, CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS);
}
}
return values;
}

private Map<String, Object> clearUnsupportedConfigsForLogging(
Map<String, Object> values,
List<String> unSupportConfig
) {
Map<String, Object> loggingLog = new HashMap<>(values);
unSupportConfig.forEach(loggingLog::remove);
return loggingLog;
}

public ConsumerConfig(Properties props) {
super(CONFIG, props);
}
Expand Down
10 changes: 9 additions & 1 deletion 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 @@ -366,13 +366,21 @@ public Map<String, Object> valuesWithPrefixAllOrNothing(String prefix) {
return nonInternalConfigs;
}

/**
* Won't do any filter in the abstract config, but can be overridden in subclasses.
*/
protected Map<String, Object> clearUnsupportedConfigsForLogging(Map<String, Object> values) {
return new TreeMap<>(values);
}

private void logAll() {
Map<String, Object> valuesToLog = clearUnsupportedConfigsForLogging(this.values);
StringBuilder b = new StringBuilder();
b.append(getClass().getSimpleName());
b.append(" values: ");
b.append(Utils.NL);

Copy link
Contributor

Choose a reason for hiding this comment

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

That way we can invoke this from inside logAll(), keeping the logging bits closer together.

Suggested change
Map<String, Object> valuesToLog - clearUnsupportedConfigsForLogging(values);

for (Map.Entry<String, Object> entry : new TreeMap<>(this.values).entrySet()) {
for (Map.Entry<String, Object> entry : valuesToLog.entrySet()) {
b.append('\t');
b.append(entry.getKey());
b.append(" = ");
Expand Down
18 changes: 18 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,24 @@ public void testUnsupportedConfigsWithConsumerGroupProtocol() {
testUnsupportedConfigsWithConsumerGroupProtocol(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
testUnsupportedConfigsWithConsumerGroupProtocol(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
}

@Test
public void testUnsupportedConfigsForLoggingWithConsumerGroupProtocol() {
final Map<String, Object> configs = Map.of(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass,
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass,
ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name()
);
ConsumerConfig config = new ConsumerConfig(configs);
Map<String, Object> configToLog = config.clearUnsupportedConfigsForLogging(
configs.values()
.stream()
.collect(HashMap::new, (m, v) -> m.put(v.toString(), v), HashMap::putAll)
);
ConsumerConfig.CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS.forEach(
configName -> assertFalse(configToLog.containsKey(configName))
);
}

private void testUnsupportedConfigsWithConsumerGroupProtocol(String configName, Object value) {
final Map<String, Object> configs = Map.of(
Expand Down