Package org.springframework.integration.kafka.inbound

Class KafkaShareMessageDrivenChannelAdapter<K,V>

java.lang.Object
org.springframework.integration.context.IntegrationObjectSupport
org.springframework.integration.endpoint.AbstractEndpoint
org.springframework.integration.endpoint.MessageProducerSupport
org.springframework.integration.kafka.inbound.KafkaShareMessageDrivenChannelAdapter<K,V>
Type Parameters:
K - the key type.
V - the value type.
All Implemented Interfaces:
Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, SmartInitializingSingleton, ApplicationContextAware, Lifecycle, Phased, SmartLifecycle, ComponentSourceAware, ExpressionCapable, OrderlyShutdownCapable, MessageProducer, IntegrationPattern, KafkaInboundEndpoint, NamedComponent, IntegrationInboundManagement, IntegrationManagement, ManageableLifecycle, ManageableSmartLifecycle, TrackableComponent
public class KafkaShareMessageDrivenChannelAdapter<K,V> extends MessageProducerSupport implements KafkaInboundEndpoint, OrderlyShutdownCapable
Message-driven channel adapter for Kafka share groups (KIP-932 queues), backed by a AbstractShareKafkaMessageListenerContainer.

Share consumers support only explicit topics (no topic patterns or partition assignment), no batch listeners, and no consumer seeks; this adapter therefore has no ListenerMode and no retry template, and does not implement Pausable. It also has no flag to leave a RecordFilterStrategy-discarded record unacknowledged (unlike KafkaMessageDrivenChannelAdapter.setAckDiscarded(boolean)): a discarded record is always terminally acknowledged, since an unacknowledged record in a share group is simply redelivered to another consumer rather than left uncommitted.

The behavior depends on the container's ContainerProperties.ShareAckMode:

  • ContainerProperties.ShareAckMode.EXPLICIT (the default) and ContainerProperties.ShareAckMode.IMPLICIT - the listener receives no ShareAcknowledgment; the container itself sends AcknowledgeType.ACCEPT after the flow returns normally, and consults the container's ShareConsumerRecordRecoverer (default: reject) when the flow throws.
  • ContainerProperties.ShareAckMode.MANUAL - the ShareAcknowledgment is provided in the KafkaHeaders.ACKNOWLEDGMENT message header (together with the KafkaHeaders.CONSUMER header), and whichever flow receives the message is responsible for calling exactly one of ShareAcknowledgment.acknowledge(), ShareAcknowledgment.release(), or ShareAcknowledgment.reject() on it. The container will not poll again - and will not stop - until every record from the current poll has been terminally acknowledged, so a flow that never acknowledges (or hands the message off without acknowledging) stalls the consumer thread permanently. This adapter acknowledges on the flow's behalf only when no flow ever received the message: it rejects a record that failed conversion, and accepts a record discarded by a RecordFilterStrategy.
Since:
7.2
Author:
Artem Bilan
See Also:
  • AbstractShareKafkaMessageListenerContainer
  • ShareAcknowledgment

  • Constructor Details

    • KafkaShareMessageDrivenChannelAdapter

      public KafkaShareMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractShareKafkaMessageListenerContainer<K,V> shareListenerContainer)
      Construct an instance with the provided container.
      Parameters:
      shareListenerContainer - the container.

  • Method Details

    • setRecordMessageConverter

      public void setRecordMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
      Set the message converter to use.
      Parameters:
      messageConverter - the converter.

    • setRecordFilterStrategy

      public void setRecordFilterStrategy(org.springframework.kafka.listener.adapter.RecordFilterStrategy<K,V> recordFilterStrategy)
      Specify a RecordFilterStrategy to discard records before they are converted and sent. A discarded record is terminally acknowledged (accepted) by this adapter when the container's ContainerProperties.ShareAckMode is MANUAL.
      Parameters:
      recordFilterStrategy - the RecordFilterStrategy to use.

    • setPayloadType

      public void setPayloadType(Class<?> payloadType)
      When using a type-aware message converter such as StringJsonMessageConverter, set the payload type the converter should create. Defaults to Object.
      Parameters:
      payloadType - the type.

    • setBindSourceRecord

      public void setBindSourceRecord(boolean bindSourceRecord)
      Set to true to bind the source consumer record in the header named IntegrationMessageHeaderAccessor.SOURCE_DATA.
      Parameters:
      bindSourceRecord - true to bind.

    • getComponentType

      public String getComponentType()
      Specified by:
      getComponentType in interface NamedComponent
      Overrides:
      getComponentType in class MessageProducerSupport

    • onInit

      protected void onInit()
      Description copied from class: IntegrationObjectSupport
      Subclasses may implement this for initialization logic.
      Overrides:
      onInit in class MessageProducerSupport

    • doStart

      protected void doStart()
      Description copied from class: MessageProducerSupport
      Take no action by default. Subclasses may override this if they need lifecycle-managed behavior. Protected by 'lifecycleLock'.
      Overrides:
      doStart in class MessageProducerSupport

    • doStop

      protected void doStop()
      Description copied from class: MessageProducerSupport
      Take no action by default. Subclasses may override this if they need lifecycle-managed behavior.
      Overrides:
      doStop in class MessageProducerSupport

    • beforeShutdown

      public int beforeShutdown()
      Description copied from interface: OrderlyShutdownCapable
      Called before shutdown begins. Implementations should stop accepting new messages. Can optionally return the number of active messages in process.
      Specified by:
      beforeShutdown in interface OrderlyShutdownCapable
      Returns:
      The number of active messages if available.

    • afterShutdown

      public int afterShutdown()
      Description copied from interface: OrderlyShutdownCapable
      Called after normal shutdown of schedulers, executors etc, and after the shutdown delay has elapsed, but before any forced shutdown of any remaining active scheduler/executor threads.Can optionally return the number of active messages still in process.
      Specified by:
      afterShutdown in interface OrderlyShutdownCapable
      Returns:
      The number of active messages if available.

    • getErrorMessageAttributes

      protected AttributeAccessor getErrorMessageAttributes(@Nullable Message<?> message)
      Description copied from class: MessageProducerSupport
      Populate an AttributeAccessor to be used when building an error message with the errorMessageStrategy.
      Overrides:
      getErrorMessageAttributes in class MessageProducerSupport
      Parameters:
      message - the message.
      Returns:
      the attributes.