Skip to content

Simplify KafkaConsumer constructor logic #8707

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

Closed
nresare wants to merge 1 commit into from
Closed

Simplify KafkaConsumer constructor logic #8707

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
12 changes: 0 additions & 12 deletions 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 @@ -573,18 +573,6 @@ public static Map<String, Object> addDeserializerToConfig(Map<String, Object> co
return newConfigs;
}

public static Properties addDeserializerToConfig(Properties properties,
Deserializer<?> keyDeserializer,
Deserializer<?> valueDeserializer) {
Properties newProperties = new Properties();
newProperties.putAll(properties);
if (keyDeserializer != null)
newProperties.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
if (valueDeserializer != null)
newProperties.put(VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
return newProperties;
}

boolean maybeOverrideEnableAutoCommit() {
Optional<String> groupId = Optional.ofNullable(getString(CommonClientConfigs.GROUP_ID_CONFIG));
boolean enableAutoCommit = getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
Expand Down
45 changes: 20 additions & 25 deletions clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@
import java.util.regex.Pattern;

import static org.apache.kafka.clients.consumer.internals.PartitionAssignorAdapter.getAssignorInstances;
import static org.apache.kafka.common.utils.Utils.propsToMap;

/**
* A client that consumes records from a Kafka cluster.
Expand Down Expand Up @@ -611,27 +612,6 @@ public KafkaConsumer(Map<String, Object> configs) {
this(configs, null, null);
}

/**
* A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value {@link Deserializer}.
* <p>
* Valid configuration strings are documented at {@link ConsumerConfig}.
* <p>
* Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
*
* @param configs The consumer configs
* @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
* won't be called in the consumer when the deserializer is passed in directly.
* @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
* won't be called in the consumer when the deserializer is passed in directly.
*/
public KafkaConsumer(Map<String, Object> configs,
Deserializer<K> keyDeserializer,
Deserializer<V> valueDeserializer) {
this(new ConsumerConfig(ConsumerConfig.addDeserializerToConfig(configs, keyDeserializer, valueDeserializer)),
keyDeserializer,
valueDeserializer);
}

/**
* A consumer is instantiated by providing a {@link java.util.Properties} object as configuration.
* <p>
Expand All @@ -642,7 +622,7 @@ public KafkaConsumer(Map<String, Object> configs,
* @param properties The consumer configuration properties
*/
public KafkaConsumer(Properties properties) {
this(properties, null, null);
this(propsToMap(properties), null, null);
}

/**
Expand All @@ -662,12 +642,27 @@ public KafkaConsumer(Properties properties) {
public KafkaConsumer(Properties properties,
Deserializer<K> keyDeserializer,
Deserializer<V> valueDeserializer) {
this(new ConsumerConfig(ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer, valueDeserializer)),
keyDeserializer, valueDeserializer);
this(propsToMap(properties), keyDeserializer, valueDeserializer);
}

/**
* A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value {@link Deserializer}.
* <p>
* Valid configuration strings are documented at {@link ConsumerConfig}.
* <p>
* Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
*
* @param configs The consumer configs
* @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
* won't be called in the consumer when the deserializer is passed in directly.
* @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
* won't be called in the consumer when the deserializer is passed in directly.
*/
@SuppressWarnings("unchecked")
private KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
public KafkaConsumer(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
ConsumerConfig config = new ConsumerConfig(ConsumerConfig.addDeserializerToConfig(configs, keyDeserializer,
valueDeserializer));

try {
GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(config,
GroupRebalanceConfig.ProtocolType.CONSUMER);
Expand Down
15 changes: 2 additions & 13 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand All @@ -85,6 +84,8 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import static org.apache.kafka.common.utils.Utils.propsToMap;


/**
* A Kafka client that publishes records to the Kafka cluster.
Expand Down Expand Up @@ -1235,18 +1236,6 @@ private void close(Duration timeout, boolean swallowException) {
log.debug("Kafka producer has been closed");
}

private static Map<String, Object> propsToMap(Properties properties) {
Map<String, Object> map = new HashMap<>(properties.size());
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
if (entry.getKey() instanceof String) {
String k = (String) entry.getKey();
map.put(k, properties.get(k));
} else {
throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
}
}
return map;
}

private ClusterResourceListeners configureClusterResourceListeners(Serializer<K> keySerializer, Serializer<V> valueSerializer, List<?>... candidateLists) {
ClusterResourceListeners clusterResourceListeners = new ClusterResourceListeners();
Expand Down
14 changes: 14 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.util.SortedSet;
import java.util.TreeSet;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.ConfigException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -740,6 +741,19 @@ public static <K, V> Map<K, V> mkMap(final Map.Entry<K, V>... entries) {
return result;
}

public static Map<String, Object> propsToMap(Properties properties) {
Map<String, Object> map = new HashMap<>(properties.size());
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
if (entry.getKey() instanceof String) {
String k = (String) entry.getKey();
map.put(k, properties.get(k));
} else {
throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
}
}
return map;
}

/**
* Creates a {@link Properties} from a map
*
Expand Down
27 changes: 0 additions & 27 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 @@ -68,33 +68,6 @@ public void testOverrideEnableAutoCommit() {
}
}

@Test
public void testDeserializerToPropertyConfig() {
Properties properties = new Properties();
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClassName);
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClassName);
Properties newProperties = ConsumerConfig.addDeserializerToConfig(properties, null, null);
assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);

properties.clear();
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClassName);
newProperties = ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer, null);
assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);

properties.clear();
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClassName);
newProperties = ConsumerConfig.addDeserializerToConfig(properties, null, valueDeserializer);
assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);

properties.clear();
newProperties = ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer, valueDeserializer);
assertEquals(newProperties.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), keyDeserializerClassName);
assertEquals(newProperties.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), valueDeserializerClassName);
}

@Test
public void testDeserializerToMapConfig() {
Map<String, Object> configs = new HashMap<>();
Expand Down