Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1283: Unique client.id for each producer

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/1283

Avoid `InstanceAlreadyExistsException` s when the user supplies a custom
`client.id`.

**cherry-pick to all supported branches**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
#	spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
This commit is contained in:
Gary Russell
2019-10-28 13:34:51 -04:00
committed by Artem Bilan
parent 8c22b78225
commit 36fdb2d31f
2 changed files with 22 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -93,6 +93,8 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
private final Map<String, CloseSafeProducer<K, V>> consumerProducers = new HashMap<>();
private final AtomicInteger clientIdCounter = new AtomicInteger();
private volatile CloseSafeProducer<K, V> producer;
private Serializer<K> keySerializer;
@@ -107,6 +109,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
private boolean producerPerConsumerPartition = true;
private String clientIdPrefix;
/**
* Construct a factory with the provided configuration.
* @param configs the configuration.
@@ -117,9 +120,13 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
public DefaultKafkaProducerFactory(Map<String, Object> configs, Serializer<K> keySerializer,
Serializer<V> valueSerializer) {
this.configs = new HashMap<>(configs);
this.keySerializer = keySerializer;
this.valueSerializer = valueSerializer;
if (configs.get(ProducerConfig.CLIENT_ID_CONFIG) instanceof String) {
this.clientIdPrefix = (String) configs.get(ProducerConfig.CLIENT_ID_CONFIG);
}
}
public void setKeySerializer(Serializer<K> keySerializer) {
@@ -274,7 +281,15 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
* @return the producer.
*/
protected Producer<K, V> createKafkaProducer() {
return new KafkaProducer<K, V>(this.configs, this.keySerializer, this.valueSerializer);
if (this.clientIdPrefix == null) {
return new KafkaProducer<K, V>(this.configs, this.keySerializer, this.valueSerializer);
}
else {
Map<String, Object> newConfigs = new HashMap<>(this.configs);
newConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
return new KafkaProducer<>(newConfigs, this.keySerializer, this.valueSerializer);
}
}
Producer<K, V> createTransactionalProducerForPartition() {
@@ -328,6 +343,10 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
Producer<K, V> newProducer;
Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix);
if (this.clientIdPrefix != null) {
newProducerConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
}
newProducer = new KafkaProducer<K, V>(newProducerConfigs, this.keySerializer, this.valueSerializer);
newProducer.initTransactions();
return new CloseSafeProducer<K, V>(newProducer, this.cache, remover,

1
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java
View File

@@ -99,6 +99,7 @@ public class KafkaTemplateTransactionTests {
public void testLocalTransaction() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put(ProducerConfig.RETRIES_CONFIG, 1);
senderProps.put(ProducerConfig.CLIENT_ID_CONFIG, "customClientId");
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
pf.setKeySerializer(new StringSerializer());
pf.setTransactionIdPrefix("my.transaction.");