root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 5 Packages Projects Releases Wiki Activity

KAFKA-18079 consumer-config does not work with console-share-consumer (#17925) 

Reviewers: Andrew Schofield <aschofield@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
PoAn Yang 2024-11-24 20:13:14 +08:00 committed by GitHub
parent ccbb73111c
commit 70babd5716
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 8 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptions.java
View File

@ -182,7 +182,8 @@ public final class ConsoleShareConsumerOptions extends CommandDefaultOptions {
}
private Properties buildConsumerProps(Properties consumerPropsFromFile, Properties extraConsumerProps, Set<String> groupIdsProvided) {
Properties consumerProps = new Properties(consumerPropsFromFile);
Properties consumerProps = new Properties();
consumerProps.putAll(consumerPropsFromFile);
consumerProps.putAll(extraConsumerProps);
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer());
if (consumerProps.getProperty(ConsumerConfig.CLIENT_ID_CONFIG) == null) {

8
tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
View File

@ -19,6 +19,7 @@ package org.apache.kafka.tools.consumer;
import org.apache.kafka.clients.consumer.AcknowledgeType;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.tools.ToolsTestUtils;
import org.junit.jupiter.api.Test;
@ -101,8 +102,11 @@ public class ConsoleShareConsumerOptionsTest {
ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
assertEquals("1000", config.consumerProps().getProperty("request.timeout.ms"));
assertEquals("group1", config.consumerProps().getProperty("group.id"));
// KafkaShareConsumer uses Utils.propsToMap to convert the properties to a map,
// so using the same method to check the map has the expected values
Map<String, Object> configMap = Utils.propsToMap(config.consumerProps());
assertEquals("1000", configMap.get("request.timeout.ms"));
assertEquals("group1", configMap.get("group.id"));
}
@Test