org.springframework.xd.dirt.integration.kafka

Class KafkaMessageBus

java.lang.Object

org.springframework.xd.dirt.integration.bus.MessageBusSupport

org.springframework.xd.dirt.integration.kafka.KafkaMessageBus

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, MessageBus

public class KafkaMessageBus
extends MessageBusSupport
implements org.springframework.beans.factory.DisposableBean

A message bus that uses Kafka as the underlying middleware. The general implementation mapping between XD concepts and Kafka concepts is as follows: A message bus that uses Kafka as the underlying middleware. The general implementation mapping between XD concepts and Kafka concepts is as follows:

Stream definition	Kafka topic	Kafka partitions	Notes
foo = "http | log"	foo.0	1 partition	1 producer, 1 consumer
foo = "http | log", log.count=x	foo.0	x partitions	1 producer, x consumers with static group 'springXD', achieves queue semantics
foo = "http | log", log.count=x + XD partitioning	still 1 topic 'foo.0'	x partitions + use key computed by XD	1 producer, x consumers with static group 'springXD', achieves queue semantics
foo = "http | log", log.count=x, concurrency=y	foo.0	x*y partitions	1 producer, x XD consumers, each with y threads

Author:

Eric Bottard, Marius Bogoevici, Ilayaperumal Gopinathan, David Turanski, Gary Russell, Martin Dam

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaMessageBus.Mode
static class	KafkaMessageBus.OffsetManagement

Nested classes/interfaces inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

MessageBusSupport.DirectHandler, MessageBusSupport.PartitioningMetadata, MessageBusSupport.SetBuilder, MessageBusSupport.SharedChannelProvider<T extends org.springframework.messaging.MessageChannel>

Nested classes/interfaces inherited from interface org.springframework.xd.dirt.integration.bus.MessageBus

MessageBus.Capability

Field Summary

Fields

Modifier and Type	Field and Description
static java.lang.String	AUTO_COMMIT_OFFSET_ENABLED
static org.apache.kafka.common.serialization.ByteArraySerializer	BYTE_ARRAY_SERIALIZER
static java.lang.String	COMPRESSION_CODEC
static java.lang.String	FETCH_SIZE
static int	METADATA_VERIFICATION_MAX_INTERVAL
static int	METADATA_VERIFICATION_RETRY_ATTEMPTS
static double	METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER
static int	METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL
protected static java.util.Set<java.lang.Object>	PRODUCER_COMPRESSION_PROPERTIES
static java.lang.String	QUEUE_SIZE
static java.lang.String	REQUIRED_ACKS
static java.lang.String	SYNC_PRODUCER
static java.lang.String	SYNC_PRODUCER_TIMEOUT
static org.I0Itec.zkclient.serialize.ZkSerializer	utf8Serializer Used when writing directly to ZK.

Fields inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

CONSUMER_RETRY_PROPERTIES, CONSUMER_STANDARD_PROPERTIES, defaultBackOffInitialInterval, defaultBackOffMaxInterval, defaultBackOffMultiplier, defaultBatchBufferLimit, defaultBatchingEnabled, defaultBatchSize, defaultBatchTimeout, defaultCompress, defaultConcurrency, defaultMaxAttempts, directChannelProvider, evaluationContext, JOB_CHANNEL_TYPE_PREFIX, logger, MEDIATYPES_MEDIATYPE_ALL, P2P_NAMED_CHANNEL_TYPE_PREFIX, PARTITION_HEADER, PRODUCER_BATCHING_ADVANCED_PROPERTIES, PRODUCER_BATCHING_BASIC_PROPERTIES, PRODUCER_PARTITIONING_PROPERTIES, PRODUCER_STANDARD_PROPERTIES, PUBSUB_NAMED_CHANNEL_TYPE_PREFIX, TAP_TYPE_PREFIX

Constructor Summary

Constructors

Constructor and Description
KafkaMessageBus(org.springframework.integration.kafka.support.ZookeeperConnect zookeeperConnect, java.lang.String brokers, java.lang.String zkAddress, org.springframework.integration.codec.Codec codec, java.lang.String... headersToMap)

Method Summary

Methods

Modifier and Type	Method and Description
void	afterPropertiesSet()
void	bindConsumer(java.lang.String name, org.springframework.messaging.MessageChannel moduleInputChannel, java.util.Properties properties) Bind a message consumer on a p2p channel
void	bindProducer(java.lang.String name, org.springframework.messaging.MessageChannel moduleOutputChannel, java.util.Properties properties) Bind a message producer on a p2p channel.
void	bindPubSubConsumer(java.lang.String name, org.springframework.messaging.MessageChannel inputChannel, java.util.Properties properties) Bind a message consumer on a pub/sub channel
void	bindPubSubProducer(java.lang.String name, org.springframework.messaging.MessageChannel outputChannel, java.util.Properties properties) Bind a message producer on a pub/sub channel.
void	bindReplier(java.lang.String name, org.springframework.messaging.MessageChannel requests, org.springframework.messaging.MessageChannel replies, java.util.Properties properties) Bind a consumer that handles requests from a requestor and asynchronously sends replies.
void	bindRequestor(java.lang.String name, org.springframework.messaging.MessageChannel requests, org.springframework.messaging.MessageChannel replies, java.util.Properties properties) Bind a producer that expects async replies.
org.springframework.integration.kafka.listener.KafkaMessageListenerContainer	createMessageListenerContainer(java.util.Properties properties, java.lang.String group, int maxConcurrency, java.lang.String topic, long referencePoint, boolean resetOffsets)
void	destroy()
void	doManualAck(java.util.LinkedList<org.springframework.messaging.MessageHeaders> messageHeadersList) Perform manual acknowledgement based on the metadata stored in message bus.
static java.lang.String	escapeTopicName(java.lang.String original) Allowed chars are ASCII alphanumerics, '.', '_' and '-'.
org.springframework.integration.kafka.core.ConnectionFactory	getConnectionFactory()
boolean	isCapable(MessageBus.Capability capability) Return true if the bus supports the capability.
void	setDefaultAutoCommitOffsetEnabled(boolean defaultAutoCommitOffsetEnabled) Set the default auto commit enabled property; This is used to commit the offset either automatically or manually.
void	setDefaultCompressionCodec(java.lang.String defaultCompressionCodec)
void	setDefaultFetchSize(int defaultFetchSize)
void	setDefaultMaxWait(int defaultMaxWait)
void	setDefaultMinPartitionCount(int defaultMinPartitionCount)
void	setDefaultQueueSize(int defaultQueueSize)
void	setDefaultReplicationFactor(int defaultReplicationFactor)
void	setDefaultRequiredAcks(int defaultRequiredAcks)
void	setDefaultSyncProducer(boolean syncProducer)
void	setDefaultSyncProducerTimeout(int timeout)
void	setMode(KafkaMessageBus.Mode mode)
void	setOffsetManagement(KafkaMessageBus.OffsetManagement offsetManagement) Set the KafkaMessageBus.OffsetManagement to use.
void	setOffsetStoreBatchBytes(int offsetStoreBatchBytes)
void	setOffsetStoreBatchTime(int offsetStoreBatchTime)
void	setOffsetStoreMaxFetchSize(int offsetStoreMaxFetchSize)
void	setOffsetStoreRequiredAcks(int offsetStoreRequiredAcks)
void	setOffsetStoreRetentionTime(int offsetStoreRetentionTime)
void	setOffsetStoreSegmentSize(int offsetStoreSegmentSize)
void	setOffsetStoreTopic(java.lang.String offsetStoreTopic)
void	setOffsetUpdateCount(int offsetUpdateCount)
void	setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout)
void	setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow)
void	setRetryOperations(org.springframework.retry.RetryOperations retryOperations) Retry configuration for operations such as validating topic creation
void	setSocketBufferSize(int socketBufferSize)

Methods inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

addBinding, applyPrefix, applyPubSub, applyRequests, bindDynamicProducer, bindDynamicPubSubProducer, bindExistingProducerDirectlyIfPossible, bindNewProducerDirectlyIfPossible, buildPartitionRoutingExpression, buildRetryTemplateIfRetryEnabled, constructDLQName, deleteBinding, deleteBindings, deserializePayloadIfNecessary, deserializePayloadIfNecessary, determinePartition, doBindDynamicProducer, doBindDynamicPubSubProducer, getApplicationContext, getBeanFactory, getIdGenerator, isNamedChannel, onInit, serializePayloadIfNecessary, setApplicationContext, setCodec, setDefaultBackOffInitialInterval, setDefaultBackOffMaxInterval, setDefaultBackOffMultiplier, setDefaultBatchBufferLimit, setDefaultBatchingEnabled, setDefaultBatchSize, setDefaultBatchTimeout, setDefaultCompress, setDefaultConcurrency, setDefaultDurableSubscription, setDefaultMaxAttempts, setIntegrationEvaluationContext, setPartitionSelector, stopBindings, unbindConsumer, unbindConsumers, unbindProducer, unbindProducers, validateConsumerProperties, validateProducerProperties

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

BYTE_ARRAY_SERIALIZER

public static final org.apache.kafka.common.serialization.ByteArraySerializer BYTE_ARRAY_SERIALIZER

METADATA_VERIFICATION_RETRY_ATTEMPTS

public static final int METADATA_VERIFICATION_RETRY_ATTEMPTS

See Also:

Constant Field Values

METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER

public static final double METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER

See Also:

Constant Field Values

METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL

public static final int METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL

See Also:

Constant Field Values

METADATA_VERIFICATION_MAX_INTERVAL

public static final int METADATA_VERIFICATION_MAX_INTERVAL

See Also:

Constant Field Values

FETCH_SIZE

public static final java.lang.String FETCH_SIZE

See Also:

Constant Field Values

QUEUE_SIZE

public static final java.lang.String QUEUE_SIZE

See Also:

Constant Field Values

REQUIRED_ACKS

public static final java.lang.String REQUIRED_ACKS

See Also:

Constant Field Values

COMPRESSION_CODEC

public static final java.lang.String COMPRESSION_CODEC

See Also:

Constant Field Values

AUTO_COMMIT_OFFSET_ENABLED

public static final java.lang.String AUTO_COMMIT_OFFSET_ENABLED

See Also:

Constant Field Values

SYNC_PRODUCER

public static final java.lang.String SYNC_PRODUCER

See Also:

Constant Field Values

SYNC_PRODUCER_TIMEOUT

public static final java.lang.String SYNC_PRODUCER_TIMEOUT

See Also:

Constant Field Values

utf8Serializer

public static final org.I0Itec.zkclient.serialize.ZkSerializer utf8Serializer

Used when writing directly to ZK. This is what Kafka expects.

PRODUCER_COMPRESSION_PROPERTIES

protected static final java.util.Set<java.lang.Object> PRODUCER_COMPRESSION_PROPERTIES

Constructor Detail

KafkaMessageBus

public KafkaMessageBus(org.springframework.integration.kafka.support.ZookeeperConnect zookeeperConnect,
               java.lang.String brokers,
               java.lang.String zkAddress,
               org.springframework.integration.codec.Codec codec,
               java.lang.String... headersToMap)

Method Detail

setOffsetStoreTopic

public void setOffsetStoreTopic(java.lang.String offsetStoreTopic)

setOffsetStoreSegmentSize

public void setOffsetStoreSegmentSize(int offsetStoreSegmentSize)

setOffsetStoreRetentionTime

public void setOffsetStoreRetentionTime(int offsetStoreRetentionTime)

setSocketBufferSize

public void setSocketBufferSize(int socketBufferSize)

setOffsetStoreRequiredAcks

public void setOffsetStoreRequiredAcks(int offsetStoreRequiredAcks)

setOffsetStoreMaxFetchSize

public void setOffsetStoreMaxFetchSize(int offsetStoreMaxFetchSize)

setOffsetUpdateTimeWindow

public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow)

setOffsetUpdateCount

public void setOffsetUpdateCount(int offsetUpdateCount)

setOffsetUpdateShutdownTimeout

public void setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout)

setOffsetStoreBatchBytes

public void setOffsetStoreBatchBytes(int offsetStoreBatchBytes)

setOffsetStoreBatchTime

public void setOffsetStoreBatchTime(int offsetStoreBatchTime)

getConnectionFactory

public org.springframework.integration.kafka.core.ConnectionFactory getConnectionFactory()

setRetryOperations

public void setRetryOperations(org.springframework.retry.RetryOperations retryOperations)

Retry configuration for operations such as validating topic creation

Parameters:

retryOperations - the retry configuration

afterPropertiesSet

public void afterPropertiesSet()
                        throws java.lang.Exception

Specified by:

afterPropertiesSet in interface org.springframework.beans.factory.InitializingBean

Overrides:

afterPropertiesSet in class MessageBusSupport

Throws:

java.lang.Exception

destroy

public void destroy()
             throws java.lang.Exception

Specified by:

destroy in interface org.springframework.beans.factory.DisposableBean

Throws:

java.lang.Exception

isCapable

public boolean isCapable(MessageBus.Capability capability)

Description copied from interface: MessageBus

Return true if the bus supports the capability.

Specified by:

isCapable in interface MessageBus

Overrides:

isCapable in class MessageBusSupport

Parameters:

capability - the capability.

Returns:

true if the capability is supported.

escapeTopicName

public static java.lang.String escapeTopicName(java.lang.String original)

Allowed chars are ASCII alphanumerics, '.', '_' and '-'. '_' is used as escaped char in the form '_xx' where xx is the hexadecimal value of the byte(s) needed to represent an illegal char in utf8.

setDefaultReplicationFactor

public void setDefaultReplicationFactor(int defaultReplicationFactor)

setDefaultCompressionCodec

public void setDefaultCompressionCodec(java.lang.String defaultCompressionCodec)

setDefaultRequiredAcks

public void setDefaultRequiredAcks(int defaultRequiredAcks)

setDefaultAutoCommitOffsetEnabled

public void setDefaultAutoCommitOffsetEnabled(boolean defaultAutoCommitOffsetEnabled)

Set the default auto commit enabled property; This is used to commit the offset either automatically or manually.

Parameters:

defaultAutoCommitOffsetEnabled -

setDefaultQueueSize

public void setDefaultQueueSize(int defaultQueueSize)

setDefaultFetchSize

public void setDefaultFetchSize(int defaultFetchSize)

setDefaultMinPartitionCount

public void setDefaultMinPartitionCount(int defaultMinPartitionCount)

setDefaultMaxWait

public void setDefaultMaxWait(int defaultMaxWait)

setDefaultSyncProducer

public void setDefaultSyncProducer(boolean syncProducer)

setDefaultSyncProducerTimeout

public void setDefaultSyncProducerTimeout(int timeout)

setMode

public void setMode(KafkaMessageBus.Mode mode)

setOffsetManagement

public void setOffsetManagement(KafkaMessageBus.OffsetManagement offsetManagement)

Set the KafkaMessageBus.OffsetManagement to use. Default: KafkaMessageBus.OffsetManagement.kafkaTopic.

Parameters:

offsetManagement - the offsetManagement.

bindConsumer

public void bindConsumer(java.lang.String name,
                org.springframework.messaging.MessageChannel moduleInputChannel,
                java.util.Properties properties)

Description copied from interface: MessageBus

Bind a message consumer on a p2p channel

Specified by:

bindConsumer in interface MessageBus

Parameters:

name - the logical identity of the message source

moduleInputChannel - the channel bound as a consumer

properties - arbitrary String key/value pairs that will be used in the binding

bindPubSubConsumer

public void bindPubSubConsumer(java.lang.String name,
                      org.springframework.messaging.MessageChannel inputChannel,
                      java.util.Properties properties)

Description copied from interface: MessageBus

Bind a message consumer on a pub/sub channel

Specified by:

bindPubSubConsumer in interface MessageBus

Parameters:

name - the logical identity of the message source

inputChannel - the channel bound as a pub/sub consumer

properties - arbitrary String key/value pairs that will be used in the binding

bindProducer

public void bindProducer(java.lang.String name,
                org.springframework.messaging.MessageChannel moduleOutputChannel,
                java.util.Properties properties)

Description copied from interface: MessageBus

Bind a message producer on a p2p channel.

Specified by:

bindProducer in interface MessageBus

Parameters:

name - the logical identity of the message target

moduleOutputChannel - the channel bound as a producer

properties - arbitrary String key/value pairs that will be used in the binding

bindPubSubProducer

public void bindPubSubProducer(java.lang.String name,
                      org.springframework.messaging.MessageChannel outputChannel,
                      java.util.Properties properties)

Description copied from interface: MessageBus

Bind a message producer on a pub/sub channel.

Specified by:

bindPubSubProducer in interface MessageBus

Parameters:

name - the logical identity of the message target

outputChannel - the channel bound as a producer

properties - arbitrary String key/value pairs that will be used in the binding

bindRequestor

public void bindRequestor(java.lang.String name,
                 org.springframework.messaging.MessageChannel requests,
                 org.springframework.messaging.MessageChannel replies,
                 java.util.Properties properties)

Description copied from interface: MessageBus

Bind a producer that expects async replies. To unbind, invoke unbindProducer() and unbindConsumer().

Specified by:

bindRequestor in interface MessageBus

Parameters:

name - The name of the requestor.

requests - The request channel - sends requests.

replies - The reply channel - receives replies.

properties - arbitrary String key/value pairs that will be used in the binding.

bindReplier

public void bindReplier(java.lang.String name,
               org.springframework.messaging.MessageChannel requests,
               org.springframework.messaging.MessageChannel replies,
               java.util.Properties properties)

Description copied from interface: MessageBus

Bind a consumer that handles requests from a requestor and asynchronously sends replies. To unbind, invoke unbindProducer() and unbindConsumer().

Specified by:

bindReplier in interface MessageBus

Parameters:

name - The name of the requestor for which this replier will handle requests.

requests - The request channel - receives requests.

replies - The reply channel - sends replies.

properties - arbitrary String key/value pairs that will be used in the binding.

createMessageListenerContainer

public org.springframework.integration.kafka.listener.KafkaMessageListenerContainer createMessageListenerContainer(java.util.Properties properties,
                                                                                                          java.lang.String group,
                                                                                                          int maxConcurrency,
                                                                                                          java.lang.String topic,
                                                                                                          long referencePoint,
                                                                                                          boolean resetOffsets)

doManualAck

public void doManualAck(java.util.LinkedList<org.springframework.messaging.MessageHeaders> messageHeadersList)

Description copied from class: MessageBusSupport

Perform manual acknowledgement based on the metadata stored in message bus.

Overrides:

doManualAck in class MessageBusSupport