Package org.springframework.kafka.config

Class StreamsBuilderFactoryBean

java.lang.Object
org.springframework.beans.factory.config.AbstractFactoryBean<org.apache.kafka.streams.StreamsBuilder>
org.springframework.kafka.config.StreamsBuilderFactoryBean
All Implemented Interfaces:
Aware, BeanClassLoaderAware, BeanFactoryAware, BeanNameAware, DisposableBean, FactoryBean<org.apache.kafka.streams.StreamsBuilder>, InitializingBean, SmartInitializingSingleton, Lifecycle, Phased, SmartLifecycle
public class StreamsBuilderFactoryBean extends AbstractFactoryBean<org.apache.kafka.streams.StreamsBuilder> implements SmartLifecycle, BeanNameAware, SmartInitializingSingleton
An AbstractFactoryBean for the StreamsBuilder instance and lifecycle control for the internal KafkaStreams instance.

A fine grained control on KafkaStreams can be achieved by KafkaStreamsCustomizers.

Since:
1.1.4
Author:
Artem Bilan, Ivan Ursul, Soby Chacko, Zach Olauson, Nurettin Yilmaz, Denis Washington, Gary Russell, Julien Wittouck, Sanghyeok An, Cédric Schaller

  • Field Details

    • DEFAULT_CLOSE_TIMEOUT

      public static final Duration DEFAULT_CLOSE_TIMEOUT
      The default Duration of 10 seconds for close timeout.
      See Also:
      • KafkaStreams.close(Duration)

  • Constructor Details

    • StreamsBuilderFactoryBean

      public StreamsBuilderFactoryBean()
      Default constructor that creates the factory without configuration Properties. It is the factory user's responsibility to properly set Properties using setStreamsConfiguration(Properties).
      Since:
      2.1.3.

    • StreamsBuilderFactoryBean

      public StreamsBuilderFactoryBean(KafkaStreamsConfiguration streamsConfig, CleanupConfig cleanupConfig)
      Construct an instance with the supplied streams configuration and clean up configuration.
      Parameters:
      streamsConfig - the streams configuration.
      cleanupConfig - the cleanup configuration.
      Since:
      2.2

    • StreamsBuilderFactoryBean

      public StreamsBuilderFactoryBean(KafkaStreamsConfiguration streamsConfig)
      Construct an instance with the supplied streams configuration.
      Parameters:
      streamsConfig - the streams configuration.
      Since:
      2.2

  • Method Details

    • setBeanName

      public void setBeanName(String name)
      Specified by:
      setBeanName in interface BeanNameAware

    • setStreamsConfiguration

      public void setStreamsConfiguration(Properties streamsConfig)
      Set the streams configuration Properties on this factory.
      Parameters:
      streamsConfig - the streams configuration.
      Since:
      2.2

    • getStreamsConfiguration

      @Nullable public Properties getStreamsConfiguration()

    • setClientSupplier

      public void setClientSupplier(org.apache.kafka.streams.KafkaClientSupplier clientSupplier)

    • setInfrastructureCustomizer

      public void setInfrastructureCustomizer(KafkaStreamsInfrastructureCustomizer infrastructureCustomizer)
      Set a customizer to configure the builder and/or topology before creating the stream.
      Parameters:
      infrastructureCustomizer - the customizer
      Since:
      2.4.1

    • setKafkaStreamsCustomizer

      public void setKafkaStreamsCustomizer(KafkaStreamsCustomizer kafkaStreamsCustomizer)
      Specify a KafkaStreamsCustomizer to customize a KafkaStreams instance during start().
      Parameters:
      kafkaStreamsCustomizer - the KafkaStreamsCustomizer to use.
      Since:
      2.1.5

    • setStateListener

      public void setStateListener(org.apache.kafka.streams.KafkaStreams.StateListener stateListener)

    • setStreamsUncaughtExceptionHandler

      public void setStreamsUncaughtExceptionHandler(org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler)
      Set a StreamsUncaughtExceptionHandler.
      Parameters:
      streamsUncaughtExceptionHandler - the handler.
      Since:
      2.8

    • getStreamsUncaughtExceptionHandler

      @Nullable public org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler getStreamsUncaughtExceptionHandler()
      Retrieves the current StreamsUncaughtExceptionHandler set on this factory bean.
      Returns:
      StreamsUncaughtExceptionHandler
      Since:
      2.8.4

    • setStateRestoreListener

      public void setStateRestoreListener(org.apache.kafka.streams.processor.StateRestoreListener stateRestoreListener)

    • setCloseTimeout

      public void setCloseTimeout(int closeTimeout)
      Specify the timeout in seconds for the KafkaStreams.close(Duration) operation. Defaults to DEFAULT_CLOSE_TIMEOUT seconds.
      Parameters:
      closeTimeout - the timeout for close in seconds.
      See Also:
      • KafkaStreams.close(Duration)

    • setLeaveGroupOnClose

      public void setLeaveGroupOnClose(boolean leaveGroupOnClose)
      Specify if the consumer should leave the group when stopping Kafka Streams. Defaults to false.
      Parameters:
      leaveGroupOnClose - true to leave the group when stopping the Streams
      Since:
      3.2.0

    • getTopology

      @Nullable public org.apache.kafka.streams.Topology getTopology()
      Providing access to the associated Topology of this StreamsBuilderFactoryBean.
      Returns:
      Topology object
      Since:
      2.4.4

    • getObjectType

      public Class<?> getObjectType()
      Specified by:
      getObjectType in interface FactoryBean<org.apache.kafka.streams.StreamsBuilder>
      Specified by:
      getObjectType in class AbstractFactoryBean<org.apache.kafka.streams.StreamsBuilder>

    • setAutoStartup

      public void setAutoStartup(boolean autoStartup)

    • setPhase

      public void setPhase(int phase)

    • getPhase

      public int getPhase()
      Specified by:
      getPhase in interface Phased
      Specified by:
      getPhase in interface SmartLifecycle

    • setCleanupConfig

      public void setCleanupConfig(CleanupConfig cleanupConfig)

    • getKafkaStreams

      @Nullable public org.apache.kafka.streams.KafkaStreams getKafkaStreams()
      Get a managed by this StreamsBuilderFactoryBean KafkaStreams instance.
      Returns:
      KafkaStreams managed instance; may be null if this StreamsBuilderFactoryBean hasn't been started.
      Since:
      1.1.4

    • getListeners

      public List<StreamsBuilderFactoryBean.Listener> getListeners()
      Get the current list of listeners.
      Returns:
      the listeners.
      Since:
      2.5.3

    • addListener

      public void addListener(StreamsBuilderFactoryBean.Listener listener)
      Add a StreamsBuilderFactoryBean.Listener which will be called after starting and stopping the streams.
      Parameters:
      listener - the listener.
      Since:
      2.5.3

    • removeListener

      public boolean removeListener(StreamsBuilderFactoryBean.Listener listener)
      Remove a listener.
      Parameters:
      listener - the listener.
      Returns:
      true if removed.
      Since:
      2.5.3

    • createInstance

      protected org.apache.kafka.streams.StreamsBuilder createInstance()
      Specified by:
      createInstance in class AbstractFactoryBean<org.apache.kafka.streams.StreamsBuilder>

    • isAutoStartup

      public boolean isAutoStartup()
      Specified by:
      isAutoStartup in interface SmartLifecycle

    • stop

      public void stop(Runnable callback)
      Specified by:
      stop in interface SmartLifecycle

    • start

      public void start()
      Specified by:
      start in interface Lifecycle

    • stop

      public void stop()
      Specified by:
      stop in interface Lifecycle

    • isRunning

      public boolean isRunning()
      Specified by:
      isRunning in interface Lifecycle

    • afterSingletonsInstantiated

      public void afterSingletonsInstantiated()
      Specified by:
      afterSingletonsInstantiated in interface SmartInitializingSingleton