Class DefaultKafkaProducerFactory<K,V>
- java.lang.Object
-
- org.springframework.kafka.core.KafkaResourceFactory
-
- org.springframework.kafka.core.DefaultKafkaProducerFactory<K,V>
-
- Type Parameters:
K
- the key type.V
- the value type.
- All Implemented Interfaces:
java.util.EventListener
,org.springframework.beans.factory.Aware
,org.springframework.beans.factory.BeanNameAware
,org.springframework.beans.factory.DisposableBean
,org.springframework.context.ApplicationContextAware
,org.springframework.context.ApplicationListener<org.springframework.context.event.ContextStoppedEvent>
,ProducerFactory<K,V>
public class DefaultKafkaProducerFactory<K,V> extends KafkaResourceFactory implements ProducerFactory<K,V>, org.springframework.context.ApplicationContextAware, org.springframework.beans.factory.BeanNameAware, org.springframework.context.ApplicationListener<org.springframework.context.event.ContextStoppedEvent>, org.springframework.beans.factory.DisposableBean
TheProducerFactory
implementation for asingleton
sharedProducer
instance.This implementation will return the same
Producer
instance (if transactions are not enabled) for the providedMap
configs
and optionalSerializer
implementations on eachcreateProducer()
invocation.If you are using
Serializer
s that have no-arg constructors and require no setup, then simplest to specifySerializer
classes againstProducerConfig.KEY_SERIALIZER_CLASS_CONFIG
andProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG
keys in theconfigs
passed to theDefaultKafkaProducerFactory
constructor.If that is not possible, but you are sure that at least one of the following is true:
- only one
Producer
will use theSerializer
s - you are using
Serializer
s that may be shared betweenProducer
instances (and specifically that their close() method is a no-op) - you are certain that there is no risk of any single
Producer
being closed while otherProducer
instances with the sameSerializer
s are in use
Serializer
instances for one or both of the key and value serializers.If none of the above is true then you may provide a
Supplier
function for one or bothSerializer
s which will be used to obtainSerializer
(s) each time aProducer
is created by the factory.The
Producer
is wrapped and the underlyingKafkaProducer
instance is not actually closed whenProducer.close()
is invoked. TheKafkaProducer
is physically closed whenDisposableBean.destroy()
is invoked or when the application context publishes aContextStoppedEvent
. You can also invokereset()
.Setting
setTransactionIdPrefix(String)
enables transactions; in which case, a cache of producers is maintained; closing a producer returns it to the cache. The producers are closed and the cache is cleared when the factory is destroyed, the application context stopped, or thereset()
method is called.- Author:
- Gary Russell, Murali Reddy, Nakul Mishra, Artem Bilan, Chris Gilbert, Thomas Strauß
-
-
Nested Class Summary
Nested Classes Modifier and Type Class Description protected static class
DefaultKafkaProducerFactory.CloseSafeProducer<K,V>
A wrapper class for the delegate.-
Nested classes/interfaces inherited from interface org.springframework.kafka.core.ProducerFactory
ProducerFactory.Listener<K,V>
-
-
Field Summary
-
Fields inherited from interface org.springframework.kafka.core.ProducerFactory
DEFAULT_PHYSICAL_CLOSE_TIMEOUT
-
-
Constructor Summary
Constructors Constructor Description DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs)
Construct a factory with the provided configuration.DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier, java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier)
Construct a factory with the provided configuration andSerializer
Suppliers.DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier, java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier, boolean configureSerializers)
Construct a factory with the provided configuration andSerializer
Suppliers.DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, org.apache.kafka.common.serialization.Serializer<K> keySerializer, org.apache.kafka.common.serialization.Serializer<V> valueSerializer)
Construct a factory with the provided configuration andSerializer
s.DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, org.apache.kafka.common.serialization.Serializer<K> keySerializer, org.apache.kafka.common.serialization.Serializer<V> valueSerializer, boolean configureSerializers)
Construct a factory with the provided configuration andSerializer
s.
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description void
addListener(int index, ProducerFactory.Listener<K,V> listener)
Add a listener at a specific index.void
addListener(ProducerFactory.Listener<K,V> listener)
Add a listener.void
addPostProcessor(ProducerPostProcessor<K,V> postProcessor)
Add a post processor.void
closeProducerFor(java.lang.String suffix)
Remove the specified producer from the cache and close it.void
closeThreadBoundProducer()
When usingsetProducerPerThread(boolean)
(true), call this method to close and release this thread's producer.ProducerFactory<K,V>
copyWithConfigurationOverride(java.util.Map<java.lang.String,java.lang.Object> overrideProperties)
Copy properties of the instance and the given properties to create a new producer factory.protected org.apache.kafka.clients.producer.Producer<K,V>
createKafkaProducer()
Subclasses must return a raw producer which will be wrapped in aDefaultKafkaProducerFactory.CloseSafeProducer
.org.apache.kafka.clients.producer.Producer<K,V>
createNonTransactionalProducer()
Create a non-transactional producer.org.apache.kafka.clients.producer.Producer<K,V>
createProducer()
Create a producer which will be transactional if the factory is so configured.org.apache.kafka.clients.producer.Producer<K,V>
createProducer(java.lang.String txIdPrefixArg)
Create a producer with an overridden transaction id prefix.protected org.apache.kafka.clients.producer.Producer<K,V>
createRawProducer(java.util.Map<java.lang.String,java.lang.Object> rawConfigs)
protected org.apache.kafka.clients.producer.Producer<K,V>
createTransactionalProducer()
Subclasses must return a producer from thegetCache()
or a new raw producer wrapped in aDefaultKafkaProducerFactory.CloseSafeProducer
.protected org.apache.kafka.clients.producer.Producer<K,V>
createTransactionalProducer(java.lang.String txIdPrefix)
protected org.apache.kafka.clients.producer.Producer<K,V>
createTransactionalProducerForPartition()
protected org.apache.kafka.clients.producer.Producer<K,V>
createTransactionalProducerForPartition(java.lang.String txIdPrefix)
void
destroy()
protected java.util.concurrent.BlockingQueue<DefaultKafkaProducerFactory.CloseSafeProducer<K,V>>
getCache()
protected java.util.concurrent.BlockingQueue<DefaultKafkaProducerFactory.CloseSafeProducer<K,V>>
getCache(java.lang.String txIdPrefix)
java.util.Map<java.lang.String,java.lang.Object>
getConfigurationProperties()
Return an unmodifiable reference to the configuration map for this factory.org.apache.kafka.common.serialization.Serializer<K>
getKeySerializer()
Return the configured key serializer (if provided as an object instead of a class name in the properties).java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>>
getKeySerializerSupplier()
Return a supplier for a key serializer.java.util.List<ProducerFactory.Listener<K,V>>
getListeners()
Get the current list of listeners.java.time.Duration
getPhysicalCloseTimeout()
Get the physical close timeout.java.util.List<ProducerPostProcessor<K,V>>
getPostProcessors()
Get the current list of post processors.protected java.util.Map<java.lang.String,java.lang.Object>
getProducerConfigs()
Return the configuration of a producer.java.lang.String
getTransactionIdPrefix()
Return the transaction id prefix.protected java.util.Map<java.lang.String,java.lang.Object>
getTxProducerConfigs(java.lang.String transactionId)
Return the configuration of a transactional producer.org.apache.kafka.common.serialization.Serializer<V>
getValueSerializer()
Return the configured value serializer (if provided as an object instead of a class name in the properties).java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>>
getValueSerializerSupplier()
Return a supplier for a value serializer.boolean
isConfigureSerializers()
If true (default), programmatically provided serializers (via constructor or setters) will be configured using the producer configuration.boolean
isProducerPerConsumerPartition()
Return the producerPerConsumerPartition.boolean
isProducerPerThread()
Return true when there is a producer per thread.void
onApplicationEvent(org.springframework.context.event.ContextStoppedEvent event)
void
removeConfig(java.lang.String configKey)
Remove the specified key from the configuration map.boolean
removeListener(ProducerFactory.Listener<K,V> listener)
Remove a listener.boolean
removePostProcessor(ProducerPostProcessor<K,V> postProcessor)
Remove a post processor.protected boolean
removeProducer(DefaultKafkaProducerFactory.CloseSafeProducer<K,V> producerToRemove, java.time.Duration timeout)
Remove the single shared producer and a thread-bound instance if present.void
reset()
Close theProducer
(s) and clear the cache of transactionalProducer
(s).void
setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
void
setBeanName(java.lang.String name)
void
setConfigureSerializers(boolean configureSerializers)
Set to false (default true) to prevent programmatically provided serializers (via constructor or setters) from being configured using the producer configuration, e.g.void
setKeySerializer(org.apache.kafka.common.serialization.Serializer<K> keySerializer)
Set a key serializer.void
setKeySerializerSupplier(java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier)
Set a supplier to supply instances of the key serializer.void
setMaxAge(java.time.Duration maxAge)
Set the maximum age for a producer; useful when using transactions and the broker might expire atransactional.id
due to inactivity.void
setPhysicalCloseTimeout(int physicalCloseTimeout)
The time to wait when physically closing the producer via the factory rather than closing the producer itself (whenreset()
,#closeProducerFor(String)
, orcloseThreadBoundProducer()
are invoked).void
setProducerPerConsumerPartition(boolean producerPerConsumerPartition)
Set to false to revert to the previous behavior of a simple incrementing transactional.id suffix for each producer instead of maintaining a producer for each group/topic/partition.void
setProducerPerThread(boolean producerPerThread)
Set to true to create a producer per thread instead of singleton that is shared by all clients.void
setTransactionIdPrefix(java.lang.String transactionIdPrefix)
Set a prefix for theProducerConfig.TRANSACTIONAL_ID_CONFIG
config.void
setValueSerializer(org.apache.kafka.common.serialization.Serializer<V> valueSerializer)
Set a value serializer.void
setValueSerializerSupplier(java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier)
Set a supplier to supply instances of the value serializer.boolean
transactionCapable()
Return true if the factory supports transactions.void
updateConfigs(java.util.Map<java.lang.String,java.lang.Object> updates)
Update the producer configuration map; useful for situations such as credential rotation.-
Methods inherited from class org.springframework.kafka.core.KafkaResourceFactory
checkBootstrap, getBootstrapServers, setBootstrapServersSupplier
-
-
-
-
Constructor Detail
-
DefaultKafkaProducerFactory
public DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs)
Construct a factory with the provided configuration.- Parameters:
configs
- the configuration.
-
DefaultKafkaProducerFactory
public DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, @Nullable org.apache.kafka.common.serialization.Serializer<K> keySerializer, @Nullable org.apache.kafka.common.serialization.Serializer<V> valueSerializer)
Construct a factory with the provided configuration andSerializer
s. Also configures atransactionIdPrefix
as a value from theProducerConfig.TRANSACTIONAL_ID_CONFIG
if provided. This config is going to be overridden with a suffix for targetProducer
instance. The serializers'configure()
methods will be called with the configuration map.- Parameters:
configs
- the configuration.keySerializer
- the keySerializer
.valueSerializer
- the valueSerializer
.
-
DefaultKafkaProducerFactory
public DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, @Nullable org.apache.kafka.common.serialization.Serializer<K> keySerializer, @Nullable org.apache.kafka.common.serialization.Serializer<V> valueSerializer, boolean configureSerializers)
Construct a factory with the provided configuration andSerializer
s. Also configures atransactionIdPrefix
as a value from theProducerConfig.TRANSACTIONAL_ID_CONFIG
if provided. This config is going to be overridden with a suffix for targetProducer
instance. The serializers'configure()
methods will be called with the configuration map unlessconfigureSerializers
is false..- Parameters:
configs
- the configuration.keySerializer
- the keySerializer
.valueSerializer
- the valueSerializer
.configureSerializers
- set to false if serializers are already fully configured.- Since:
- 2.8.7
-
DefaultKafkaProducerFactory
public DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, @Nullable java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier, @Nullable java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier)
Construct a factory with the provided configuration andSerializer
Suppliers. Also configures atransactionIdPrefix
as a value from theProducerConfig.TRANSACTIONAL_ID_CONFIG
if provided. This config is going to be overridden with a suffix for targetProducer
instance. When the suppliers are invoked to get an instance, the serializers'configure()
methods will be called with the configuration map.- Parameters:
configs
- the configuration.keySerializerSupplier
- the keySerializer
supplier function.valueSerializerSupplier
- the valueSerializer
supplier function.- Since:
- 2.3
-
DefaultKafkaProducerFactory
public DefaultKafkaProducerFactory(java.util.Map<java.lang.String,java.lang.Object> configs, @Nullable java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier, @Nullable java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier, boolean configureSerializers)
Construct a factory with the provided configuration andSerializer
Suppliers. Also configures atransactionIdPrefix
as a value from theProducerConfig.TRANSACTIONAL_ID_CONFIG
if provided. This config is going to be overridden with a suffix for targetProducer
instance. When the suppliers are invoked to get an instance, the serializers'configure()
methods will be called with the configuration map unlessconfigureSerializers
is false.- Parameters:
configs
- the configuration.keySerializerSupplier
- the keySerializer
supplier function.valueSerializerSupplier
- the valueSerializer
supplier function.configureSerializers
- set to false if serializers are already fully configured.- Since:
- 2.8.7
-
-
Method Detail
-
setApplicationContext
public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext) throws org.springframework.beans.BeansException
- Specified by:
setApplicationContext
in interfaceorg.springframework.context.ApplicationContextAware
- Throws:
org.springframework.beans.BeansException
-
setBeanName
public void setBeanName(java.lang.String name)
- Specified by:
setBeanName
in interfaceorg.springframework.beans.factory.BeanNameAware
-
setKeySerializer
public void setKeySerializer(@Nullable org.apache.kafka.common.serialization.Serializer<K> keySerializer)
Set a key serializer. The serializer will be configured using the producer configuration, unlessconfigureSerializers
is false.- Parameters:
keySerializer
- the key serializer.- See Also:
setConfigureSerializers(boolean)
-
setValueSerializer
public void setValueSerializer(@Nullable org.apache.kafka.common.serialization.Serializer<V> valueSerializer)
Set a value serializer. The serializer will be configured using the producer configuration, unlessconfigureSerializers
is false.- Parameters:
valueSerializer
- the value serializer.- See Also:
setConfigureSerializers(boolean)
-
setKeySerializerSupplier
public void setKeySerializerSupplier(java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> keySerializerSupplier)
Set a supplier to supply instances of the key serializer. The serializer will be configured using the producer configuration, unlessconfigureSerializers
is false.- Parameters:
keySerializerSupplier
- the supplier.- Since:
- 2.8
- See Also:
setConfigureSerializers(boolean)
-
setValueSerializerSupplier
public void setValueSerializerSupplier(java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> valueSerializerSupplier)
Set a supplier to supply instances of the value serializer.- Parameters:
valueSerializerSupplier
- the supplier. The serializer will be configured using the producer configuration, unlessconfigureSerializers
is false.- Since:
- 2.8
- See Also:
setConfigureSerializers(boolean)
-
isConfigureSerializers
public boolean isConfigureSerializers()
If true (default), programmatically provided serializers (via constructor or setters) will be configured using the producer configuration. Set to false if the serializers are already fully configured.- Returns:
- true to configure.
- Since:
- 2.8.7
- See Also:
setKeySerializer(Serializer)
,setKeySerializerSupplier(Supplier)
,setValueSerializer(Serializer)
,setValueSerializerSupplier(Supplier)
-
setConfigureSerializers
public void setConfigureSerializers(boolean configureSerializers)
Set to false (default true) to prevent programmatically provided serializers (via constructor or setters) from being configured using the producer configuration, e.g. if the serializers are already fully configured.- Parameters:
configureSerializers
- false to not configure.- Since:
- 2.8.7
- See Also:
setKeySerializer(Serializer)
,setKeySerializerSupplier(Supplier)
,setValueSerializer(Serializer)
,setValueSerializerSupplier(Supplier)
-
setPhysicalCloseTimeout
public void setPhysicalCloseTimeout(int physicalCloseTimeout)
The time to wait when physically closing the producer via the factory rather than closing the producer itself (whenreset()
,#closeProducerFor(String)
, orcloseThreadBoundProducer()
are invoked). Specified in seconds; defaultProducerFactory.DEFAULT_PHYSICAL_CLOSE_TIMEOUT
.- Parameters:
physicalCloseTimeout
- the timeout in seconds.- Since:
- 1.0.7
-
getPhysicalCloseTimeout
public java.time.Duration getPhysicalCloseTimeout()
Get the physical close timeout.- Specified by:
getPhysicalCloseTimeout
in interfaceProducerFactory<K,V>
- Returns:
- the timeout.
- Since:
- 2.5
-
setTransactionIdPrefix
public final void setTransactionIdPrefix(java.lang.String transactionIdPrefix)
Set a prefix for theProducerConfig.TRANSACTIONAL_ID_CONFIG
config. By default aProducerConfig.TRANSACTIONAL_ID_CONFIG
value from configs is used as a prefix in the target producer configs.- Parameters:
transactionIdPrefix
- the prefix.- Since:
- 1.3
-
getTransactionIdPrefix
@Nullable public java.lang.String getTransactionIdPrefix()
Description copied from interface:ProducerFactory
Return the transaction id prefix.- Specified by:
getTransactionIdPrefix
in interfaceProducerFactory<K,V>
- Returns:
- the prefix or null if not configured.
-
setProducerPerThread
public void setProducerPerThread(boolean producerPerThread)
Set to true to create a producer per thread instead of singleton that is shared by all clients. Clients must callcloseThreadBoundProducer()
to physically close the producer when it is no longer needed. These producers will not be closed bydestroy()
orreset()
.- Parameters:
producerPerThread
- true for a producer per thread.- Since:
- 2.3
- See Also:
closeThreadBoundProducer()
-
isProducerPerThread
public boolean isProducerPerThread()
Description copied from interface:ProducerFactory
Return true when there is a producer per thread.- Specified by:
isProducerPerThread
in interfaceProducerFactory<K,V>
- Returns:
- the producer per thread.
-
setProducerPerConsumerPartition
public void setProducerPerConsumerPartition(boolean producerPerConsumerPartition)
Set to false to revert to the previous behavior of a simple incrementing transactional.id suffix for each producer instead of maintaining a producer for each group/topic/partition.- Parameters:
producerPerConsumerPartition
- false to revert.- Since:
- 1.3.7
-
isProducerPerConsumerPartition
public boolean isProducerPerConsumerPartition()
Return the producerPerConsumerPartition.- Specified by:
isProducerPerConsumerPartition
in interfaceProducerFactory<K,V>
- Returns:
- the producerPerConsumerPartition.
- Since:
- 1.3.8
-
getKeySerializer
@Nullable public org.apache.kafka.common.serialization.Serializer<K> getKeySerializer()
Description copied from interface:ProducerFactory
Return the configured key serializer (if provided as an object instead of a class name in the properties).- Specified by:
getKeySerializer
in interfaceProducerFactory<K,V>
- Returns:
- the serializer.
-
getValueSerializer
@Nullable public org.apache.kafka.common.serialization.Serializer<V> getValueSerializer()
Description copied from interface:ProducerFactory
Return the configured value serializer (if provided as an object instead of a class name in the properties).- Specified by:
getValueSerializer
in interfaceProducerFactory<K,V>
- Returns:
- the serializer.
-
getKeySerializerSupplier
public java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<K>> getKeySerializerSupplier()
Description copied from interface:ProducerFactory
Return a supplier for a key serializer. Useful for cloning to make a similar factory.- Specified by:
getKeySerializerSupplier
in interfaceProducerFactory<K,V>
- Returns:
- the supplier.
-
getValueSerializerSupplier
public java.util.function.Supplier<org.apache.kafka.common.serialization.Serializer<V>> getValueSerializerSupplier()
Description copied from interface:ProducerFactory
Return a supplier for a value serializer. Useful for cloning to make a similar factory.- Specified by:
getValueSerializerSupplier
in interfaceProducerFactory<K,V>
- Returns:
- the supplier.
-
getConfigurationProperties
public java.util.Map<java.lang.String,java.lang.Object> getConfigurationProperties()
Return an unmodifiable reference to the configuration map for this factory. Useful for cloning to make a similar factory.- Specified by:
getConfigurationProperties
in interfaceProducerFactory<K,V>
- Returns:
- the configs.
- Since:
- 1.3
-
getListeners
public java.util.List<ProducerFactory.Listener<K,V>> getListeners()
Get the current list of listeners.- Specified by:
getListeners
in interfaceProducerFactory<K,V>
- Returns:
- the listeners.
- Since:
- 2.5
-
getPostProcessors
public java.util.List<ProducerPostProcessor<K,V>> getPostProcessors()
Description copied from interface:ProducerFactory
Get the current list of post processors.- Specified by:
getPostProcessors
in interfaceProducerFactory<K,V>
- Returns:
- the post processors.
-
setMaxAge
public void setMaxAge(java.time.Duration maxAge)
Set the maximum age for a producer; useful when using transactions and the broker might expire atransactional.id
due to inactivity.- Parameters:
maxAge
- the maxAge to set- Since:
- 2.5.8
-
copyWithConfigurationOverride
public ProducerFactory<K,V> copyWithConfigurationOverride(java.util.Map<java.lang.String,java.lang.Object> overrideProperties)
Copy properties of the instance and the given properties to create a new producer factory.If the
DefaultKafkaProducerFactory
makes a copy of itself, the transaction id prefix is recovered from the properties. If you want to change the ID config, add a newProducerConfig.TRANSACTIONAL_ID_CONFIG
key to the override config.- Specified by:
copyWithConfigurationOverride
in interfaceProducerFactory<K,V>
- Parameters:
overrideProperties
- the properties to be applied to the new factory- Returns:
DefaultKafkaProducerFactory
with properties applied- See Also:
KafkaTemplate(ProducerFactory, java.util.Map)
-
addListener
public void addListener(ProducerFactory.Listener<K,V> listener)
Add a listener.- Specified by:
addListener
in interfaceProducerFactory<K,V>
- Parameters:
listener
- the listener.- Since:
- 2.5
-
addListener
public void addListener(int index, ProducerFactory.Listener<K,V> listener)
Add a listener at a specific index.- Specified by:
addListener
in interfaceProducerFactory<K,V>
- Parameters:
index
- the index (list position).listener
- the listener.- Since:
- 2.5
-
removeListener
public boolean removeListener(ProducerFactory.Listener<K,V> listener)
Remove a listener.- Specified by:
removeListener
in interfaceProducerFactory<K,V>
- Parameters:
listener
- the listener.- Returns:
- true if removed.
- Since:
- 2.5
-
addPostProcessor
public void addPostProcessor(ProducerPostProcessor<K,V> postProcessor)
Description copied from interface:ProducerFactory
Add a post processor.- Specified by:
addPostProcessor
in interfaceProducerFactory<K,V>
- Parameters:
postProcessor
- the post processor.
-
removePostProcessor
public boolean removePostProcessor(ProducerPostProcessor<K,V> postProcessor)
Description copied from interface:ProducerFactory
Remove a post processor.- Specified by:
removePostProcessor
in interfaceProducerFactory<K,V>
- Parameters:
postProcessor
- the post processor.- Returns:
- true if removed.
-
updateConfigs
public void updateConfigs(java.util.Map<java.lang.String,java.lang.Object> updates)
Description copied from interface:ProducerFactory
Update the producer configuration map; useful for situations such as credential rotation.- Specified by:
updateConfigs
in interfaceProducerFactory<K,V>
- Parameters:
updates
- the configuration properties to update.
-
removeConfig
public void removeConfig(java.lang.String configKey)
Description copied from interface:ProducerFactory
Remove the specified key from the configuration map.- Specified by:
removeConfig
in interfaceProducerFactory<K,V>
- Parameters:
configKey
- the key to remove.
-
transactionCapable
public boolean transactionCapable()
Description copied from interface:ProducerFactory
Return true if the factory supports transactions.- Specified by:
transactionCapable
in interfaceProducerFactory<K,V>
- Returns:
- true if transactional.
-
destroy
public void destroy()
- Specified by:
destroy
in interfaceorg.springframework.beans.factory.DisposableBean
-
onApplicationEvent
public void onApplicationEvent(org.springframework.context.event.ContextStoppedEvent event)
- Specified by:
onApplicationEvent
in interfaceorg.springframework.context.ApplicationListener<K>
-
reset
public void reset()
Close theProducer
(s) and clear the cache of transactionalProducer
(s).- Specified by:
reset
in interfaceProducerFactory<K,V>
- Since:
- 2.2
-
createProducer
public org.apache.kafka.clients.producer.Producer<K,V> createProducer()
Description copied from interface:ProducerFactory
Create a producer which will be transactional if the factory is so configured.- Specified by:
createProducer
in interfaceProducerFactory<K,V>
- Returns:
- the producer.
- See Also:
ProducerFactory.transactionCapable()
-
createProducer
public org.apache.kafka.clients.producer.Producer<K,V> createProducer(@Nullable java.lang.String txIdPrefixArg)
Description copied from interface:ProducerFactory
Create a producer with an overridden transaction id prefix.- Specified by:
createProducer
in interfaceProducerFactory<K,V>
- Parameters:
txIdPrefixArg
- the transaction id prefix.- Returns:
- the producer.
-
createNonTransactionalProducer
public org.apache.kafka.clients.producer.Producer<K,V> createNonTransactionalProducer()
Description copied from interface:ProducerFactory
Create a non-transactional producer.- Specified by:
createNonTransactionalProducer
in interfaceProducerFactory<K,V>
- Returns:
- the producer.
- See Also:
ProducerFactory.transactionCapable()
-
createKafkaProducer
protected org.apache.kafka.clients.producer.Producer<K,V> createKafkaProducer()
Subclasses must return a raw producer which will be wrapped in aDefaultKafkaProducerFactory.CloseSafeProducer
.- Returns:
- the producer.
-
createTransactionalProducerForPartition
protected org.apache.kafka.clients.producer.Producer<K,V> createTransactionalProducerForPartition()
-
createTransactionalProducerForPartition
protected org.apache.kafka.clients.producer.Producer<K,V> createTransactionalProducerForPartition(java.lang.String txIdPrefix)
-
removeProducer
protected final boolean removeProducer(DefaultKafkaProducerFactory.CloseSafeProducer<K,V> producerToRemove, java.time.Duration timeout)
Remove the single shared producer and a thread-bound instance if present.- Parameters:
producerToRemove
- the producer.timeout
- the close timeout.- Returns:
- always true.
- Since:
- 2.2.13
-
createTransactionalProducer
protected org.apache.kafka.clients.producer.Producer<K,V> createTransactionalProducer()
Subclasses must return a producer from thegetCache()
or a new raw producer wrapped in aDefaultKafkaProducerFactory.CloseSafeProducer
.- Returns:
- the producer - cannot be null.
- Since:
- 1.3
-
createTransactionalProducer
protected org.apache.kafka.clients.producer.Producer<K,V> createTransactionalProducer(java.lang.String txIdPrefix)
-
createRawProducer
protected org.apache.kafka.clients.producer.Producer<K,V> createRawProducer(java.util.Map<java.lang.String,java.lang.Object> rawConfigs)
-
getCache
@Nullable protected java.util.concurrent.BlockingQueue<DefaultKafkaProducerFactory.CloseSafeProducer<K,V>> getCache()
-
getCache
@Nullable protected java.util.concurrent.BlockingQueue<DefaultKafkaProducerFactory.CloseSafeProducer<K,V>> getCache(java.lang.String txIdPrefix)
-
closeProducerFor
public void closeProducerFor(java.lang.String suffix)
Description copied from interface:ProducerFactory
Remove the specified producer from the cache and close it.- Specified by:
closeProducerFor
in interfaceProducerFactory<K,V>
- Parameters:
suffix
- the producer's transaction id suffix.
-
closeThreadBoundProducer
public void closeThreadBoundProducer()
When usingsetProducerPerThread(boolean)
(true), call this method to close and release this thread's producer. Thread bound producers are not closed bydestroy()
orreset()
methods.- Specified by:
closeThreadBoundProducer
in interfaceProducerFactory<K,V>
- Since:
- 2.3
- See Also:
setProducerPerThread(boolean)
-
getProducerConfigs
protected java.util.Map<java.lang.String,java.lang.Object> getProducerConfigs()
Return the configuration of a producer.- Returns:
- the configuration of a producer.
- Since:
- 2.8.3
- See Also:
createKafkaProducer()
-
getTxProducerConfigs
protected java.util.Map<java.lang.String,java.lang.Object> getTxProducerConfigs(java.lang.String transactionId)
Return the configuration of a transactional producer.- Parameters:
transactionId
- the transactionId.- Returns:
- the configuration of a transactional producer.
- Since:
- 2.8.3
- See Also:
doCreateTxProducer(String, String, BiPredicate)
-
-