Package org.springframework.kafka.listener

Class ConcurrentMessageListenerContainer<K,V>

java.lang.Object
org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V>
org.springframework.kafka.listener.ConcurrentMessageListenerContainer<K,V>
Type Parameters:
K - the key type.
V - the value type.
All Implemented Interfaces:
Aware, BeanNameAware, DisposableBean, ApplicationContextAware, ApplicationEventPublisherAware, Lifecycle, Phased, SmartLifecycle, GenericMessageListenerContainer<K,V>, MessageListenerContainer
public class ConcurrentMessageListenerContainer<K,V> extends AbstractMessageListenerContainer<K,V>
Creates 1 or more KafkaMessageListenerContainers based on concurrency. If the ContainerProperties is configured with TopicPartitions, the TopicPartitions are distributed evenly across the instances.
Author:
Marius Bogoevici, Gary Russell, Murali Reddy, Jerome Mirc, Artem Bilan, Vladimir Tsanev, Tomaz Fernandes

  • Constructor Details

    • ConcurrentMessageListenerContainer

      public ConcurrentMessageListenerContainer(ConsumerFactory<? super K,? super V> consumerFactory, ContainerProperties containerProperties)
      Construct an instance with the supplied configuration properties. The topic partitions are distributed evenly across the delegate KafkaMessageListenerContainers.
      Parameters:
      consumerFactory - the consumer factory.
      containerProperties - the container properties.

  • Method Details

    • getConcurrency

      public int getConcurrency()

    • setConcurrency

      public void setConcurrency(int concurrency)
      The maximum number of concurrent KafkaMessageListenerContainers running. Messages from within the same partition will be processed sequentially.
      Parameters:
      concurrency - the concurrency.

    • setAlwaysClientIdSuffix

      public void setAlwaysClientIdSuffix(boolean alwaysClientIdSuffix)
      Set to false to suppress adding a suffix to the child container's client.id when the concurrency is only 1.
      Parameters:
      alwaysClientIdSuffix - false to suppress, true (default) to include.
      Since:
      2.2.14

    • getContainers

      public List<KafkaMessageListenerContainer<K,V>> getContainers()
      Return the list of KafkaMessageListenerContainers created by this container.
      Returns:
      the list of KafkaMessageListenerContainers created by this container.

    • getContainerFor

      public MessageListenerContainer getContainerFor(String topic, int partition)
      Description copied from interface: MessageListenerContainer
      If this container has child containers, return the child container that is assigned the topic/partition. Return this when there are no child containers.
      Parameters:
      topic - the topic.
      partition - the partition.
      Returns:
      the container.

    • getAssignedPartitions

      public Collection<org.apache.kafka.common.TopicPartition> getAssignedPartitions()
      Description copied from interface: MessageListenerContainer
      Return the assigned topics/partitions for this container.
      Returns:
      the topics/partitions.

    • getAssignmentsByClientId

      public Map<String,Collection<org.apache.kafka.common.TopicPartition>> getAssignmentsByClientId()
      Description copied from interface: MessageListenerContainer
      Return the assigned topics/partitions for this container, by client.id.
      Returns:
      the topics/partitions.

    • isContainerPaused

      public boolean isContainerPaused()
      Description copied from interface: MessageListenerContainer
      Return true if MessageListenerContainer.pause() has been called; and all consumers in this container have actually paused.
      Returns:
      true if the container is paused.

    • isChildRunning

      public boolean isChildRunning()
      Description copied from interface: MessageListenerContainer
      If this container has child containers, return true if at least one child is running. If there are not child containers, returns Lifecycle.isRunning().
      Returns:
      true if a child is running.

    • metrics

      public Map<String,Map<org.apache.kafka.common.MetricName,? extends org.apache.kafka.common.Metric>> metrics()
      Description copied from interface: MessageListenerContainer
      Return metrics kept by this container's consumer(s), grouped by client-id.
      Returns:
      the consumer(s) metrics grouped by client-id
      See Also:
      • Consumer.metrics()

    • doStart

      protected void doStart()
      Specified by:
      doStart in class AbstractMessageListenerContainer<K,V>

    • doStop

      protected void doStop(Runnable callback, boolean normal)
      Description copied from class: AbstractMessageListenerContainer
      Stop the container normally or abnormally.
      Specified by:
      doStop in class AbstractMessageListenerContainer<K,V>
      Parameters:
      callback - the callback.
      normal - true for an expected stop.

    • childStopped

      public void childStopped(MessageListenerContainer child, ConsumerStoppedEvent.Reason reason)
      Description copied from interface: MessageListenerContainer
      Notify a parent container that a child container has stopped.
      Parameters:
      child - the container.
      reason - the reason.

    • enforceRebalance

      public void enforceRebalance()
      Description copied from interface: MessageListenerContainer
      Alerting the consumer to trigger an enforced rebalance. The actual enforce will happen when the next poll() operation is invoked.
      See Also:
      • KafkaConsumer.enforceRebalance()

    • pause

      public void pause()
      Description copied from interface: MessageListenerContainer
      Pause this container before the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.
      Specified by:
      pause in interface MessageListenerContainer
      Overrides:
      pause in class AbstractMessageListenerContainer<K,V>
      See Also:
      • KafkaConsumer.pause(Collection)

    • resume

      public void resume()
      Description copied from interface: MessageListenerContainer
      Resume this container, if paused, after the next poll(). This is a thread-safe operation, the actual resume is processed by the consumer thread.
      Specified by:
      resume in interface MessageListenerContainer
      Overrides:
      resume in class AbstractMessageListenerContainer<K,V>
      See Also:
      • KafkaConsumer.resume(Collection)

    • pausePartition

      public void pausePartition(org.apache.kafka.common.TopicPartition topicPartition)
      Description copied from interface: MessageListenerContainer
      Pause this partition before the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.
      Specified by:
      pausePartition in interface MessageListenerContainer
      Overrides:
      pausePartition in class AbstractMessageListenerContainer<K,V>
      Parameters:
      topicPartition - the topicPartition to pause.

    • resumePartition

      public void resumePartition(org.apache.kafka.common.TopicPartition topicPartition)
      Description copied from interface: MessageListenerContainer
      Resume this partition, if paused, after the next poll(). This is a thread-safe operation, the actual pause is processed by the consumer thread.
      Specified by:
      resumePartition in interface MessageListenerContainer
      Overrides:
      resumePartition in class AbstractMessageListenerContainer<K,V>
      Parameters:
      topicPartition - the topicPartition to resume.

    • isPartitionPaused

      public boolean isPartitionPaused(org.apache.kafka.common.TopicPartition topicPartition)
      Description copied from interface: MessageListenerContainer
      Whether or not this topic's partition is currently paused.
      Parameters:
      topicPartition - the topic partition to check
      Returns:
      true if this partition has been paused.

    • isInExpectedState

      public boolean isInExpectedState()
      Description copied from interface: MessageListenerContainer
      Return true if the container is running, has never been started, or has been stopped.
      Returns:
      true if the state is as expected.
      See Also:

    • toString

      public String toString()
      Overrides:
      toString in class Object