From ae2a1a08cd8a2b9c15395ddd46810f5aabcb5ca7 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Tue, 3 Feb 2015 21:29:30 +0200 Subject: [PATCH] INTEXT-131-134: Add KafkaTopicOffsetManager JIRA: https://jira.spring.io/browse/INTEXT-131, https://jira.spring.io/browse/INTEXT-134 Add deletion support to OffsetManagers - Separate MetadataStore-related functionality and extract an AbstractOffsetManager to handle resetting/retrieving default offsets from Kafka - Move connection parameters to the Configuration class - Add deletion support to OffsetManagers - Introduce KafkaOffsetManager Renamed to TopicUtils Use an embedded broker for outbound tests. INTEXT-135 Dynamic producing to topics in the KafkaProducingMessageHandler Rework how DefaultConnectionFactory manages cached metadata -Renamed PartitionBrokerMap to a more apt MetadataCache -Removed bidirectional mapping -MetadataCache is still immutable but now allows to create new instances with new TopicMetadata -As a result, DefaultConnectionFactory can load topic data as required, without requiring explicit resets - DefaultConnectionFactory will refresh internally if data about a topic is not available, transparently retrieving data from the server Corrections after PR comments: - use a readLock for getLeaders as well - renamed `refreshLeaders` to `refreshMetadata` - various minor fixes Double lock check Make AbstractOffsetManager Disposable an close on that Corrections, logs, etc. INTEXT-131-134: Polishing * Add `initialOffsets` "free" `AbstractOffsetManager` * Get rid of `connectionFactory` ctor argument for the `KafkaTopicOffsetManager` * `start/stop` checks to the `KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests` * Add note to the `README.md` about `KafkaTopicOffsetManager` --- .../kafka/core/AbstractConfiguration.java | 120 +++- .../integration/kafka/core/BrokerAddress.java | 32 +- .../integration/kafka/core/Configuration.java | 48 +- .../integration/kafka/core/Connection.java | 3 +- .../kafka/core/ConnectionFactory.java | 4 +- .../kafka/core/DefaultConnectionFactory.java | 229 ++++---- .../integration/kafka/core/MetadataCache.java | 111 ++++ .../integration/kafka/core/Partition.java | 9 +- .../kafka/core/PartitionBrokerMap.java | 66 --- .../core/PartitionNotFoundException.java | 30 + .../kafka/core/ZookeeperConfiguration.java | 23 +- .../kafka/listener/AbstractOffsetManager.java | 165 ++++++ .../KafkaMessageListenerContainer.java | 8 +- .../listener/KafkaTopicOffsetManager.java | 521 ++++++++++++++++++ .../kafka/listener/LongEncoderDecoder.java | 66 +++ .../listener/MetadataStoreOffsetManager.java | 131 +---- .../kafka/listener/OffsetManager.java | 9 +- .../kafka/support/KafkaProducerContext.java | 4 +- .../kafka/support/ZookeeperConnect.java | 11 + .../integration/kafka/util/LoggingUtils.java | 45 ++ .../integration/kafka/util/MessageUtils.java | 8 +- .../integration/kafka/util/TopicUtils.java | 156 ++++++ .../kafka/core/MetadataCacheTests.java | 262 +++++++++ .../kafka/listener/AbstractBrokerTests.java | 80 +-- ...nelAdapterWithKafkaOffsetManagerTests.java | 135 +++++ ...venChannelAdapterWithWrongOffsetTests.java | 5 + .../offset/AbstractOffsetManagerTests.java | 295 ++++++++++ .../integration/kafka/offset/CodecTests.java | 52 ++ .../kafka/offset/KafkaOffsetManagerTests.java | 47 ++ .../MetadataStoreOffsetManagerTests.java | 53 ++ .../kafka/outbound/OutboundTests.java | 226 ++++++-- .../integration/kafka/rule/KafkaEmbedded.java | 1 + .../integration/kafka/rule/KafkaRule.java | 2 + .../integration/kafka/rule/KafkaRunning.java | 5 + 34 files changed, 2514 insertions(+), 448 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/MetadataCache.java delete mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionBrokerMap.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionNotFoundException.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaTopicOffsetManager.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/LongEncoderDecoder.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/LoggingUtils.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/TopicUtils.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/core/MetadataCacheTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/AbstractOffsetManagerTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/CodecTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/MetadataStoreOffsetManagerTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java index b1ed97f897..1d85fd94c8 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java @@ -14,7 +14,6 @@ * limitations under the License. */ - package org.springframework.integration.kafka.core; import java.util.List; @@ -25,7 +24,8 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; /** - * Default implementation of a {@link Configuration}, storing the default topic and partitions explicitly. + * Default implementation of a {@link Configuration}, storing the default topic and partitions, + * as well as connectivity parameters. * * Implementors must provide a strategy for retrieving the seed brokers. * @@ -37,10 +37,107 @@ public abstract class AbstractConfiguration implements InitializingBean, Configu private String defaultTopic; + private String clientId = KafkaConsumerDefaults.GROUP_ID; + + private int minBytes = KafkaConsumerDefaults.MIN_FETCH_BYTES; + + private int maxWait = KafkaConsumerDefaults.MAX_WAIT_TIME_IN_MS; + + private int bufferSize = KafkaConsumerDefaults.SOCKET_BUFFER_SIZE_INT; + + private int socketTimeout = KafkaConsumerDefaults.SOCKET_TIMEOUT_INT; + + private int fetchMetadataTimeout = KafkaConsumerDefaults.FETCH_METADATA_TIMEOUT; + + + /** + * The minimum amount of data that a server fetch operation will wait for before returning, + * unless {@code maxWait} has elapsed. + * In conjunction with {@link Configuration#getMaxWait()}}, controls latency + * and throughput. + * Smaller values increase responsiveness, but may increase the number of poll operations, + * potentially reducing throughput and increasing CPU consumption. + * @param minBytes the amount of data to fetch + */ + public void setMinBytes(int minBytes) { + this.minBytes = minBytes; + } + @Override - public void afterPropertiesSet() throws Exception { - Assert.isTrue(CollectionUtils.isEmpty(defaultPartitions) || StringUtils.isEmpty(defaultTopic) - , "A list of default partitions or a default topic may be specified, but not both"); + public int getMinBytes() { + return this.minBytes; + } + + /** + * The maximum amount of time that a server fetch operation will wait before returning + * (unless {@code minFetchSizeInBytes}) are available. + * In conjunction with {@link AbstractConfiguration#setMinBytes(int)}, + * controls latency and throughput. + * Smaller intervals increase responsiveness, but may increase + * the number of poll operations, potentially increasing CPU + * consumption and reducing throughput. + * @param maxWait timeout to wait + */ + public void setMaxWait(int maxWait) { + this.maxWait = maxWait; + } + + @Override + public int getMaxWait() { + return this.maxWait; + } + + @Override + public String getClientId() { + return this.clientId; + } + + /** + * A client name to be used throughout this connection. + * @param clientId the client name + */ + public void setClientId(String clientId) { + this.clientId = clientId; + } + + @Override + public int getBufferSize() { + return this.bufferSize; + } + + /** + * The buffer size for this client + * @param bufferSize the buffer size + */ + public void setBufferSize(int bufferSize) { + this.bufferSize = bufferSize; + } + + @Override + public int getSocketTimeout() { + return this.socketTimeout; + } + + /** + * The socket timeout for this client + * @param socketTimeout the socket timeout + */ + public void setSocketTimeout(int socketTimeout) { + this.socketTimeout = socketTimeout; + } + + + /** + * The timeout on fetching metadata (e.g. partition leaders) + * @param fetchMetadataTimeout timeout + */ + public void setFetchMetadataTimeout(int fetchMetadataTimeout) { + this.fetchMetadataTimeout = fetchMetadataTimeout; + } + + @Override + public int getFetchMetadataTimeout() { + return this.fetchMetadataTimeout; } @Override @@ -48,11 +145,9 @@ public abstract class AbstractConfiguration implements InitializingBean, Configu return doGetBrokerAddresses(); } - protected abstract List doGetBrokerAddresses(); - @Override public List getDefaultPartitions() { - return defaultPartitions; + return this.defaultPartitions; } public void setDefaultPartitions(List defaultPartitions) { @@ -61,12 +156,19 @@ public abstract class AbstractConfiguration implements InitializingBean, Configu @Override public String getDefaultTopic() { - return defaultTopic; + return this.defaultTopic; } public void setDefaultTopic(String defaultTopic) { this.defaultTopic = defaultTopic; } + @Override + public void afterPropertiesSet() throws Exception { + Assert.isTrue(CollectionUtils.isEmpty(this.defaultPartitions) || StringUtils.isEmpty(this.defaultTopic) + , "A list of default partitions or a default topic may be specified, but not both"); + } + + protected abstract List doGetBrokerAddresses(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddress.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddress.java index 9bdcc893f7..f81dab2df7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddress.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddress.java @@ -17,7 +17,9 @@ package org.springframework.integration.kafka.core; -import org.springframework.util.StringUtils; +import org.springframework.util.Assert; + +import kafka.cluster.Broker; /** * Encapsulates the address of a Kafka broker. @@ -33,9 +35,7 @@ public class BrokerAddress { private final int port; public BrokerAddress(String host, int port) { - if (StringUtils.isEmpty(host)) { - throw new IllegalArgumentException("Host cannot be empty"); - } + Assert.hasText(host, "Host cannot be empty"); this.host = host; this.port = port; } @@ -44,6 +44,12 @@ public class BrokerAddress { this(host, DEFAULT_PORT); } + public BrokerAddress(Broker broker) { + Assert.notNull(broker, "Broker cannot be null"); + this.host = broker.host(); + this.port = broker.port(); + } + public static BrokerAddress fromAddress(String address) { String[] split = address.split(":"); if (split.length == 0 || split.length > 2) { @@ -55,20 +61,19 @@ public class BrokerAddress { else { return new BrokerAddress(split[0]); } - } public String getHost() { - return host; + return this.host; } public int getPort() { - return port; + return this.port; } @Override public int hashCode() { - return 31 * host.hashCode() + port; + return 31 * this.host.hashCode() + this.port; } @Override @@ -82,19 +87,12 @@ public class BrokerAddress { BrokerAddress brokerAddress = (BrokerAddress) o; - if (port != brokerAddress.port) { - return false; - } - if (!host.equals(brokerAddress.host)) { - return false; - } - - return true; + return this.port == brokerAddress.port && this.host.equals(brokerAddress.host); } @Override public String toString() { - return host + ":" + port; + return this.host + ":" + this.port; } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java index 73c58b7ec0..e4de8b96fb 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java @@ -14,7 +14,6 @@ * limitations under the License. */ - package org.springframework.integration.kafka.core; import java.util.List; @@ -26,6 +25,53 @@ import java.util.List; */ public interface Configuration { + /** + * The minimum amount of data that a server fetch operation will wait for before returning, + * unless {@code maxWait} has elapsed. + * In conjunction with {@link Configuration#getMaxWait()}}, controls latency + * and throughput. + * Smaller values increase responsiveness, but may increase the number of poll operations, + * potentially reducing throughput and increasing CPU consumption. + * @return the minimum amount of data for a fetch operation + */ + int getMinBytes(); + + /** + * The maximum amount of time that a server fetch operation will wait before returning + * (unless {@code minFetchSizeInBytes}) are available. + * In conjunction with {@link AbstractConfiguration#setMinBytes(int)}, + * controls latency and throughput. + * Smaller intervals increase responsiveness, but may increase + * the number of poll operations, potentially increasing CPU + * consumption and reducing throughput. + * @return the maximum wait time for a fetch operation + */ + int getMaxWait(); + + /** + * The client name to be used throughout this connection. + * @return the client id for a connection + */ + String getClientId(); + + /** + * The buffer size for this client + * @return the buffer size + */ + int getBufferSize(); + + /** + * The socket timeout for this client + * @return the socket timeout + */ + int getSocketTimeout(); + + /** + * The timeout on fetching metadata (e.g. partition leaders) + * @return the fetch metadata timeout + */ + int getFetchMetadataTimeout(); + /** * The list of seed broker addresses used by this Configuration. * @return the broker addresses diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java index 7e69a13c5e..4793f8a3bf 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java @@ -42,8 +42,7 @@ public interface Connection { * ({@link OffsetRequest#LatestTime()}) and -2 ({@link OffsetRequest#LatestTime()}) * are used . To be used to position the initial offset of a read operation. * @param referenceTime The returned values will be before this time, if they exist. The special - * values -2 ({@link OffsetRequest#LatestTime()}) - * and -1 ({@link OffsetRequest#LatestTime()}) are supported. + * values -2 ({@link OffsetRequest#LatestTime()}) and -1 ({@link OffsetRequest#LatestTime()}) are supported. * @param partitions the offsets, indexed by {@link Partition} * @return any errors, an empty {@link Result} in case of success * @throws ConsumerException the ConsumerException if any underlying error diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java index 3791ca8061..263ea17286 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java @@ -50,12 +50,12 @@ public interface ConnectionFactory { BrokerAddress getLeader(Partition partition); /** - * Refresh the broker connections and partition leader map. To be called when the topology changes + * Refresh the cached metadata (i.e. leader topology and partitions). To be called when the topology changes * are detected (i.e. brokers leave and/or partition leaders change) and that results in fetch errors, * for instance. * @param topics the topics for which to refresh the leaders */ - void refreshLeaders(Collection topics); + void refreshMetadata(Collection topics); /** * Retrieves the partitions of a given topic diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java index 922e04519f..1870c56676 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java @@ -14,35 +14,37 @@ * limitations under the License. */ - package org.springframework.integration.kafka.core; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; -import com.gs.collections.api.block.function.Function; -import com.gs.collections.impl.block.factory.Functions; -import com.gs.collections.impl.list.mutable.FastList; -import com.gs.collections.impl.map.mutable.UnifiedMap; -import com.gs.collections.impl.utility.ListIterate; -import kafka.client.ClientUtils$; -import kafka.common.KafkaException; -import kafka.javaapi.PartitionMetadata; -import kafka.javaapi.TopicMetadata; -import kafka.javaapi.TopicMetadataResponse; -import scala.collection.JavaConversions; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.SmartLifecycle; import org.springframework.util.Assert; +import com.gs.collections.api.block.function.Function; +import com.gs.collections.api.block.predicate.Predicate; +import com.gs.collections.api.partition.PartitionIterable; +import com.gs.collections.impl.block.factory.Functions; +import com.gs.collections.impl.map.mutable.UnifiedMap; +import com.gs.collections.impl.utility.Iterate; +import com.gs.collections.impl.utility.ListIterate; + +import kafka.client.ClientUtils$; +import kafka.common.ErrorMapping; +import kafka.javaapi.TopicMetadata; +import kafka.javaapi.TopicMetadataResponse; +import scala.collection.JavaConversions; + /** * Default implementation of {@link ConnectionFactory} * @@ -50,28 +52,21 @@ import org.springframework.util.Assert; */ public class DefaultConnectionFactory implements InitializingBean, ConnectionFactory, DisposableBean { + private final static Log log = LogFactory.getLog(DefaultConnectionFactory.class); + + public static final Predicate errorlessTopicMetadataPredicate = new ErrorlessTopicMetadataPredicate(); + private final GetBrokersByPartitionFunction getBrokersByPartitionFunction = new GetBrokersByPartitionFunction(); private final ConnectionInstantiationFunction connectionInstantiationFunction = new ConnectionInstantiationFunction(); private final Configuration configuration; - private final AtomicReference partitionBrokerMapReference = new AtomicReference(); + private final AtomicReference metadataCacheHolder = + new AtomicReference(new MetadataCache(Collections.emptySet())); private final ReadWriteLock lock = new ReentrantReadWriteLock(); - private String clientId = KafkaConsumerDefaults.GROUP_ID; - - private int minBytes = KafkaConsumerDefaults.MIN_FETCH_BYTES; - - private int maxWait = KafkaConsumerDefaults.MAX_WAIT_TIME_IN_MS; - - private int bufferSize = KafkaConsumerDefaults.SOCKET_BUFFER_SIZE_INT; - - private int socketTimeout = KafkaConsumerDefaults.SOCKET_TIMEOUT_INT; - - private int fetchMetadataTimeout = KafkaConsumerDefaults.FETCH_METADATA_TIMEOUT; - private final UnifiedMap kafkaBrokersCache = UnifiedMap.newMap(); public DefaultConnectionFactory(Configuration configuration) { @@ -79,92 +74,29 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac } public Configuration getConfiguration() { - return configuration; + return this.configuration; } @Override public void afterPropertiesSet() throws Exception { Assert.notNull(this.configuration, "Kafka configuration cannot be empty"); - this.refreshLeaders(configuration.getDefaultTopic() == null ? Collections.emptyList() : - Collections.singletonList(configuration.getDefaultTopic())); + refreshMetadata(this.configuration.getDefaultTopic() == null ? Collections.emptyList() : + Collections.singletonList(this.configuration.getDefaultTopic())); } @Override public void destroy() throws Exception { - for (Connection connection : kafkaBrokersCache) { + for (Connection connection : this.kafkaBrokersCache) { connection.close(); } } - /** - * The minimum amount of data that a server fetch operation will wait for before returning, - * unless {@code maxWaitTimeInMs} has elapsed. - * In conjunction with {@link DefaultConnectionFactory#setMaxWait(int)}, controls latency - * and throughput. - * Smaller values increase responsiveness, but may increase the number of poll operations, - * potentially reducing throughput and increasing CPU consumption. - * @param minBytes the amount of data to fetch - */ - public void setMinBytes(int minBytes) { - this.minBytes = minBytes; - } - - /** - * The maximum amount of time that a server fetch operation will wait before returning - * (unless {@code minFetchSizeInBytes}) are available. - * In conjunction with {@link DefaultConnectionFactory#setMinBytes(int)}, - * controls latency and throughput. - * Smaller intervals increase responsiveness, but may increase - * the number of poll operations, potentially increasing CPU - * consumption and reducing throughput. - * @param maxWait timeout to wait - */ - public void setMaxWait(int maxWait) { - this.maxWait = maxWait; - } - - public String getClientId() { - return clientId; - } - - /** - * A client name to be used throughout this connection. - * @param clientId the client name - */ - public void setClientId(String clientId) { - this.clientId = clientId; - } - - /** - * The buffer size for this client - * @param bufferSize the buffer size - */ - public void setBufferSize(int bufferSize) { - this.bufferSize = bufferSize; - } - - /** - * The socket timeout for this client - * @param socketTimeout the socket timeout - */ - public void setSocketTimeout(int socketTimeout) { - this.socketTimeout = socketTimeout; - } - - /** - * The timeout on fetching metadata (e.g. partition leaders) - * @param fetchMetadataTimeout timeout - */ - public void setFetchMetadataTimeout(int fetchMetadataTimeout) { - this.fetchMetadataTimeout = fetchMetadataTimeout; - } - /** * @see ConnectionFactory#getLeaders(Iterable) */ @Override public Map getLeaders(Iterable partitions) { - return FastList.newList(partitions).toMap(Functions.getPassThru(), getBrokersByPartitionFunction); + return Iterate.toMap(partitions, Functions.getPassThru(), getBrokersByPartitionFunction); } /** @@ -172,13 +104,32 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac */ @Override public BrokerAddress getLeader(Partition partition) { + BrokerAddress leader = null; try { - lock.readLock().lock(); - return this.getLeaders(Collections.singleton(partition)).get(partition); + this.lock.readLock().lock(); + leader = getMetadataCache().getLeader(partition); } finally { - lock.readLock().unlock(); + this.lock.readLock().unlock(); } + if (leader == null) { + try { + this.lock.writeLock().lock(); + // double lock check + leader = getMetadataCache().getLeader(partition); + if (leader == null) { + refreshMetadata(Collections.singleton(partition.getTopic())); + leader = getMetadataCache().getLeader(partition); + } + } + finally { + this.lock.writeLock().unlock(); + } + } + if (leader == null) { + throw new PartitionNotFoundException(partition); + } + return leader; } /** @@ -186,39 +137,38 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac */ @Override public Connection connect(BrokerAddress brokerAddress) { - return kafkaBrokersCache.getIfAbsentPutWithKey(brokerAddress, connectionInstantiationFunction); + return this.kafkaBrokersCache.getIfAbsentPutWithKey(brokerAddress, connectionInstantiationFunction); } /** - * @see ConnectionFactory#refreshLeaders(Collection) + * @see ConnectionFactory#refreshMetadata(Collection) */ @Override - public void refreshLeaders(Collection topics) { + public void refreshMetadata(Collection topics) { try { - lock.writeLock().lock(); - for (Connection connection : kafkaBrokersCache) { + this.lock.writeLock().lock(); + for (Connection connection : this.kafkaBrokersCache) { connection.close(); } String brokerAddressesAsString = - ListIterate.collect(configuration.getBrokerAddresses(), Functions.getToString()) - .makeString(","); + ListIterate.collect(this.configuration.getBrokerAddresses(), Functions.getToString()) + .makeString(","); TopicMetadataResponse topicMetadataResponse = new TopicMetadataResponse( ClientUtils$.MODULE$.fetchTopicMetadata( JavaConversions.asScalaSet(new HashSet(topics)), ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString), - getClientId(), fetchMetadataTimeout, 0)); - Map kafkaBrokerAddressMap = new HashMap(); - for (TopicMetadata topicMetadata : topicMetadataResponse.topicsMetadata()) { - for (PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) { - kafkaBrokerAddressMap.put(new Partition(topicMetadata.topic(), partitionMetadata.partitionId()), - new BrokerAddress(partitionMetadata.leader().host(), partitionMetadata.leader().port())); - } + this.configuration.getClientId(), this.configuration.getFetchMetadataTimeout(), 0)); + PartitionIterable selectWithoutErrors = Iterate.partition(topicMetadataResponse.topicsMetadata(), + errorlessTopicMetadataPredicate); + this.metadataCacheHolder.set(this.metadataCacheHolder.get().merge(selectWithoutErrors.getSelected())); + for (TopicMetadata topicMetadata : selectWithoutErrors.getRejected()) { + log.error(String.format("No metadata could be retrieved for '%s'", topicMetadata.topic()), + ErrorMapping.exceptionFor(topicMetadata.errorCode())); } - this.partitionBrokerMapReference.set(new PartitionBrokerMap(UnifiedMap.newMap(kafkaBrokerAddressMap))); } finally { - lock.writeLock().unlock(); + this.lock.writeLock().unlock(); } } @@ -227,14 +177,48 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac */ @Override public Collection getPartitions(String topic) { - if (!getPartitionBrokerMap().getPartitionsByTopic().containsKey(topic)) { + // first, we try to read the topic from the cache. We use the read lock to block if a write is in progress + Collection returnedPartitions = null; + try { + this.lock.readLock().lock(); + returnedPartitions = getMetadataCache().getPartitions(topic); + } + finally { + this.lock.readLock().unlock(); + } + // if we got here, it means that the data was not available, we should try a refresh. The lock is reentrant + // so we will not block ourselves + if (returnedPartitions == null) { + try { + this.lock.writeLock().lock(); + // double lock check + returnedPartitions = getMetadataCache().getPartitions(topic); + if (returnedPartitions == null) { + refreshMetadata(Collections.singleton(topic)); + // if data is not available after refreshing, it means that the topic was not found + returnedPartitions = getMetadataCache().getPartitions(topic); + } + } + finally { + lock.writeLock().unlock(); + } + } + if (returnedPartitions == null) { throw new TopicNotFoundException(topic); } - return getPartitionBrokerMap().getPartitionsByTopic().get(topic).toList(); + return returnedPartitions; } - private PartitionBrokerMap getPartitionBrokerMap() { - return partitionBrokerMapReference.get(); + private MetadataCache getMetadataCache() { + return this.metadataCacheHolder.get(); + } + + @SuppressWarnings("serial") + private static class ErrorlessTopicMetadataPredicate implements Predicate { + @Override + public boolean accept(TopicMetadata topicMetadata) { + return topicMetadata.errorCode() == ErrorMapping.NoError(); + } } @SuppressWarnings("serial") @@ -242,7 +226,12 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac @Override public Connection valueOf(BrokerAddress brokerAddress) { - return new DefaultConnection(brokerAddress, clientId, bufferSize, socketTimeout, minBytes, maxWait); + return new DefaultConnection(brokerAddress, + DefaultConnectionFactory.this.configuration.getClientId(), + DefaultConnectionFactory.this.configuration.getBufferSize(), + DefaultConnectionFactory.this.configuration.getSocketTimeout(), + DefaultConnectionFactory.this.configuration.getMinBytes(), + DefaultConnectionFactory.this.configuration.getMaxWait()); } } @@ -252,7 +241,7 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac @Override public BrokerAddress valueOf(Partition partition) { - return partitionBrokerMapReference.get().getBrokersByPartition().get(partition); + return metadataCacheHolder.get().getLeader(partition); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/MetadataCache.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/MetadataCache.java new file mode 100644 index 0000000000..39fe907c7f --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/MetadataCache.java @@ -0,0 +1,111 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.core; + +import java.util.Collection; +import java.util.Map; + +import com.gs.collections.api.block.function.Function; +import com.gs.collections.impl.map.mutable.UnifiedMap; +import com.gs.collections.impl.utility.Iterate; +import kafka.javaapi.PartitionMetadata; +import kafka.javaapi.TopicMetadata; + +/** + * Immutable store for the partition/broker mapping + * + * @author Marius Bogoevici + */ +class MetadataCache { + + public static final GetTopicNameFunction getTopicNameFunction = new GetTopicNameFunction(); + + public static final ToIndexedPartitionMetadataFunction toIndexedPartitionMetadataFunction = + new ToIndexedPartitionMetadataFunction(); + + private final Map> metadatasByTopic; + + public MetadataCache(Iterable topicMetadatas) { + UnifiedMap> topicData = UnifiedMap.newMap(); + for (TopicMetadata topicMetadata : topicMetadatas) { + Map partitionData = toPartitionMetadataMap(topicMetadata); + topicData.put(topicMetadata.topic(), partitionData); + } + this.metadatasByTopic = topicData.toImmutable().castToMap(); + } + + private MetadataCache(Map> metadatasByTopic) { + this.metadatasByTopic = metadatasByTopic; + } + + public MetadataCache merge(final Iterable topicMetadatas) { + UnifiedMap> unifiedMap = UnifiedMap.newMap(this.metadatasByTopic); + return new MetadataCache(Iterate.addToMap(topicMetadatas, getTopicNameFunction, + toIndexedPartitionMetadataFunction, unifiedMap).toImmutable().castToMap()); + } + + public Collection getPartitions(String topic) { + if (this.metadatasByTopic.containsKey(topic)) { + return this.metadatasByTopic.get(topic).keySet(); + } + else { + return null; + } + } + + public BrokerAddress getLeader(Partition partition) { + if (!this.metadatasByTopic.containsKey(partition.getTopic())) { + return null; + } + Map partitionMetadatasForTopic = this.metadatasByTopic.get(partition.getTopic()); + if (!partitionMetadatasForTopic.containsKey(partition)) { + return null; + } + return new BrokerAddress(partitionMetadatasForTopic.get(partition).leader()); + } + + + private static Map toPartitionMetadataMap(TopicMetadata topicMetadata) { + Map partitionData = UnifiedMap.newMap(); + for (PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) { + partitionData.put(new Partition(topicMetadata.topic(), partitionMetadata.partitionId()), partitionMetadata); + } + return partitionData; + } + + @SuppressWarnings("serial") + private static class GetTopicNameFunction implements Function { + + @Override + public String valueOf(TopicMetadata object) { + return object.topic(); + } + + } + + @SuppressWarnings("serial") + private static class ToIndexedPartitionMetadataFunction implements + Function> { + + @Override + public Map valueOf(TopicMetadata object) { + return toPartitionMetadataMap(object); + } + + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Partition.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Partition.java index db19f4b3c7..819fdcaf55 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Partition.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Partition.java @@ -16,6 +16,8 @@ package org.springframework.integration.kafka.core; +import org.springframework.util.Assert; + /** * A reference to a Kafka partition, with both topic and partition id * @@ -28,6 +30,8 @@ public class Partition { private int id; public Partition(String topic, int id) { + Assert.hasText(topic, "Topic name cannot be empty"); + Assert.isTrue(id >= 0, "Partition id must be greater than or equal to 0"); this.topic = topic; this.id = id; } @@ -67,10 +71,7 @@ public class Partition { if (id != partition.id) { return false; } - if (!topic.equals(partition.topic)) { - return false; - } - return true; + return topic.equals(partition.topic); } @Override diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionBrokerMap.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionBrokerMap.java deleted file mode 100644 index 6a7c0276a3..0000000000 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionBrokerMap.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2015 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.kafka.core; - -import com.gs.collections.api.block.function.Function; -import com.gs.collections.api.map.ImmutableMap; -import com.gs.collections.api.multimap.Multimap; -import com.gs.collections.impl.map.mutable.UnifiedMap; - -/** - * Immutable store for the partition/broker mapping - * - * @author Marius Bogoevici - */ -class PartitionBrokerMap { - - private static final GetTopicFunction getTopicFunction = new GetTopicFunction(); - - private final Multimap partitionsByBroker; - - private final ImmutableMap brokersByPartition; - - private final Multimap partitionsByTopic; - - public PartitionBrokerMap(UnifiedMap brokersByPartition) { - this.brokersByPartition = brokersByPartition.toImmutable(); - this.partitionsByTopic = this.brokersByPartition.keysView().groupBy(getTopicFunction); - this.partitionsByBroker = brokersByPartition.flip().toImmutable(); - } - - public Multimap getPartitionsByBroker() { - return partitionsByBroker; - } - - public ImmutableMap getBrokersByPartition() { - return brokersByPartition; - } - - public Multimap getPartitionsByTopic() { - return partitionsByTopic; - } - - @SuppressWarnings("serial") - private static class GetTopicFunction implements Function { - - @Override - public String valueOf(Partition partition) { - return partition.getTopic(); - } - - } -} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionNotFoundException.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionNotFoundException.java new file mode 100644 index 0000000000..5c1f55c947 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionNotFoundException.java @@ -0,0 +1,30 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.core; + +/** + * @author Marius Bogoevici + */ +@SuppressWarnings("serial") +public class PartitionNotFoundException extends ConsumerException { + + public PartitionNotFoundException(Partition partition) { + super(String.format("Partition [%s,%d] has no leader or has not been found", + partition.getTopic(), partition.getId())); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ZookeeperConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ZookeeperConfiguration.java index 5acca80c33..543ac4e7d6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ZookeeperConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ZookeeperConfiguration.java @@ -18,20 +18,22 @@ package org.springframework.integration.kafka.core; import java.util.List; -import com.gs.collections.api.block.function.Function; -import com.gs.collections.impl.list.mutable.FastList; -import kafka.cluster.Broker; -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils$; import org.I0Itec.zkclient.ZkClient; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import scala.collection.JavaConversions; -import scala.collection.Seq; import org.springframework.beans.factory.BeanInitializationException; import org.springframework.integration.kafka.support.ZookeeperConnect; +import com.gs.collections.api.block.function.Function; +import com.gs.collections.impl.list.mutable.FastList; + +import kafka.cluster.Broker; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils$; +import scala.collection.JavaConversions; +import scala.collection.Seq; + /** * Kafka {@link Configuration} that uses a ZooKeeper connection for retrieving the list of seed brokers. * @@ -49,6 +51,10 @@ public class ZookeeperConfiguration extends AbstractConfiguration { private int connectionTimeout; + public ZookeeperConfiguration(String zookeeperConnectionString) { + this(new ZookeeperConnect(zookeeperConnectionString)); + } + public ZookeeperConfiguration(ZookeeperConnect zookeeperConnect) { this.zookeeperServers = zookeeperConnect.getZkConnect(); try { @@ -69,7 +75,8 @@ public class ZookeeperConfiguration extends AbstractConfiguration { protected List doGetBrokerAddresses() { ZkClient zkClient = null; try { - zkClient = new ZkClient(zookeeperServers, sessionTimeout, connectionTimeout, ZKStringSerializer$.MODULE$); + zkClient = new ZkClient(this.zookeeperServers, this.sessionTimeout, this.connectionTimeout, + ZKStringSerializer$.MODULE$); Seq allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient); FastList brokers = FastList.newList(JavaConversions.asJavaCollection(allBrokersInCluster)); return brokers.collect(brokerToBrokerAddressFunction); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java new file mode 100644 index 0000000000..b27d501d71 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java @@ -0,0 +1,165 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.listener; + +import java.io.IOException; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.integration.kafka.core.Connection; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.ConsumerException; +import org.springframework.integration.kafka.core.KafkaConsumerDefaults; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.PartitionNotFoundException; +import org.springframework.integration.kafka.core.Result; +import org.springframework.util.Assert; + +import kafka.common.ErrorMapping; + +/** + * Base implementation for {@link OffsetManager}. Subclasses may customize functionality as necessary. + * + * @author Marius Bogoevici + */ +public abstract class AbstractOffsetManager implements OffsetManager, DisposableBean { + + protected final Log log = LogFactory.getLog(this.getClass()); + + protected String consumerId = KafkaConsumerDefaults.GROUP_ID; + + protected long referenceTimestamp = KafkaConsumerDefaults.DEFAULT_OFFSET_RESET; + + protected ConnectionFactory connectionFactory; + + protected Map initialOffsets; + + public AbstractOffsetManager(ConnectionFactory connectionFactory) { + this(connectionFactory, new HashMap()); + } + + public AbstractOffsetManager(ConnectionFactory connectionFactory, Map initialOffsets) { + Assert.notNull(connectionFactory, "A 'connectionFactory' can't be null"); + Assert.notNull(initialOffsets, "An initialOffsets can't be null"); + this.connectionFactory = connectionFactory; + this.initialOffsets = initialOffsets; + } + + public String getConsumerId() { + return this.consumerId; + } + + /** + * The identifier of a consumer of Kafka messages. Allows to manage offsets separately by consumer + * + * @param consumerId the consumer ID + */ + public void setConsumerId(String consumerId) { + this.consumerId = consumerId; + } + + /** + * A timestamp to be used for resetting initial offsets + * + * @param referenceTimestamp the reset timestamp for initial offsets + */ + public void setReferenceTimestamp(long referenceTimestamp) { + this.referenceTimestamp = referenceTimestamp; + } + + @Override + public void destroy() throws Exception { + try { + this.flush(); + } + catch (IOException e) { + log.error("Error while flushing the OffsetManager", e); + } + try { + this.close(); + } + catch (IOException e) { + log.error("Error while closing the OffsetManager", e); + } + } + + /** + * @see OffsetManager#updateOffset(Partition, long) + */ + @Override + public synchronized final void updateOffset(Partition partition, long offset) { + doUpdateOffset(partition, offset); + } + + /** + * @see OffsetManager#getOffset(Partition) + */ + @Override + public synchronized final long getOffset(Partition partition) { + Long storedOffset = doGetOffset(partition); + if (storedOffset == null) { + if (this.initialOffsets.containsKey(partition)) { + return this.initialOffsets.get(partition); + } + else { + BrokerAddress leader = this.connectionFactory.getLeader(partition); + if (leader == null) { + throw new PartitionNotFoundException(partition); + } + Connection connection = this.connectionFactory.connect(leader); + Result offsetResult = connection.fetchInitialOffset(this.referenceTimestamp, partition); + if (offsetResult.getErrors().size() > 0) { + throw new ConsumerException(ErrorMapping.exceptionFor(offsetResult.getError(partition))); + } + if (!offsetResult.getResults().containsKey(partition)) { + throw new IllegalStateException("Result does not contain an expected value"); + } + return offsetResult.getResult(partition); + } + } + else { + return storedOffset; + } + } + + @Override + public synchronized void resetOffsets(Collection partitionsToReset) { + // any information about those offsets is invalid and can be ignored + for (Partition partition : partitionsToReset) { + doRemoveOffset(partition); + this.initialOffsets.remove(partition); + } + } + + @Override + public synchronized void deleteOffset(Partition partition) { + doRemoveOffset(partition); + } + + protected abstract void doUpdateOffset(Partition partition, long offset); + + protected abstract void doRemoveOffset(Partition partition); + + protected abstract Long doGetOffset(Partition partition); + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 3dfab05dcc..0cac19bafd 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -238,12 +238,6 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { catch (IOException e) { log.error("Error while flushing:", e); } - try { - this.offsetManager.close(); - } - catch (IOException e) { - log.error("Error while closing:", e); - } this.messageDispatcher.stop(); } } @@ -437,7 +431,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { public void run() { FastList partitionsAsList = FastList.newList(partitionsToReset); FastList topics = partitionsAsList.collect(new PartitionToTopicFunction()).distinct(); - kafkaTemplate.getConnectionFactory().refreshLeaders(topics); + kafkaTemplate.getConnectionFactory().refreshMetadata(topics); Map leaders = kafkaTemplate.getConnectionFactory().getLeaders(partitionsToReset); synchronized (partitionsByBrokerMap) { forEachKeyValue(leaders, new AddPartitionToBrokerProcedure()); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaTopicOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaTopicOffsetManager.java new file mode 100644 index 0000000000..8089644d80 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaTopicOffsetManager.java @@ -0,0 +1,521 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.listener; + +import java.io.IOException; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.I0Itec.zkclient.ZkClient; +import org.I0Itec.zkclient.exception.ZkInterruptedException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.FetchRequest; +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.KafkaMessageBatch; +import org.springframework.integration.kafka.core.KafkaTemplate; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.Result; +import org.springframework.integration.kafka.core.TopicNotFoundException; +import org.springframework.integration.kafka.core.ZookeeperConfiguration; +import org.springframework.integration.kafka.support.ProducerFactoryBean; +import org.springframework.integration.kafka.support.ProducerMetadata; +import org.springframework.integration.kafka.support.ZookeeperConnect; +import org.springframework.integration.kafka.util.LoggingUtils; +import org.springframework.integration.kafka.util.MessageUtils; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +import kafka.admin.AdminUtils$; +import kafka.api.OffsetRequest; +import kafka.common.ErrorMapping$; +import kafka.common.TopicExistsException; +import kafka.javaapi.producer.Producer; +import kafka.producer.DefaultPartitioner; +import kafka.producer.KeyedMessage; +import kafka.serializer.Decoder; +import kafka.serializer.Encoder; +import kafka.utils.ZKStringSerializer$; + +/** + * Implementation of an {@link OffsetManager} that uses a Kafka topic as the underlying support. + * For its proper functioning, the Kafka server(s) must set {@code log.cleaner.enable=true}. It relies on the property + * {@code cleanup.policy=compact} to be set on the target topic, and if the topic is not found, + * it will create a topic with the appropriate settings. + * + * @author Marius Bogoevici + * @author Artem Bilan + */ +public class KafkaTopicOffsetManager extends AbstractOffsetManager implements InitializingBean { + + private static final KeyEncoderDecoder KEY_CODEC = new KeyEncoderDecoder(); + + private static final LongEncoderDecoder VALUE_CODEC = new LongEncoderDecoder(); + + public static final String CLEANUP_POLICY = "cleanup.policy"; + + public static final String CLEANUP_POLICY_COMPACT = "compact"; + + public static final String DELETE_RETENTION = "delete.retention.ms"; + + public static final String SEGMENT_BYTES = "segment.bytes"; + + private final ZookeeperConnect zookeeperConnect; + + private final String topic; + + private final KafkaTemplate kafkaTemplate; + + private final ConcurrentMap data = new ConcurrentHashMap(); + + private String compressionCodec = "default"; + + private Producer producer; + + private int maxSize = 10 * 1024; + + private int maxQueueBufferingTime = 1000; + + private int segmentSize = 25 * 1024; + + private int retentionTime = 60000; + + private int replicationFactor; + + private int maxBatchSize = 200; + + private boolean batchWrites = true; + + private int requiredAcks = 1; + + public KafkaTopicOffsetManager(ZookeeperConnect zookeeperConnect, String topic) { + this(zookeeperConnect, topic, new HashMap()); + } + + public KafkaTopicOffsetManager(ZookeeperConnect zookeeperConnect, String topic, + Map initialOffsets) { + super(new DefaultConnectionFactory(new ZookeeperConfiguration(zookeeperConnect)), initialOffsets); + Assert.notNull(zookeeperConnect); + this.zookeeperConnect = zookeeperConnect; + this.kafkaTemplate = new KafkaTemplate(connectionFactory); + this.topic = topic; + } + + /** + * Sets the maximum size of a fetch request, allowing to tune the initialization process. + * + * @param maxSize the maximum amount of data to be brought on a fetch + */ + public void setMaxSize(int maxSize) { + this.maxSize = maxSize; + } + + /** + * The compression codec for writing to the offset topic + * + * @param compressionCodec the compression codec + */ + public void setCompressionCodec(String compressionCodec) { + this.compressionCodec = compressionCodec; + } + + /** + * For how long will producers buffer data before writing to the topic + * + * @param maxQueueBufferingTime the maximum buffering window (in milliseconds) + */ + public void setMaxQueueBufferingTime(int maxQueueBufferingTime) { + this.maxQueueBufferingTime = maxQueueBufferingTime; + } + + /** + * The size of a segment in the offset topic + * + * @param segmentSize the segment size of an offset topic + */ + public void setSegmentSize(int segmentSize) { + this.segmentSize = segmentSize; + } + + /** + * How long are dead records retained in the offset topic + * + * @param retentionTime the retention time for dead records (in seconds) + */ + public void setRetentionTime(int retentionTime) { + this.retentionTime = retentionTime; + } + + /** + * The replication factor of the offset topic + * + * @param replicationFactor the replication factor + */ + public void setReplicationFactor(int replicationFactor) { + this.replicationFactor = replicationFactor; + } + + /** + * The maximum batch size for offset writes + * + * @param maxBatchSize maximum batching window + */ + public void setMaxBatchSize(int maxBatchSize) { + this.maxBatchSize = maxBatchSize; + } + + /** + * Whether offset writes should be batched or not + * + * @param batchWrites true if writes are batched + */ + public void setBatchWrites(boolean batchWrites) { + this.batchWrites = batchWrites; + } + + /** + * The number of required acks on write operations + * + * @param requiredAcks the number of required acks + */ + public void setRequiredAcks(int requiredAcks) { + this.requiredAcks = requiredAcks; + } + + @Override + public void afterPropertiesSet() throws Exception { + ((DefaultConnectionFactory) this.connectionFactory).afterPropertiesSet(); + ZkClient zkClient = new ZkClient(this.zookeeperConnect.getZkConnect(), + Integer.parseInt(this.zookeeperConnect.getZkSessionTimeout()), + Integer.parseInt(this.zookeeperConnect.getZkConnectionTimeout()), + ZKStringSerializer$.MODULE$); + + try { + createCompactedTopicIfNotFound(zkClient); + validateOffsetTopic(zkClient); + Partition offsetPartition = new Partition(this.topic, 0); + BrokerAddress offsetPartitionLeader = this.connectionFactory.getLeader(offsetPartition); + readOffsetData(offsetPartition, offsetPartitionLeader); + initializeProducer(offsetPartitionLeader); + } + finally { + try { + zkClient.close(); + } + catch (ZkInterruptedException e) { + log.error("Error while closing Zookeeper client", e); + } + } + } + + @Override + protected void doUpdateOffset(Partition partition, long offset) { + this.data.put(partition, offset); + this.producer.send(new KeyedMessage(this.topic, new Key(this.consumerId, partition), offset)); + } + + @Override + protected void doRemoveOffset(Partition partition) { + this.data.remove(partition); + this.producer.send(new KeyedMessage(this.topic, new Key(this.consumerId, partition), null)); + } + + @Override + protected Long doGetOffset(Partition partition) { + return this.data.get(partition); + } + + @Override + public void flush() throws IOException { + // not supported + } + + @Override + public void close() throws IOException { + this.producer.close(); + try { + ((DefaultConnectionFactory) this.connectionFactory).destroy(); + } + catch (Exception e) { + throw new IOException(e); + } + } + + + private void createCompactedTopicIfNotFound(ZkClient zkClient) { + Properties topicConfig = new Properties(); + topicConfig.setProperty(CLEANUP_POLICY, CLEANUP_POLICY_COMPACT); + topicConfig.setProperty(DELETE_RETENTION, String.valueOf(this.retentionTime)); + topicConfig.setProperty(SEGMENT_BYTES, String.valueOf(this.segmentSize)); + try { + this.replicationFactor = 1; + AdminUtils$.MODULE$.createTopic(zkClient, this.topic, 1, this.replicationFactor, topicConfig); + } + catch (TopicExistsException e) { + log.debug("Topic already exists", e); + } + } + + private void validateOffsetTopic(ZkClient zkClient) throws Exception { + //validate that the topic exists, but also prevent working with the topic until it's fully initialized + // set a retry template, since operations may fail + RetryTemplate retryValidateTopic = new RetryTemplate(); + retryValidateTopic.setRetryPolicy(new SimpleRetryPolicy(5, + Collections., Boolean>singletonMap(TopicNotFoundException.class, true))); + ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy(); + backOffPolicy.setInitialInterval(50L); + backOffPolicy.setMaxInterval(1000L); + backOffPolicy.setMultiplier(2); + retryValidateTopic.setBackOffPolicy(backOffPolicy); + + Collection partitions = + retryValidateTopic.execute(new RetryCallback, Exception>() { + + @Override + public Collection doWithRetry(RetryContext context) throws Exception { + return connectionFactory.getPartitions(topic); + } + + }); + + if (partitions.size() > 1) { + throw new BeanInitializationException("Offset management topic cannot have more than one partition"); + } + + Properties properties = AdminUtils$.MODULE$.fetchTopicConfig(zkClient, this.topic); + if (!properties.containsKey(CLEANUP_POLICY) + || !CLEANUP_POLICY_COMPACT.equals(properties.getProperty(CLEANUP_POLICY))) { + // we set the property to compact, but if using an already created topic, + // we must check if it is set up correctly + throw new BeanInitializationException("Property 'cleanup.policy' must be set to 'compact' on offset topic"); + } + } + + private void readOffsetData(Partition offsetManagementPartition, BrokerAddress leader) { + Result earliestOffsetResult = this.connectionFactory.connect(leader) + .fetchInitialOffset(OffsetRequest.EarliestTime(), offsetManagementPartition); + if (earliestOffsetResult.getErrors().size() > 0) { + throw new BeanInitializationException("Cannot initialize offset manager, unable to read earliest offset", + ErrorMapping$.MODULE$.exceptionFor(earliestOffsetResult.getError(offsetManagementPartition))); + } + Result latestOffsetResult = this.connectionFactory.connect(leader) + .fetchInitialOffset(OffsetRequest.LatestTime(), offsetManagementPartition); + if (latestOffsetResult.getErrors().size() > 0) { + throw new BeanInitializationException("Cannot initialize offset manager, unable to read latest offset"); + } + + long initialOffset = earliestOffsetResult.getResult(offsetManagementPartition); + long finalOffset = latestOffsetResult.getResult(offsetManagementPartition); + + // read repeatedly until we drain the topic and add messages to the data map + long readingOffset = initialOffset; + while (readingOffset < finalOffset) { + FetchRequest fetchRequest = new FetchRequest(offsetManagementPartition, readingOffset, maxSize); + Result receive = this.kafkaTemplate.receive(Collections.singleton(fetchRequest)); + if (receive.getErrors().size() > 0) { + throw new BeanInitializationException("Error while fetching initial offsets:", + ErrorMapping$.MODULE$.exceptionFor(receive.getError(offsetManagementPartition))); + } + KafkaMessageBatch result = receive.getResult(offsetManagementPartition); + for (KafkaMessage kafkaMessage : result.getMessages()) { + checkAndAddData(kafkaMessage); + readingOffset = kafkaMessage.getMetadata().getNextOffset(); + } + if (log.isDebugEnabled()) { + log.debug(data.size() + " entries in the final map"); + } + if (log.isTraceEnabled()) { + for (Map.Entry dataEntry : data.entrySet()) { + log.trace(String.format("Final value for %s : %s", dataEntry.getKey().toString(), + String.valueOf(dataEntry.getValue()))); + } + } + } + } + + private void checkAndAddData(KafkaMessage kafkaMessage) { + Key key = MessageUtils.decodeKey(kafkaMessage, KEY_CODEC); + Long value = MessageUtils.decodePayload(kafkaMessage, VALUE_CODEC); + if (log.isTraceEnabled()) { + log.trace("Loading key " + key + " with value " + value); + } + // we are only interested for messages that are intended for this consumer id + if (key != null && ObjectUtils.nullSafeEquals(this.consumerId, key.getConsumerId())) { + if (null != value) { + // write data in the cache, overwriting the older values + this.data.put(key.getPartition(), value); + } + else { + // a null value means that the data has been deleted, but not compacted yet + if (this.data.containsKey(key.getPartition())) { + this.data.remove(key.getPartition()); + } + } + } + } + + + private void initializeProducer(BrokerAddress leader) throws Exception { + ProducerMetadata producerMetadata = new ProducerMetadata(this.topic); + + producerMetadata.setValueEncoder(VALUE_CODEC); + producerMetadata.setValueClassType(Long.class); + producerMetadata.setKeyEncoder(KEY_CODEC); + producerMetadata.setKeyClassType(Key.class); + producerMetadata.setPartitioner(new DefaultPartitioner(null)); + + Properties additionalProps = new Properties(); + + producerMetadata.setAsync(this.batchWrites); + if (this.batchWrites) { + producerMetadata.setBatchNumMessages(String.valueOf(this.maxBatchSize)); + producerMetadata.setCompressionCodec(this.compressionCodec); + additionalProps.put("request.required.acks", String.valueOf(this.requiredAcks)); + additionalProps.put("queue.buffering.max.ms", String.valueOf(this.maxQueueBufferingTime)); + } + + ProducerFactoryBean producerFB + = new ProducerFactoryBean(producerMetadata, leader.toString(), additionalProps); + + this.producer = producerFB.getObject(); + } + + private static int intFromBytes(byte[] bytes, int start) { + return bytes[start] << 24 | (bytes[start + 1] & 0xFF) << 16 + | (bytes[start + 2] & 0xFF) << 8 | (bytes[start + 3] & 0xFF); + } + + private static byte[] intToBytes(Integer message) { + int value = message; + return new byte[] { + (byte) (value >>> 24), + (byte) (value >>> 16), + (byte) (value >>> 8), + (byte) value + }; + } + + /** + * Wraps the partition and consumer information and will be used as a key on the Kafka topic + */ + public static class Key { + + String consumerId; + + Partition partition; + + public Key(String consumerID, Partition partition) { + Assert.notNull(consumerID, "Consumer Id cannot be null"); + Assert.notNull(partition, "Partition cannot be null"); + this.consumerId = consumerID; + this.partition = partition; + } + + public String getConsumerId() { + return this.consumerId; + } + + public Partition getPartition() { + return this.partition; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + Key key = (Key) o; + + return this.consumerId.equals(key.consumerId) && this.partition.equals(key.partition); + + } + + @Override + public int hashCode() { + int result = this.consumerId.hashCode(); + result = 31 * result + this.partition.hashCode(); + return result; + } + + } + + public static class KeyEncoderDecoder implements Encoder, Decoder { + + private static final Log log = LogFactory.getLog(KeyEncoderDecoder.class); + + @Override + public Key fromBytes(byte[] bytes) { + if (bytes == null || bytes.length <= 0) { + return null; + } + try { + // calculate the offsets in the key array + int consumerIdSize = intFromBytes(bytes, 0); + int topicIdSize = intFromBytes(bytes, consumerIdSize + 4); + // reconstruct the key + return new Key(new String(bytes, 4, consumerIdSize), + new Partition(new String(bytes, consumerIdSize + 8, topicIdSize), + intFromBytes(bytes, consumerIdSize + topicIdSize + 8))); + } + catch (Exception e) { + if (log.isDebugEnabled()) { + log.debug("Cannot decode key:" + LoggingUtils.asCommaSeparatedHexDump(bytes)); + } + return null; + } + } + + @Override + public byte[] toBytes(Key key) { + if (key == null) { + return null; + } + try { + byte[] consumerIdBytes = key.consumerId.getBytes("UTF-8"); + byte[] topicNameBytes = key.partition.getTopic().getBytes("UTF-8"); + byte[] partitionIdBytes = intToBytes(key.partition.getId()); + byte[] result = new byte[4 + consumerIdBytes.length + 4 + topicNameBytes.length + 4]; + System.arraycopy(intToBytes(consumerIdBytes.length), 0, result, 0, 4); + System.arraycopy(consumerIdBytes, 0, result, 4, consumerIdBytes.length); + System.arraycopy(intToBytes(topicNameBytes.length), 0, result, consumerIdBytes.length + 4, 4); + System.arraycopy(topicNameBytes, 0, result, consumerIdBytes.length + 8, topicNameBytes.length); + System.arraycopy(partitionIdBytes, 0, result, consumerIdBytes.length + topicNameBytes.length + 8, 4); + return result; + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/LongEncoderDecoder.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/LongEncoderDecoder.java new file mode 100644 index 0000000000..83bf1fdbf5 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/LongEncoderDecoder.java @@ -0,0 +1,66 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.listener; + +import java.nio.ByteBuffer; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.kafka.util.LoggingUtils; + +import kafka.serializer.Decoder; +import kafka.serializer.Encoder; + +/** + * Kafka {@link Encoder} and {@link Decoder} for Long values. + * + * @author Marius Bogoevici + */ +public class LongEncoderDecoder implements Encoder, Decoder { + + private Log log = LogFactory.getLog(LongEncoderDecoder.class); + + @Override + public Long fromBytes(byte[] bytes) { + if (bytes == null || bytes.length <= 0) { + return null; + } + else { + try { + return ByteBuffer.wrap(bytes).getLong(0); + } + catch (Exception e) { + if (log.isDebugEnabled()) { + log.debug("Cannot decode value: " + LoggingUtils.asCommaSeparatedHexDump(bytes)); + } + return null; + } + } + } + + @Override + public byte[] toBytes(Long value) { + if (value == null) { + return null; + } + else { + return ByteBuffer.allocate(8).putLong(value).array(); + } + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/MetadataStoreOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/MetadataStoreOffsetManager.java index 92a3611ff2..3b3ed615f8 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/MetadataStoreOffsetManager.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/MetadataStoreOffsetManager.java @@ -19,21 +19,10 @@ package org.springframework.integration.kafka.listener; import java.io.Closeable; import java.io.Flushable; import java.io.IOException; -import java.util.Collection; -import java.util.HashMap; import java.util.Map; -import kafka.common.ErrorMapping; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.integration.kafka.core.BrokerAddress; -import org.springframework.integration.kafka.core.Connection; import org.springframework.integration.kafka.core.ConnectionFactory; -import org.springframework.integration.kafka.core.ConsumerException; -import org.springframework.integration.kafka.core.KafkaConsumerDefaults; import org.springframework.integration.kafka.core.Partition; -import org.springframework.integration.kafka.core.Result; import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; @@ -42,133 +31,55 @@ import org.springframework.integration.metadata.SimpleMetadataStore; * * @author Marius Bogoevici */ -public class MetadataStoreOffsetManager implements OffsetManager { - - private final static Log log = LogFactory.getLog(MetadataStoreOffsetManager.class); - - private String consumerId = KafkaConsumerDefaults.GROUP_ID; +public class MetadataStoreOffsetManager extends AbstractOffsetManager { private MetadataStore metadataStore = new SimpleMetadataStore(); - private ConnectionFactory connectionFactory; - - private Map initialOffsets; - - private long referenceTimestamp = KafkaConsumerDefaults.DEFAULT_OFFSET_RESET; - public MetadataStoreOffsetManager(ConnectionFactory connectionFactory) { - this(connectionFactory, null); + super(connectionFactory); } public MetadataStoreOffsetManager(ConnectionFactory connectionFactory, Map initialOffsets) { - this.connectionFactory = connectionFactory; - this.initialOffsets = initialOffsets == null? new HashMap() : initialOffsets; - } - - public String getConsumerId() { - return consumerId; - } - - /** - * The identifier of a consumer of Kafka messages. Allows to store separate sets of offsets in the - * {@link MetadataStore}. - * @param consumerId the consumer ID - */ - public void setConsumerId(String consumerId) { - this.consumerId = consumerId; - } - - public MetadataStore getMetadataStore() { - return metadataStore; + super(connectionFactory, initialOffsets); } /** * The backing {@link MetadataStore} for storing offsets. + * * @param metadataStore a fully configured {@link MetadataStore} instance */ public void setMetadataStore(MetadataStore metadataStore) { this.metadataStore = metadataStore; } - public long getReferenceTimestamp() { - return referenceTimestamp; - } - - /** - * A timestamp to be used for resetting initial offsets, if they are not available in the {@link MetadataStore} - * @param referenceTimestamp the reset timestamp for initial offsets - */ - public void setReferenceTimestamp(long referenceTimestamp) { - this.referenceTimestamp = referenceTimestamp; - } - - /** - * @see OffsetManager#updateOffset(Partition, long) - */ - @Override - public synchronized void updateOffset(Partition partition, long offset) { - metadataStore.put(asKey(partition), Long.toString(offset)); - } - - /** - * @see OffsetManager#getOffset(Partition) - */ - @Override - public synchronized long getOffset(Partition partition) { - Long offsetInMetadataStore = getOffsetFromMetadataStore(partition); - if (offsetInMetadataStore == null) { - if (this.initialOffsets.containsKey(partition)) { - return this.initialOffsets.get(partition); - } - else { - BrokerAddress leader = this.connectionFactory.getLeader(partition); - if (leader == null) { - throw new ConsumerException("No leader found for " + partition.toString()); - } - Connection connection = this.connectionFactory.connect(leader); - Result offsetResult = connection.fetchInitialOffset(referenceTimestamp, partition); - if (offsetResult.getErrors().size() > 0) { - throw new ConsumerException(ErrorMapping.exceptionFor(offsetResult.getError(partition))); - } - if (!offsetResult.getResults().containsKey(partition)) { - throw new IllegalStateException("Result does not contain an expected value"); - } - return offsetResult.getResult(partition); - } - } - else { - return offsetInMetadataStore; - } - } - - @Override - public synchronized void resetOffsets(Collection partitionsToReset) { - // any information about those offsets is invalid and can be ignored - for (Partition partition : partitionsToReset) { - metadataStore.remove(asKey(partition)); - initialOffsets.remove(partition); - } - } - @Override public void close() throws IOException { // Flush before closing. This may be redundant, but ensures that the metadata store is closed properly flush(); - if (metadataStore instanceof Closeable) { - ((Closeable) metadataStore).close(); + if (this.metadataStore instanceof Closeable) { + ((Closeable) this.metadataStore).close(); } } @Override public void flush() throws IOException { - if (metadataStore instanceof Flushable) { - ((Flushable) metadataStore).flush(); + if (this.metadataStore instanceof Flushable) { + ((Flushable) this.metadataStore).flush(); } } + @Override + protected void doUpdateOffset(Partition partition, long offset) { + this.metadataStore.put(generateKey(partition), Long.toString(offset)); + } - private Long getOffsetFromMetadataStore(Partition partition) { - String storedOffsetValueAsString = this.metadataStore.get(asKey(partition)); + @Override + protected void doRemoveOffset(Partition partition) { + this.metadataStore.remove(generateKey(partition)); + } + + protected Long doGetOffset(Partition partition) { + String storedOffsetValueAsString = this.metadataStore.get(generateKey(partition)); Long storedOffsetValue = null; if (storedOffsetValueAsString != null) { try { @@ -181,8 +92,8 @@ public class MetadataStoreOffsetManager implements OffsetManager { return storedOffsetValue; } - private String asKey(Partition partition) { - return partition.getTopic() + ":" + partition.getId() + ":" + consumerId; + public String generateKey(Partition partition) { + return partition.getTopic() + ":" + partition.getId() + ":" + getConsumerId(); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/OffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/OffsetManager.java index b41207bcfc..3d93c17594 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/OffsetManager.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/OffsetManager.java @@ -38,12 +38,19 @@ public interface OffsetManager extends Closeable, Flushable { /** * Retrieves the offset for a given {@link Partition} - * * @param partition the partition to be * @return the offset value */ long getOffset(Partition partition); + /** + * Removes the offset for a given {@link Partition}. Useful + * for components that need to clean up after themselves. + * @param partition for which to delete the JavaDoc + * + */ + void deleteOffset(Partition partition); + /** * Resets offsets for the given {@link Partition}s. To be invoked when the values stored are invalid, * so a client cannot resume from that position. Implementations must decide on the best strategy to follow. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java index fd66562a71..66c5b33d2e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java @@ -34,6 +34,7 @@ import org.springframework.messaging.Message; * @author Ilayaperumal Gopinathan * @author Gary Russell * @author Artem Bilan + * @author Marius Bogoevici * @since 0.5 */ public class KafkaProducerContext implements SmartLifecycle, NamedComponent, BeanNameAware { @@ -66,7 +67,6 @@ public class KafkaProducerContext implements SmartLifecycle, NamedComponen return producerConfiguration; } } - logger.error("No producer-configuration defined for topic " + topic + ". Cannot send message"); return null; } @@ -194,7 +194,7 @@ public class KafkaProducerContext implements SmartLifecycle, NamedComponen } // if there is a single producer configuration then use that config to send message. else if (this.theProducerConfiguration != null) { - this.theProducerConfiguration.send(null, messageKey, message); + this.theProducerConfiguration.send(topic, messageKey, message); } else { throw new IllegalStateException("Could not send messages as there are multiple producer configurations " + diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ZookeeperConnect.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ZookeeperConnect.java index 99c304b319..cb52e28600 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ZookeeperConnect.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ZookeeperConnect.java @@ -22,11 +22,22 @@ import org.springframework.integration.kafka.core.ZookeeperConnectDefaults; * @since 0.5 */ public class ZookeeperConnect { + private String zkConnect = ZookeeperConnectDefaults.ZK_CONNECT; + private String zkConnectionTimeout = ZookeeperConnectDefaults.ZK_CONNECTION_TIMEOUT; + private String zkSessionTimeout = ZookeeperConnectDefaults.ZK_SESSION_TIMEOUT; + private String zkSyncTime = ZookeeperConnectDefaults.ZK_SYNC_TIME; + public ZookeeperConnect() { + } + + public ZookeeperConnect(String zkConnect) { + this.zkConnect = zkConnect; + } + public String getZkConnect() { return zkConnect; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/LoggingUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/LoggingUtils.java new file mode 100644 index 0000000000..a5d775e1bf --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/LoggingUtils.java @@ -0,0 +1,45 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.util; + +/** + * Utilities for logging data + * + * @author Marius Bogoevici + */ +public class LoggingUtils { + + public static String asCommaSeparatedHexDump(byte[] bytes) { + if (bytes == null || bytes.length == 0) { + return "[]"; + } + else if (bytes.length == 1) { + return String.format("[%s]", Integer.toHexString(bytes[0])); + } + else { + StringBuilder buffer = new StringBuilder("["); + for (int i = 0; i < bytes.length - 1; i++) { + buffer.append(Integer.toHexString(bytes[i])); + buffer.append(","); + } + buffer.append(Integer.toHexString(bytes[bytes.length])); + buffer.append("]"); + return buffer.toString(); + } + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/MessageUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/MessageUtils.java index b50db52c19..94c702b0dc 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/MessageUtils.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/MessageUtils.java @@ -14,21 +14,20 @@ * limitations under the License. */ - package org.springframework.integration.kafka.util; +import org.springframework.integration.kafka.core.KafkaMessage; + import kafka.serializer.Decoder; import kafka.utils.Utils$; -import org.springframework.integration.kafka.core.KafkaMessage; - /** * @author Marius Bogoevici */ public class MessageUtils { public static T decodeKey(KafkaMessage message, Decoder decoder) { - if (message.getMessage().isNull() || !message.getMessage().hasKey()) { + if (!message.getMessage().hasKey()) { return null; } return decoder.fromBytes(Utils$.MODULE$.readBytes(message.getMessage().key())); @@ -40,4 +39,5 @@ public class MessageUtils { } return decoder.fromBytes(Utils$.MODULE$.readBytes(message.getMessage().payload())); } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/TopicUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/TopicUtils.java new file mode 100644 index 0000000000..b41b9e8b4d --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/TopicUtils.java @@ -0,0 +1,156 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.util; + +import java.util.List; +import java.util.Properties; + +import org.I0Itec.zkclient.ZkClient; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.kafka.core.TopicNotFoundException; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryPolicy; +import org.springframework.retry.backoff.ExponentialBackOffPolicy; +import org.springframework.retry.policy.CompositeRetryPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.policy.TimeoutRetryPolicy; +import org.springframework.retry.support.RetryTemplate; + +import kafka.admin.AdminUtils; +import kafka.api.TopicMetadata; +import kafka.common.ErrorMapping; +import kafka.javaapi.PartitionMetadata; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; +import scala.collection.Map; +import scala.collection.Seq; + +/** + * Utilities for interacting with Kafka topics + * + * @author Marius Bogoevici + */ +public class TopicUtils { + + private static Log log = LogFactory.getLog(TopicUtils.class); + + public static final int METADATA_VERIFICATION_TIMEOUT = 5000; + + public static final int METADATA_VERIFICATION_RETRY_ATTEMPTS = 10; + + public static final double METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER = 1.5; + + public static final int METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL = 100; + + public static final int METADATA_VERIFICATION_MAX_INTERVAL = 1000; + + /** + * Creates a topic in Kafka or validates that it exists with the requested number of partitions, + * and returns only after the topic has been fully created + * @param zkAddress the address of the Kafka ZooKeeper instance + * @param topicName the name of the topic + * @param numPartitions the number of partitions + * @param replicationFactor the replication factor + * @return {@link TopicMetadata} information for the topic + */ + public static TopicMetadata ensureTopicCreated(final String zkAddress, final String topicName, + final int numPartitions, int replicationFactor) { + + final int sessionTimeoutMs = 10000; + final int connectionTimeoutMs = 10000; + final ZkClient zkClient = new ZkClient(zkAddress, sessionTimeoutMs, connectionTimeoutMs, + ZKStringSerializer$.MODULE$); + try { + // The following is basically copy/paste from AdminUtils.createTopic() with + // createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true) + Properties topicConfig = new Properties(); + Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); + scala.collection.Map> replicaAssignment = + AdminUtils.assignReplicasToBrokers(brokerList, numPartitions, replicationFactor, -1, -1); + return ensureTopicCreated(zkClient, topicName, numPartitions, topicConfig, replicaAssignment); + + } + finally { + zkClient.close(); + } + } + + /** + * Creates a topic in Kafka and returns only after the topic has been fully and an produce metadata. + * @param zkClient an open {@link ZkClient} connection to Zookeeper + * @param topicName the name of the topic + * @param numPartitions the number of partitions for the topic + * @param topicConfig additional topic configuration properties + * @param replicaAssignment the mapping of partitions to broker + * @return {@link TopicMetadata} information for the topic + */ + public static TopicMetadata ensureTopicCreated(final ZkClient zkClient, final String topicName, + final int numPartitions, Properties topicConfig, Map> replicaAssignment) { + AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, replicaAssignment, topicConfig, + true); + + RetryTemplate retryTemplate = new RetryTemplate(); + + CompositeRetryPolicy policy = new CompositeRetryPolicy(); + TimeoutRetryPolicy timeoutRetryPolicy = new TimeoutRetryPolicy(); + timeoutRetryPolicy.setTimeout(METADATA_VERIFICATION_TIMEOUT); + SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(); + simpleRetryPolicy.setMaxAttempts(METADATA_VERIFICATION_RETRY_ATTEMPTS); + policy.setPolicies(new RetryPolicy[] {timeoutRetryPolicy, simpleRetryPolicy}); + retryTemplate.setRetryPolicy(policy); + + ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy(); + backOffPolicy.setInitialInterval(METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL); + backOffPolicy.setMultiplier(METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER); + backOffPolicy.setMaxInterval(METADATA_VERIFICATION_MAX_INTERVAL); + retryTemplate.setBackOffPolicy(backOffPolicy); + + try { + return retryTemplate.execute(new RetryCallback() { + @Override + public TopicMetadata doWithRetry(RetryContext context) throws Exception { + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkClient); + if (topicMetadata.errorCode() != ErrorMapping.NoError() || + !topicName.equals(topicMetadata.topic())) { + // downcast to Exception because that's what the error throws + throw (Exception) ErrorMapping.exceptionFor(topicMetadata.errorCode()); + } + List partitionMetadatas = + new kafka.javaapi.TopicMetadata(topicMetadata).partitionsMetadata(); + if (partitionMetadatas.size() != numPartitions) { + throw new IllegalStateException("The number of expected partitions was: " + + numPartitions + ", but " + partitionMetadatas.size() + " have been found instead"); + } + for (PartitionMetadata partitionMetadata : partitionMetadatas) { + if (partitionMetadata.errorCode() != ErrorMapping.NoError()) { + throw (Exception) ErrorMapping.exceptionFor(partitionMetadata.errorCode()); + } + } + return topicMetadata; + } + }); + } + catch (Exception e) { + log.error(String.format("Cannot retrieve metadata for topic '%s'", topicName), e); + throw new TopicNotFoundException(topicName); + } + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/core/MetadataCacheTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/core/MetadataCacheTests.java new file mode 100644 index 0000000000..44ae27d1fd --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/core/MetadataCacheTests.java @@ -0,0 +1,262 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.core; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.hasItem; +import static org.hamcrest.CoreMatchers.notNullValue; +import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.junit.Assert.assertThat; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.junit.Test; + +import kafka.api.PartitionMetadata; +import kafka.cluster.Broker; +import kafka.common.ErrorMapping; +import kafka.javaapi.TopicMetadata; +import scala.Some; +import scala.collection.Seq; +import scala.collection.mutable.HashSet; +import scala.collection.mutable.HashSet$; + +/** + * @author Marius Bogoevici + */ +public class MetadataCacheTests { + + + public static final String NONEXISTENT_TOPIC = "nonexistent"; + + public static final String EXISTING_TOPIC = "existing"; + + public static final String EXISTING_TOPIC_2 = "existing2"; + + @Test + public void testMetadataCacheEmpty() throws Exception { + MetadataCache metadataCache = new MetadataCache(Collections.emptySet()); + + // blank metadatacache is empty + assertThat(metadataCache.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(NONEXISTENT_TOPIC), nullValue()); + } + + @Test + public void testMetadataCacheInitialized() throws Exception { + ArrayList topicMetadatas = new ArrayList(); + HashSet partitionMetadatas = HashSet$.MODULE$.empty(); + partitionMetadatas.$plus$eq(new PartitionMetadata(0, asScalaOption(new Broker(0, "host0", 1000)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + partitionMetadatas.$plus$eq(new PartitionMetadata(1, asScalaOption(new Broker(0, "host1", 1001)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + topicMetadatas.add(new TopicMetadata(new kafka.api.TopicMetadata(EXISTING_TOPIC, partitionMetadatas.toSeq(), + ErrorMapping.NoError()))); + MetadataCache metadataCache = new MetadataCache(topicMetadatas); + + // Initializing MetadataCache with some topic metadatas will find the data in + assertThat(metadataCache.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC), hasSize(2)); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 0))); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 1))); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)), notNullValue()); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)).getHost(), equalTo("host0")); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)).getPort(), equalTo(1000)); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)), notNullValue()); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)).getHost(), equalTo("host1")); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)).getPort(), equalTo(1001)); + } + + @Test + public void testMetadataCacheMerged() throws Exception { + ArrayList topicMetadatas = new ArrayList(); + HashSet partitionMetadatas = HashSet$.MODULE$.empty(); + partitionMetadatas.$plus$eq(new PartitionMetadata(0, asScalaOption(new Broker(0, "host0", 1000)), emptySeq(), + emptySeq(), ErrorMapping.NoError())); + partitionMetadatas.$plus$eq(new PartitionMetadata(1, asScalaOption(new Broker(0, "host1", 1001)), emptySeq(), + emptySeq(), ErrorMapping.NoError())); + topicMetadatas.add(new TopicMetadata(new kafka.api.TopicMetadata(EXISTING_TOPIC, partitionMetadatas.toSeq(), + ErrorMapping.NoError()))); + + MetadataCache metadataCache = new MetadataCache(Collections.emptySet()); + + assertThat(metadataCache.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC), nullValue()); + + + MetadataCache newMetadataCache = metadataCache.merge(topicMetadatas); + + // merging MetadataCache with new topic metadata results in an object that also has the new data + assertThat(newMetadataCache.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(newMetadataCache.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(newMetadataCache.getPartitions(EXISTING_TOPIC), hasSize(2)); + assertThat(newMetadataCache.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 0))); + assertThat(newMetadataCache.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 1))); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)), notNullValue()); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)).getHost(), equalTo("host0")); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)).getPort(), equalTo(1000)); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)), notNullValue()); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)).getHost(), equalTo("host1")); + assertThat(newMetadataCache.getLeader(new Partition(EXISTING_TOPIC, 1)).getPort(), equalTo(1001)); + } + + @Test + public void testMetadataCacheMergedWithOverride() throws Exception { + List topicMetadatas = new ArrayList(); + HashSet partitionMetadatas = HashSet$.MODULE$.empty(); + partitionMetadatas.$plus$eq(new PartitionMetadata(0, asScalaOption(new Broker(0, "host0", 1000)), emptySeq(), + emptySeq(), ErrorMapping.NoError())); + partitionMetadatas.$plus$eq(new PartitionMetadata(1, asScalaOption(new Broker(0, "host1", 1001)), emptySeq(), + emptySeq(), ErrorMapping.NoError())); + topicMetadatas.add(new TopicMetadata(new kafka.api.TopicMetadata(EXISTING_TOPIC, partitionMetadatas.toSeq(), + ErrorMapping.NoError()))); + + MetadataCache metadataCache = new MetadataCache(Collections.emptySet()); + + //empty MetadataCache has no data + assertThat(metadataCache.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC, 0)), nullValue()); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC), nullValue()); + assertThat(metadataCache.getLeader(new Partition(EXISTING_TOPIC_2, 0)), nullValue()); + assertThat(metadataCache.getPartitions(EXISTING_TOPIC_2), nullValue()); + + MetadataCache metadataCacheWithTopic1 = metadataCache.merge(topicMetadatas); + + // merging MetadataCache with new topic metadata results in an object that also has the new data + assertThat(metadataCacheWithTopic1.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCacheWithTopic1.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCacheWithTopic1.getPartitions(EXISTING_TOPIC), hasSize(2)); + assertThat(metadataCacheWithTopic1.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 0))); + assertThat(metadataCacheWithTopic1.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 1))); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 0)), notNullValue()); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 0)).getHost(), equalTo("host0")); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 0)).getPort(), equalTo(1000)); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 1)), notNullValue()); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 1)).getHost(), equalTo("host1")); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 1)).getPort(), equalTo(1001)); + assertThat(metadataCacheWithTopic1.getLeader(new Partition(EXISTING_TOPIC, 2)), nullValue()); + + ArrayList topic2Metadatas = new ArrayList(); + HashSet topic2partitionMetadatas = HashSet$.MODULE$.empty(); + topic2partitionMetadatas.$plus$eq(new PartitionMetadata(0, asScalaOption(new Broker(0, "host1", 1002)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + topic2partitionMetadatas.$plus$eq(new PartitionMetadata(1, asScalaOption(new Broker(0, "host2", 1003)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + topic2partitionMetadatas.$plus$eq(new PartitionMetadata(2, asScalaOption(new Broker(0, "host3", 1004)), emptySeq(), + emptySeq(), ErrorMapping.NoError())); + topic2Metadatas.add(new TopicMetadata(new kafka.api.TopicMetadata(EXISTING_TOPIC_2, topic2partitionMetadatas.toSeq(), + ErrorMapping.NoError()))); + MetadataCache metadataCacheWithTopic2 = metadataCacheWithTopic1.merge(topic2Metadatas); + + + assertThat(metadataCacheWithTopic2.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCacheWithTopic2.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC), hasSize(2)); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 0))); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 1))); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 0)), notNullValue()); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 0)).getHost(), equalTo("host0")); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 0)).getPort(), equalTo(1000)); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 1)), notNullValue()); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 1)).getHost(), equalTo("host1")); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 1)).getPort(), equalTo(1001)); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC, 2)), nullValue()); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC_2), notNullValue()); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC_2), hasSize(3)); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 0))); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 1))); + assertThat(metadataCacheWithTopic2.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 2))); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 0)), notNullValue()); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 0)).getHost(), equalTo("host1")); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 0)).getPort(), equalTo(1002)); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 1)), notNullValue()); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 1)).getHost(), equalTo("host2")); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 1)).getPort(), equalTo(1003)); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 2)), notNullValue()); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 2)).getHost(), equalTo("host3")); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 2)).getPort(), equalTo(1004)); + assertThat(metadataCacheWithTopic2.getLeader(new Partition(EXISTING_TOPIC_2, 3)), nullValue()); + + // merging overrides data + List overridingTopicMetadatas = new ArrayList(); + HashSet overridingPartitionsMetadatas = HashSet$.MODULE$.empty(); + overridingPartitionsMetadatas.$plus$eq(new PartitionMetadata(0, asScalaOption(new Broker(0, "host10", 2000)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + overridingPartitionsMetadatas.$plus$eq(new PartitionMetadata(1, asScalaOption(new Broker(0, "host11", 2001)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + overridingPartitionsMetadatas.$plus$eq(new PartitionMetadata(2, asScalaOption(new Broker(0, "host12", 2002)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + overridingPartitionsMetadatas.$plus$eq(new PartitionMetadata(3, asScalaOption(new Broker(0, "host13", 2003)), + emptySeq(), emptySeq(), ErrorMapping.NoError())); + overridingTopicMetadatas.add(new TopicMetadata(new kafka.api.TopicMetadata(EXISTING_TOPIC, + overridingPartitionsMetadatas.toSeq(), ErrorMapping.NoError()))); + + MetadataCache metadataCacheOverridden = metadataCacheWithTopic2.merge(overridingTopicMetadatas); + + assertThat(metadataCacheOverridden.getLeader(new Partition(NONEXISTENT_TOPIC, 0)), nullValue()); + assertThat(metadataCacheOverridden.getPartitions(NONEXISTENT_TOPIC), nullValue()); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC), hasSize(4)); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 0))); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 1))); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 2))); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC), hasItem(new Partition(EXISTING_TOPIC, 3))); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 0)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 0)).getHost(), equalTo("host10")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 0)).getPort(), equalTo(2000)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 1)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 1)).getHost(), equalTo("host11")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 1)).getPort(), equalTo(2001)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 2)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 2)).getHost(), equalTo("host12")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 2)).getPort(), equalTo(2002)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 3)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 3)).getHost(), equalTo("host13")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 3)).getPort(), equalTo(2003)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC, 4)), nullValue()); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC_2), hasSize(3)); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 0))); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 1))); + assertThat(metadataCacheOverridden.getPartitions(EXISTING_TOPIC_2), hasItem(new Partition(EXISTING_TOPIC_2, 2))); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 0)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 0)).getHost(), equalTo("host1")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 0)).getPort(), equalTo(1002)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 1)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 1)).getHost(), equalTo("host2")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 1)).getPort(), equalTo(1003)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 2)), notNullValue()); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 2)).getHost(), equalTo("host3")); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 2)).getPort(), equalTo(1004)); + assertThat(metadataCacheOverridden.getLeader(new Partition(EXISTING_TOPIC_2, 3)), nullValue()); + + } + + public Seq emptySeq() { + return HashSet$.MODULE$.empty().toSeq(); + } + + public Some asScalaOption(T object) { + return new Some(object); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java index dcfa537e1f..fe0782e817 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java @@ -14,15 +14,26 @@ * limitations under the License. */ - package org.springframework.integration.kafka.listener; +import static org.springframework.integration.kafka.util.TopicUtils.ensureTopicCreated; import static scala.collection.JavaConversions.asScalaBuffer; import java.util.ArrayList; import java.util.List; import java.util.Properties; +import org.I0Itec.zkclient.ZkClient; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; + +import org.springframework.integration.kafka.core.BrokerAddressListConfiguration; +import org.springframework.integration.kafka.core.Configuration; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.rule.KafkaRule; + import com.gs.collections.api.RichIterable; import com.gs.collections.api.block.function.Function2; import com.gs.collections.api.multimap.Multimap; @@ -30,27 +41,19 @@ import com.gs.collections.api.multimap.MutableMultimap; import com.gs.collections.api.tuple.Pair; import com.gs.collections.impl.factory.Multimaps; import com.gs.collections.impl.tuple.Tuples; + import kafka.admin.AdminUtils; import kafka.producer.KeyedMessage; import kafka.producer.Producer; import kafka.producer.ProducerConfig; import kafka.serializer.StringEncoder; import kafka.utils.TestUtils; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.After; import scala.collection.JavaConversions; import scala.collection.Map; import scala.collection.immutable.List$; import scala.collection.immutable.Map$; import scala.collection.immutable.Seq; -import org.springframework.integration.kafka.core.Configuration; -import org.springframework.integration.kafka.core.ConnectionFactory; -import org.springframework.integration.kafka.core.DefaultConnectionFactory; -import org.springframework.integration.kafka.core.BrokerAddressListConfiguration; -import org.springframework.integration.kafka.rule.KafkaRule; - /** * @author Marius Bogoevici */ @@ -67,32 +70,30 @@ public abstract class AbstractBrokerTests { deleteTopic(TEST_TOPIC); } - @SuppressWarnings("unchecked") public void createTopic(String topicName, int partitionCount, int brokers, int replication) { - MutableMultimap partitionDistribution = createPartitionDistribution(partitionCount, brokers, replication); - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(getKafkaRule().getZkClient(), - topicName, toKafkaPartitionMap(partitionDistribution), - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK$default$4(), - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK$default$5()); - if (getKafkaRule().isEmbedded()) { - for (int i = 0; i < partitionDistribution.keysView().size(); i++) { - TestUtils.waitUntilMetadataIsPropagated(asScalaBuffer(getKafkaRule().getKafkaServers()), topicName, i, 5000L); - } - } else { - sleep(partitionCount * 200); - } + createTopic(getKafkaRule().getZkClient(), topicName, partitionCount, brokers, replication); + } + + @SuppressWarnings("unchecked") + public void createTopic(ZkClient zkClient, String topicName, int partitionCount, int brokers, int replication) { + MutableMultimap partitionDistribution = + createPartitionDistribution(partitionCount, brokers, replication); + ensureTopicCreated(zkClient, topicName, partitionCount, new Properties(), + toKafkaPartitionMap(partitionDistribution)); } public void deleteTopic(String topicName) { AdminUtils.deleteTopic(getKafkaRule().getZkClient(), topicName); if (getKafkaRule().isEmbedded()) { TestUtils.waitUntilMetadataIsPropagated(asScalaBuffer(getKafkaRule().getKafkaServers()), topicName, 0, 5000L); - } else { + } + else { sleep(1000); } } - public MutableMultimap createPartitionDistribution(int partitionCount, int brokers, int replication) { + public MutableMultimap createPartitionDistribution(int partitionCount, int brokers, + int replication) { MutableMultimap partitionDistribution = Multimaps.mutable.list.with(); for (int i = 0; i < partitionCount; i++) { for (int j = 0; j < replication; j++) { @@ -108,12 +109,13 @@ public abstract class AbstractBrokerTests { } public static scala.collection.Seq> createMessages(int count, String topic) { - return createMessagesInRange(0,count-1,topic); + return createMessagesInRange(0, count - 1, topic); } - public static scala.collection.Seq> createMessagesInRange(int start, int end, String topic) { - List> messages = new ArrayList>(); - for (int i=start; i<= end; i++) { + public static scala.collection.Seq> createMessagesInRange(int start, int end, + String topic) { + List> messages = new ArrayList>(); + for (int i = start; i <= end; i++) { messages.add(new KeyedMessage(topic, "Key " + i, i, "Message " + i)); } return asScalaBuffer(messages).toSeq(); @@ -123,8 +125,8 @@ public abstract class AbstractBrokerTests { Properties producerConfig = TestUtils.getProducerConfig(getKafkaRule().getBrokersAsString(), TestPartitioner.class.getCanonicalName()); producerConfig.put("serializer.class", StringEncoder.class.getCanonicalName()); - producerConfig.put("key.serializer.class", StringEncoder.class.getCanonicalName()); - producerConfig.put("compression.codec", Integer.toString(compression)); + producerConfig.put("key.serializer.class", StringEncoder.class.getCanonicalName()); + producerConfig.put("compression.codec", Integer.toString(compression)); return new Producer(new ProducerConfig(producerConfig)); } @@ -136,12 +138,16 @@ public abstract class AbstractBrokerTests { @SuppressWarnings({"rawtypes", "serial", "deprecation"}) private Map toKafkaPartitionMap(Multimap partitions) { - java.util.Map> m = partitions.toMap().collect(new Function2, Pair>>() { - @Override - public Pair> value(Integer argument1, RichIterable argument2) { - return Tuples.pair((Object) argument1, List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq()); - } - }); + java.util.Map> m = partitions.toMap() + .collect(new Function2, Pair>>() { + + @Override + public Pair> value(Integer argument1, RichIterable argument2) { + return Tuples.pair((Object) argument1, + List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq()); + } + + }); return Map$.MODULE$.apply(JavaConversions.asScalaMap(m).toSeq()); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests.java new file mode 100644 index 0000000000..f2de815521 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests.java @@ -0,0 +1,135 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.listener; + +import static org.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.hamcrest.core.IsEqual.equalTo; +import static org.junit.Assert.assertThat; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.kafka.rule.KafkaRule; +import org.springframework.integration.kafka.serializer.common.StringDecoder; +import org.springframework.integration.kafka.support.KafkaHeaders; +import org.springframework.integration.kafka.support.ZookeeperConnect; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; + +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; + +import kafka.message.NoCompressionCodec$; + +/** + * @author Marius Bogoevici + */ +public class KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests extends AbstractMessageListenerContainerTests { + + @Rule + public final KafkaEmbedded kafkaEmbeddedBrokerRule = new KafkaEmbedded(1); + + @Override + public KafkaRule getKafkaRule() { + return kafkaEmbeddedBrokerRule; + } + + @Test + public void testKafkaOffsetManager() throws Exception { + createTopic(TEST_TOPIC, 5, 1, 1); + + ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory(); + ArrayList readPartitions = new ArrayList(); + for (int i = 0; i < 5; i++) { + readPartitions.add(new Partition(TEST_TOPIC, i)); + } + + final KafkaMessageListenerContainer kafkaMessageListenerContainer = + new KafkaMessageListenerContainer(connectionFactory, + readPartitions.toArray(new Partition[readPartitions.size()])); + KafkaTopicOffsetManager offsetManager = new KafkaTopicOffsetManager( + new ZookeeperConnect(getKafkaRule().getZookeeperConnectionString()), "si-offsets"); + offsetManager.afterPropertiesSet(); + kafkaMessageListenerContainer.setOffsetManager( + offsetManager); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(2); + + int expectedMessageCount = 100; + + final MutableListMultimap receivedData = + new SynchronizedPutFastListMultimap(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + + KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter(kafkaMessageListenerContainer); + + StringDecoder decoder = new StringDecoder(); + kafkaMessageDrivenChannelAdapter.setKeyDecoder(decoder); + kafkaMessageDrivenChannelAdapter.setPayloadDecoder(decoder); + kafkaMessageDrivenChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + kafkaMessageDrivenChannelAdapter.setOutputChannel(new FixedSubscriberChannel(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + receivedData.put( + (Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID), + new KeyedMessageWithOffset( + (String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY), + (String) message.getPayload(), + (Long) message.getHeaders().get(KafkaHeaders.OFFSET), + Thread.currentThread().getName(), + (Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID))); + latch.countDown(); + } + + })); + + kafkaMessageDrivenChannelAdapter.afterPropertiesSet(); + kafkaMessageDrivenChannelAdapter.start(); + + createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(expectedMessageCount, TEST_TOPIC)); + + Thread.sleep(100); + + kafkaMessageDrivenChannelAdapter.stop(); + kafkaMessageDrivenChannelAdapter.start(); + + latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES); + kafkaMessageListenerContainer.stop(); + + assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); + assertThat(latch.getCount(), equalTo(0L)); + System.out.println("All messages received ... checking "); + + validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1); + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java index caeca2a2fa..87c9a88342 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java @@ -215,6 +215,11 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra offsetManager.close(); } + @Override + public void deleteOffset(Partition partition) { + offsetManager.deleteOffset(partition); + } + @Override public void resetOffsets(Collection partitionsToReset) { System.out.println("resetting " + offsetResetLatch.getCount() + " partitions: " diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/AbstractOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/AbstractOffsetManagerTests.java new file mode 100644 index 0000000000..338a527761 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/AbstractOffsetManagerTests.java @@ -0,0 +1,295 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.offset; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertThat; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.integration.kafka.core.DefaultConnectionFactory; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.PartitionNotFoundException; +import org.springframework.integration.kafka.core.ZookeeperConfiguration; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.kafka.listener.TestPartitioner; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.kafka.rule.KafkaRule; +import org.springframework.integration.kafka.util.TopicUtils; + +import kafka.api.OffsetRequest; +import kafka.javaapi.producer.Producer; +import kafka.producer.KeyedMessage; +import kafka.producer.ProducerConfig; +import kafka.serializer.StringEncoder; +import kafka.utils.TestUtils; + +/** + * @author Marius Bogoevici + */ +public abstract class AbstractOffsetManagerTests { + + @Rule + public KafkaRule kafkaRule = new KafkaEmbedded(1); + + private String TEST_TOPIC = "si_test_topic"; + + @Test(expected = PartitionNotFoundException.class) + public void testFailureWhenPartitionDoesNotExist() throws Exception { + OffsetManager offsetManager = createOffsetManager(OffsetRequest.EarliestTime(), "offset"); + offsetManager.getOffset(new Partition(TEST_TOPIC, 1)); + } + + @Test + public void testInitialization() throws Exception { + + int numPartitions = 3; + + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TEST_TOPIC, numPartitions, 1); + + OffsetManager offsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + + Partition[] partitions = new Partition[numPartitions]; + for (int i = 0; i < numPartitions; i++) { + partitions[i] = new Partition(TEST_TOPIC, i); + } + // Earliest + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // send data to increase offsets on topics + Producer producer = createProducer(); + for (int i = 0; i < 10; i++) { + producer.send(new KeyedMessage(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i))); + } + + // earliest time resets at the start of the queue + OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.EarliestTime(), "offset2"); + assertThat(offsetManager2.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager2.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager2.getOffset(partitions[2]), equalTo(0L)); + + // latest time resets at the end of the queue + OffsetManager offsetManager3 = createOffsetManager(OffsetRequest.LatestTime(), "offset3"); + assertThat(offsetManager3.getOffset(partitions[0]), equalTo(4L)); + assertThat(offsetManager3.getOffset(partitions[1]), equalTo(3L)); + assertThat(offsetManager3.getOffset(partitions[2]), equalTo(3L)); + + } + + @Test + public void testUpdate() throws Exception { + int numPartitions = 3; + + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TEST_TOPIC, numPartitions, 1); + + OffsetManager offsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + + Partition[] partitions = new Partition[numPartitions]; + for (int i = 0; i < numPartitions; i++) { + partitions[i] = new Partition(TEST_TOPIC, i); + } + // Earliest + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // send data to increase offsets on topics + Producer producer = createProducer(); + for (int i = 0; i < 10; i++) { + producer.send(new KeyedMessage(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i))); + } + + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + offsetManager1.updateOffset(partitions[0], 5L); + offsetManager1.updateOffset(partitions[1], 4L); + offsetManager1.updateOffset(partitions[2], 3L); + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(5L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(4L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(3L)); + + + // a new offset manager with the same consumerId will observe the updates + OffsetManager newOffsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + assertThat(newOffsetManager1.getOffset(partitions[0]), equalTo(5L)); + assertThat(newOffsetManager1.getOffset(partitions[1]), equalTo(4L)); + assertThat(newOffsetManager1.getOffset(partitions[2]), equalTo(3L)); + + // a new offset manager with a different consumerId will not observe the updates + OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.EarliestTime(), "offset2"); + assertThat(offsetManager2.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager2.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager2.getOffset(partitions[2]), equalTo(0L)); + + } + + @Test + public void testRemove() throws Exception { + int numPartitions = 3; + + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TEST_TOPIC, numPartitions, 1); + + OffsetManager offsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + + Partition[] partitions = new Partition[numPartitions]; + for (int i = 0; i < numPartitions; i++) { + partitions[i] = new Partition(TEST_TOPIC, i); + } + // Earliest + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // send data to increase offsets on topics + Producer producer = createProducer(); + for (int i = 0; i < 6; i++) { + producer.send(new KeyedMessage(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i))); + } + + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + offsetManager1.updateOffset(partitions[0], 5L); + offsetManager1.updateOffset(partitions[1], 4L); + offsetManager1.updateOffset(partitions[2], 3L); + + offsetManager1.deleteOffset(partitions[1]); + offsetManager1.deleteOffset(partitions[2]); + + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(5L)); + // partition 1,2 will be reset + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // a new offset manager with the same consumerId will observe the updates + OffsetManager newOffsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + assertThat(newOffsetManager1.getOffset(partitions[0]), equalTo(5L)); + assertThat(newOffsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(newOffsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // a new offset manager with a different consumerId will not observe the updates + OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.LatestTime(), "offset2"); + assertThat(offsetManager2.getOffset(partitions[0]), equalTo(2L)); + assertThat(offsetManager2.getOffset(partitions[1]), equalTo(2L)); + assertThat(offsetManager2.getOffset(partitions[2]), equalTo(2L)); + } + + @Test + public void testInitialOffsets() throws Exception { + int numPartitions = 3; + + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TEST_TOPIC, numPartitions, 1); + + Partition[] partitions = new Partition[numPartitions]; + for (int i = 0; i < numPartitions; i++) { + partitions[i] = new Partition(TEST_TOPIC, i); + } + + HashMap initialOffsets = new HashMap(); + initialOffsets.put(partitions[0], 8L); + initialOffsets.put(partitions[1], 9L); + initialOffsets.put(partitions[2], 10L); + + OffsetManager offsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1", initialOffsets); + + // send data to increase offsets on topics + Producer producer = createProducer(); + for (int i = 0; i < 6; i++) { + producer.send(new KeyedMessage(TEST_TOPIC, String.valueOf(i), i, String.valueOf(i))); + } + + // The offset manager starts at the configured offsets + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(8L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(9L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(10L)); + + // a new offset manager with the same consumerId will not observe any updates yet + OffsetManager newOffsetManager1 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + assertThat(newOffsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(newOffsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(newOffsetManager1.getOffset(partitions[2]), equalTo(0L)); + + offsetManager1.updateOffset(partitions[0], 9L); + offsetManager1.updateOffset(partitions[1], 10L); + offsetManager1.updateOffset(partitions[2], 11L); + + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(9L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(10L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(11L)); + + // a new offset manager with the same consumerId will observe the updates + OffsetManager newOffsetManager2 = createOffsetManager(OffsetRequest.EarliestTime(), "offset1"); + assertThat(newOffsetManager2.getOffset(partitions[0]), equalTo(9L)); + assertThat(newOffsetManager2.getOffset(partitions[1]), equalTo(10L)); + assertThat(newOffsetManager2.getOffset(partitions[2]), equalTo(11L)); + + offsetManager1.deleteOffset(partitions[0]); + offsetManager1.deleteOffset(partitions[1]); + + // after delete, the offset manager starts in an initial position + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(8L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(9L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(11L)); + + offsetManager1.resetOffsets(Arrays.asList(partitions)); + // after reset, the offset manager starts at the reference timestamp + assertThat(offsetManager1.getOffset(partitions[0]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[1]), equalTo(0L)); + assertThat(offsetManager1.getOffset(partitions[2]), equalTo(0L)); + + // a new offset manager with a different consumerId will not observe the updates + OffsetManager offsetManager2 = createOffsetManager(OffsetRequest.LatestTime(), "offset2"); + assertThat(offsetManager2.getOffset(partitions[0]), equalTo(2L)); + assertThat(offsetManager2.getOffset(partitions[1]), equalTo(2L)); + assertThat(offsetManager2.getOffset(partitions[2]), equalTo(2L)); + } + + protected DefaultConnectionFactory createConnectionFactory() throws Exception { + DefaultConnectionFactory connectionFactory + = new DefaultConnectionFactory(new ZookeeperConfiguration(kafkaRule.getZookeeperConnectionString())); + connectionFactory.afterPropertiesSet(); + return connectionFactory; + } + + protected Producer createProducer() { + Properties properties = TestUtils.getProducerConfig(kafkaRule.getBrokersAsString(), + TestPartitioner.class.getCanonicalName()); + properties.put("serializer.class", StringEncoder.class.getCanonicalName()); + properties.put("key.serializer.class", StringEncoder.class.getCanonicalName()); + ProducerConfig producerConfig = new ProducerConfig(properties); + return new Producer(new kafka.producer.Producer(producerConfig)); + } + + protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId) throws Exception { + return createOffsetManager(referenceTimestamp, consumerId, new HashMap()); + } + + protected abstract OffsetManager createOffsetManager(long referenceTimestamp, + String consumerId, Map initialOffsets) throws Exception; + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/CodecTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/CodecTests.java new file mode 100644 index 0000000000..7a6346af34 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/CodecTests.java @@ -0,0 +1,52 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.offset; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertThat; + +import org.junit.Test; + +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.listener.KafkaTopicOffsetManager; + +/** + * @author Marius Bogoevici + */ +public class CodecTests { + + public static final String SOME_CONSUMER = "someConsumer"; + + public static final String SOME_TOPIC = "someTopic"; + + public static final int SOME_PARTITION_ID = 42; + + @Test + public void testKeyCodec() { + KafkaTopicOffsetManager.KeyEncoderDecoder codec = new KafkaTopicOffsetManager.KeyEncoderDecoder(); + + byte[] encodedValue = codec.toBytes(new KafkaTopicOffsetManager.Key(SOME_CONSUMER, + new Partition(SOME_TOPIC, SOME_PARTITION_ID))); + + KafkaTopicOffsetManager.Key key = codec.fromBytes(encodedValue); + + assertThat(key.getConsumerId(), equalTo(SOME_CONSUMER)); + assertThat(key.getPartition().getTopic(), equalTo(SOME_TOPIC)); + assertThat(key.getPartition().getId(), equalTo(SOME_PARTITION_ID)); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java new file mode 100644 index 0000000000..2bd3593765 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java @@ -0,0 +1,47 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.offset; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.listener.KafkaTopicOffsetManager; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.kafka.support.ZookeeperConnect; + +/** + * @author Marius Bogoevici + */ +public class KafkaOffsetManagerTests extends AbstractOffsetManagerTests { + + @Override + protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId, + Map initialOffsets) throws Exception { + KafkaTopicOffsetManager kafkaTopicOffsetManager = + new KafkaTopicOffsetManager(new ZookeeperConnect(kafkaRule.getZookeeperConnectionString()), + "offsets-spring-test", + initialOffsets); + kafkaTopicOffsetManager.setConsumerId(consumerId); + // do not batch writes during tests for deterministic behaviour + kafkaTopicOffsetManager.setBatchWrites(false); + kafkaTopicOffsetManager.afterPropertiesSet(); + kafkaTopicOffsetManager.setReferenceTimestamp(referenceTimestamp); + return kafkaTopicOffsetManager; + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/MetadataStoreOffsetManagerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/MetadataStoreOffsetManagerTests.java new file mode 100644 index 0000000000..0ceae0f0ae --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/MetadataStoreOffsetManagerTests.java @@ -0,0 +1,53 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.offset; + +import java.util.Map; + +import org.junit.Before; + +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.listener.MetadataStoreOffsetManager; +import org.springframework.integration.kafka.listener.OffsetManager; +import org.springframework.integration.metadata.SimpleMetadataStore; + +/** + * @author Marius Bogoevici + */ +public class MetadataStoreOffsetManagerTests extends AbstractOffsetManagerTests { + + + // tests share a MetadataStore instance internally, simulating the behaviour of persistent stores + private SimpleMetadataStore metadataStore; + + @Before + public void setUp() { + metadataStore = new SimpleMetadataStore(); + } + + @Override + protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId, + Map initialOffsets) throws Exception { + MetadataStoreOffsetManager metadataStoreOffsetManager; + metadataStoreOffsetManager = new MetadataStoreOffsetManager(createConnectionFactory(), initialOffsets); + metadataStoreOffsetManager.setMetadataStore(this.metadataStore); + metadataStoreOffsetManager.setReferenceTimestamp(referenceTimestamp); + metadataStoreOffsetManager.setConsumerId(consumerId); + return metadataStoreOffsetManager; + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java index fe4ddb2948..b7bdbc38a8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java @@ -16,8 +16,10 @@ package org.springframework.integration.kafka.outbound; +import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.endsWith; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItem; import static org.junit.Assert.assertThat; import java.util.ArrayList; @@ -28,13 +30,8 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import kafka.admin.AdminUtils; -import kafka.api.OffsetRequest; -import kafka.common.TopicExistsException; -import kafka.serializer.Decoder; -import kafka.serializer.Encoder; -import org.junit.AfterClass; -import org.junit.ClassRule; +import org.junit.After; +import org.junit.Rule; import org.junit.Test; import org.springframework.context.expression.MapAccessor; @@ -46,7 +43,8 @@ import org.springframework.integration.kafka.core.ZookeeperConfiguration; import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; import org.springframework.integration.kafka.listener.MessageListener; import org.springframework.integration.kafka.listener.MetadataStoreOffsetManager; -import org.springframework.integration.kafka.rule.KafkaRunning; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.kafka.rule.KafkaRule; import org.springframework.integration.kafka.serializer.common.StringDecoder; import org.springframework.integration.kafka.serializer.common.StringEncoder; import org.springframework.integration.kafka.support.KafkaHeaders; @@ -56,8 +54,18 @@ import org.springframework.integration.kafka.support.ProducerFactoryBean; import org.springframework.integration.kafka.support.ProducerMetadata; import org.springframework.integration.kafka.support.ZookeeperConnect; import org.springframework.integration.kafka.util.MessageUtils; +import org.springframework.integration.kafka.util.TopicUtils; import org.springframework.messaging.support.MessageBuilder; +import com.gs.collections.api.multimap.MutableMultimap; +import com.gs.collections.impl.factory.Multimaps; + +import kafka.admin.AdminUtils; +import kafka.api.OffsetRequest; +import kafka.common.TopicExistsException; +import kafka.serializer.Decoder; +import kafka.serializer.Encoder; + /** * @author Gary Russell * @author Marius Bogoevici @@ -68,13 +76,17 @@ public class OutboundTests { private static final String TOPIC = "springintegrationtest"; - @ClassRule - public static KafkaRunning kafkaRunning = KafkaRunning.isRunning(); + private static final String TOPIC2 = "springintegrationtest2"; - @AfterClass - public static void tearDown() { + @Rule + public KafkaRule kafkaRule = new KafkaEmbedded(1); + + private final Decoder decoder = new StringDecoder(); + + @After + public void tearDown() { try { - AdminUtils.deleteTopic(kafkaRunning.getZkClient(), TOPIC); + AdminUtils.deleteTopic(kafkaRule.getZkClient(), TOPIC); } catch (Exception e) { } @@ -86,7 +98,7 @@ public class OutboundTests { // create the topic try { - AdminUtils.createTopic(kafkaRunning.getZkClient(), TOPIC, 1, 1, new Properties()); + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TOPIC, 1, 1); } catch (TopicExistsException e) { // do nothing @@ -94,31 +106,154 @@ public class OutboundTests { final String suffix = UUID.randomUUID().toString(); - ZookeeperConfiguration configuration = new ZookeeperConfiguration(new ZookeeperConnect()); + KafkaMessageListenerContainer kafkaMessageListenerContainer = createMessageListenerContainer(TOPIC); + + int expectedMessageCount = 2; + final List payloads = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + kafkaMessageListenerContainer.setMessageListener(new MessageListener() { + + @Override + public void onMessage(KafkaMessage message) { + payloads.add(MessageUtils.decodePayload(message, decoder)); + latch.countDown(); + } + + }); + + kafkaMessageListenerContainer.start(); + + KafkaProducerContext producerContext = createProducerContext(); + KafkaProducerMessageHandler handler = + new KafkaProducerMessageHandler(producerContext); + + handler.handleMessage(MessageBuilder.withPayload("foo" + suffix) + .setHeader(KafkaHeaders.MESSAGE_KEY, "3") + .setHeader(KafkaHeaders.TOPIC, TOPIC) + .build()); + + SpelExpressionParser parser = new SpelExpressionParser(); + handler.setMessageKeyExpression(parser.parseExpression("headers.foo")); + handler.setTopicExpression(parser.parseExpression("headers.bar")); + StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + evaluationContext.addPropertyAccessor(new MapAccessor()); + handler.setIntegrationEvaluationContext(evaluationContext); + handler.handleMessage(MessageBuilder.withPayload("bar" + suffix) + .setHeader("foo", "3") + .setHeader("bar", TOPIC) + .build()); + + producerContext.stop(); + + latch.await(1000, TimeUnit.MILLISECONDS); + assertThat(latch.getCount(), equalTo(0L)); + for (String payload : payloads) { + assertThat(payload, endsWith(suffix)); + } + kafkaMessageListenerContainer.stop(); + } + + @Test + public void testHeaderRouting() throws Exception { + + // create the topic + + try { + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TOPIC, 1, 1); + } + catch (TopicExistsException e) { + // do nothing + } + + try { + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TOPIC2, 1, 1); + } + catch (TopicExistsException e) { + // do nothing + } + + final String suffix = UUID.randomUUID().toString(); + + KafkaMessageListenerContainer kafkaMessageListenerContainer = createMessageListenerContainer(TOPIC,TOPIC2); + + final Decoder decoder = new StringDecoder(); + + int expectedMessageCount = 4; + final MutableMultimap payloadsByTopic = Multimaps.mutable.list.with(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + kafkaMessageListenerContainer.setMessageListener(new MessageListener() { + + @Override + public void onMessage(KafkaMessage message) { + payloadsByTopic.put(message.getMetadata().getPartition().getTopic(), + MessageUtils.decodePayload(message, decoder)); + latch.countDown(); + } + + }); + + kafkaMessageListenerContainer.start(); + + KafkaProducerContext producerContext = createProducerContext(); + KafkaProducerMessageHandler handler + = new KafkaProducerMessageHandler(producerContext); + + handler.handleMessage(MessageBuilder.withPayload("fooTopic1" + suffix) + .setHeader(KafkaHeaders.MESSAGE_KEY, "3") + .setHeader(KafkaHeaders.TOPIC, TOPIC) + .build()); + + handler.handleMessage(MessageBuilder.withPayload("fooTopic2" + suffix) + .setHeader(KafkaHeaders.MESSAGE_KEY, "3") + .setHeader(KafkaHeaders.TOPIC, TOPIC2) + .build()); + + SpelExpressionParser parser = new SpelExpressionParser(); + handler.setMessageKeyExpression(parser.parseExpression("headers.foo")); + handler.setTopicExpression(parser.parseExpression("headers.bar")); + StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + evaluationContext.addPropertyAccessor(new MapAccessor()); + handler.setIntegrationEvaluationContext(evaluationContext); + handler.handleMessage(MessageBuilder.withPayload("bar1" + suffix) + .setHeader("foo", "3") + .setHeader("bar", TOPIC) + .build()); + + handler.handleMessage(MessageBuilder.withPayload("bar2" + suffix) + .setHeader("foo", "3") + .setHeader("bar", TOPIC2) + .build()); + + producerContext.stop(); + + latch.await(1000, TimeUnit.MILLISECONDS); + assertThat(latch.getCount(), equalTo(0L)); + // messages are routed to both topics + assertThat(payloadsByTopic.keysView(), hasItem(TOPIC)); + assertThat(payloadsByTopic.keysView(), hasItem(TOPIC2)); + assertThat(payloadsByTopic.toMap().get(TOPIC), contains("fooTopic1" + suffix, "bar1" + suffix)); + assertThat(payloadsByTopic.toMap().get(TOPIC2), contains("fooTopic2" + suffix, "bar2" + suffix)); + + kafkaMessageListenerContainer.stop(); + } + + private KafkaMessageListenerContainer createMessageListenerContainer(String... topics) throws Exception { + ZookeeperConfiguration configuration = + new ZookeeperConfiguration(new ZookeeperConnect(kafkaRule.getZookeeperConnectionString())); DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(configuration); connectionFactory.afterPropertiesSet(); - final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, TOPIC); + final KafkaMessageListenerContainer kafkaMessageListenerContainer = + new KafkaMessageListenerContainer(connectionFactory, topics); kafkaMessageListenerContainer.setMaxFetch(100); kafkaMessageListenerContainer.setConcurrency(1); MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory); // start reading at the end of the offsetManager.setReferenceTimestamp(OffsetRequest.LatestTime()); kafkaMessageListenerContainer.setOffsetManager(offsetManager); - final Decoder decoder = new StringDecoder(); - - int expectedMessageCount = 2; - final List payloads = new ArrayList(); - final CountDownLatch latch = new CountDownLatch(expectedMessageCount); - kafkaMessageListenerContainer.setMessageListener(new MessageListener() { - @Override - public void onMessage(KafkaMessage message) { - payloads.add(MessageUtils.decodePayload(message, decoder)); - latch.countDown(); - } - }); - - kafkaMessageListenerContainer.start(); + return kafkaMessageListenerContainer; + } + private KafkaProducerContext createProducerContext() throws Exception { KafkaProducerContext kafkaProducerContext = new KafkaProducerContext(); ProducerMetadata producerMetadata = new ProducerMetadata(TOPIC); producerMetadata.setValueClassType(String.class); @@ -130,36 +265,11 @@ public class OutboundTests { Properties props = new Properties(); props.put("queue.buffering.max.ms", "15000"); ProducerFactoryBean producer = - new ProducerFactoryBean(producerMetadata, "localhost:9092", props); + new ProducerFactoryBean(producerMetadata, kafkaRule.getBrokersAsString(), props); ProducerConfiguration config = new ProducerConfiguration(producerMetadata, producer.getObject()); kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config)); - KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(kafkaProducerContext); - - handler.handleMessage(MessageBuilder.withPayload("foo"+suffix) - .setHeader(KafkaHeaders.MESSAGE_KEY, "3") - .setHeader(KafkaHeaders.TOPIC, TOPIC) - .build()); - - SpelExpressionParser parser = new SpelExpressionParser(); - handler.setMessageKeyExpression(parser.parseExpression("headers.foo")); - handler.setTopicExpression(parser.parseExpression("headers.bar")); - StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); - evaluationContext.addPropertyAccessor(new MapAccessor()); - handler.setIntegrationEvaluationContext(evaluationContext); - handler.handleMessage(MessageBuilder.withPayload("bar"+suffix) - .setHeader("foo", "3") - .setHeader("bar", TOPIC) - .build()); - - kafkaProducerContext.stop(); - - latch.await(1000, TimeUnit.MILLISECONDS); - assertThat(latch.getCount(), equalTo(0L)); - for (String payload : payloads) { - assertThat(payload, endsWith(suffix)); - } - kafkaMessageListenerContainer.stop(); + return kafkaProducerContext; } - + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java index eecec93a08..72ffa30132 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java @@ -139,6 +139,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule { return zookeeperClient; } + @Override public String getZookeeperConnectionString() { return zookeeper.connectString(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java index 8e28208895..8b3d5e551f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java @@ -34,6 +34,8 @@ public interface KafkaRule extends TestRule { ZkClient getZkClient(); + String getZookeeperConnectionString(); + BrokerAddress[] getBrokerAddresses(); String getBrokersAsString(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java index e1cf4cfa8c..8cc053df63 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java @@ -74,6 +74,11 @@ public class KafkaRunning extends TestWatcher implements KafkaRule { return zkClient; } + @Override + public String getZookeeperConnectionString() { + return ZOOKEEPER_CONNECT_STRING; + } + @Override @SuppressWarnings("serial") public BrokerAddress[] getBrokerAddresses() {