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.InitializingBean, org.springframework.context.ApplicationContextAware, org.springframework.integration.expression.IntegrationEvaluationContextAware, MessageBus

public class KafkaMessageBus
extends MessageBusSupport

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
foo = "http | log", log.count=0, x actual log containers	foo.0	10(configurable) partitions	1 producer, x XD consumers. Can't know the number of partitions beforehand, so decide a number that better be greater than number of containers

Author:

Eric Bottard, Marius Bogoevici

Nested Class Summary

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>

Field Summary

Fields

Modifier and Type	Field and Description
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, defaultBackOffInitialInterval, defaultBackOffMaxInterval, defaultBackOffMultiplier, defaultConcurrency, defaultMaxAttempts, directChannelProvider, evaluationContext, JOB_CHANNEL_TYPE_PREFIX, logger, MEDIATYPES_MEDIATYPE_ALL, ORIGINAL_CONTENT_TYPE_HEADER, P2P_NAMED_CHANNEL_TYPE_PREFIX, PARTITION_HEADER, PRODUCER_PARTITIONING_PROPERTIES, PRODUCER_STANDARD_PROPERTIES, PUBSUB_NAMED_CHANNEL_TYPE_PREFIX

Constructor Summary

Constructors

Constructor and Description
KafkaMessageBus(java.lang.String brokers, java.lang.String zkAddress, MultiTypeCodec<java.lang.Object> codec, java.lang.String... headersToMap)

Method Summary

Methods

Modifier and Type	Method and Description
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.
void	setReplicationFactor(int replicationFactor)

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

addBinding, afterPropertiesSet, bindDynamicProducer, bindDynamicPubSubProducer, bindExistingProducerDirectlyIfPossible, bindNewProducerDirectlyIfPossible, buildPartitionRoutingExpression, buildRetryTemplateIfRetryEnabled, deleteBinding, deleteBindings, deserializePayloadIfNecessary, determinePartition, doBindDynamicProducer, doBindDynamicPubSubProducer, getApplicationContext, getBeanFactory, getIdGenerator, isNamedChannel, onInit, serializePayloadIfNecessary, setApplicationContext, setCodec, setDefaultBackOffInitialInterval, setDefaultBackOffMaxInterval, setDefaultBackOffMultiplier, setDefaultConcurrency, 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

utf8Serializer

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

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

Constructor Detail

KafkaMessageBus

public KafkaMessageBus(java.lang.String brokers,
               java.lang.String zkAddress,
               MultiTypeCodec<java.lang.Object> codec,
               java.lang.String... headersToMap)

Method Detail

setReplicationFactor

public void setReplicationFactor(int replicationFactor)

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

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

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.

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.

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().

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().

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.