Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-131-134: Add KafkaTopicOffsetManager

Browse Source 
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`
This commit is contained in:
Marius Bogoevici
2015-02-03 21:29:30 +02:00
committed by Artem Bilan
parent d85677f61e
commit ae2a1a08cd
34 changed files with 2514 additions and 448 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/AbstractConfiguration.java
View File

@@ -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<BrokerAddress> doGetBrokerAddresses();
@Override
public List<Partition> getDefaultPartitions() {
return defaultPartitions;
return this.defaultPartitions;
}
public void setDefaultPartitions(List<Partition> 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<BrokerAddress> doGetBrokerAddresses();
}

32
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/BrokerAddress.java
View File

@@ -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;
}
}

48
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Configuration.java
View File

@@ -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

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Connection.java
View File

@@ -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

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ConnectionFactory.java
View File

@@ -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<String> topics);
void refreshMetadata(Collection<String> topics);
/**
* Retrieves the partitions of a given topic

229
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java
View File

@@ -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<TopicMetadata> errorlessTopicMetadataPredicate = new ErrorlessTopicMetadataPredicate();
private final GetBrokersByPartitionFunction getBrokersByPartitionFunction = new GetBrokersByPartitionFunction();
private final ConnectionInstantiationFunction connectionInstantiationFunction = new ConnectionInstantiationFunction();
private final Configuration configuration;
private final AtomicReference<PartitionBrokerMap> partitionBrokerMapReference = new AtomicReference<PartitionBrokerMap>();
private final AtomicReference<MetadataCache> metadataCacheHolder =
new AtomicReference<MetadataCache>(new MetadataCache(Collections.<TopicMetadata>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<BrokerAddress, Connection> 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.<String>emptyList() :
Collections.singletonList(configuration.getDefaultTopic()));
refreshMetadata(this.configuration.getDefaultTopic() == null ? Collections.<String>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<Partition, BrokerAddress> getLeaders(Iterable<Partition> partitions) {
return FastList.newList(partitions).toMap(Functions.<Partition>getPassThru(), getBrokersByPartitionFunction);
return Iterate.toMap(partitions, Functions.<Partition>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<String> topics) {
public void refreshMetadata(Collection<String> 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<String>(topics)),
ClientUtils$.MODULE$.parseBrokerList(brokerAddressesAsString),
getClientId(), fetchMetadataTimeout, 0));
Map<Partition, BrokerAddress> kafkaBrokerAddressMap = new HashMap<Partition, BrokerAddress>();
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<TopicMetadata> 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<Partition> 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<Partition> 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<TopicMetadata> {
@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);
}
}

111
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/MetadataCache.java Normal file
View File

@@ -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<String, Map<Partition, PartitionMetadata>> metadatasByTopic;
public MetadataCache(Iterable<TopicMetadata> topicMetadatas) {
UnifiedMap<String, Map<Partition, PartitionMetadata>> topicData = UnifiedMap.newMap();
for (TopicMetadata topicMetadata : topicMetadatas) {
Map<Partition, PartitionMetadata> partitionData = toPartitionMetadataMap(topicMetadata);
topicData.put(topicMetadata.topic(), partitionData);
}
this.metadatasByTopic = topicData.toImmutable().castToMap();
}
private MetadataCache(Map<String, Map<Partition, PartitionMetadata>> metadatasByTopic) {
this.metadatasByTopic = metadatasByTopic;
}
public MetadataCache merge(final Iterable<TopicMetadata> topicMetadatas) {
UnifiedMap<String, Map<Partition, PartitionMetadata>> unifiedMap = UnifiedMap.newMap(this.metadatasByTopic);
return new MetadataCache(Iterate.addToMap(topicMetadatas, getTopicNameFunction,
toIndexedPartitionMetadataFunction, unifiedMap).toImmutable().castToMap());
}
public Collection<Partition> 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<Partition, PartitionMetadata> partitionMetadatasForTopic = this.metadatasByTopic.get(partition.getTopic());
if (!partitionMetadatasForTopic.containsKey(partition)) {
return null;
}
return new BrokerAddress(partitionMetadatasForTopic.get(partition).leader());
}
private static Map<Partition, PartitionMetadata> toPartitionMetadataMap(TopicMetadata topicMetadata) {
Map<Partition, PartitionMetadata> 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<TopicMetadata, String> {
@Override
public String valueOf(TopicMetadata object) {
return object.topic();
}
}
@SuppressWarnings("serial")
private static class ToIndexedPartitionMetadataFunction implements
Function<TopicMetadata, Map<Partition, PartitionMetadata>> {
@Override
public Map<Partition, PartitionMetadata> valueOf(TopicMetadata object) {
return toPartitionMetadataMap(object);
}
}
}

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/Partition.java
View File

@@ -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

66
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionBrokerMap.java
View File

@@ -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<BrokerAddress, Partition> partitionsByBroker;
private final ImmutableMap<Partition, BrokerAddress> brokersByPartition;
private final Multimap<String, Partition> partitionsByTopic;
public PartitionBrokerMap(UnifiedMap<Partition, BrokerAddress> brokersByPartition) {
this.brokersByPartition = brokersByPartition.toImmutable();
this.partitionsByTopic = this.brokersByPartition.keysView().groupBy(getTopicFunction);
this.partitionsByBroker = brokersByPartition.flip().toImmutable();
}
public Multimap<BrokerAddress, Partition> getPartitionsByBroker() {
return partitionsByBroker;
}
public ImmutableMap<Partition, BrokerAddress> getBrokersByPartition() {
return brokersByPartition;
}
public Multimap<String, Partition> getPartitionsByTopic() {
return partitionsByTopic;
}
@SuppressWarnings("serial")
private static class GetTopicFunction implements Function<Partition, String> {
@Override
public String valueOf(Partition partition) {
return partition.getTopic();
}
}
}

30
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/PartitionNotFoundException.java Normal file
View File

@@ -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()));
}
}

23
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/ZookeeperConfiguration.java
View File

@@ -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<BrokerAddress> 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<Broker> allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient);
FastList<Broker> brokers = FastList.newList(JavaConversions.asJavaCollection(allBrokersInCluster));
return brokers.collect(brokerToBrokerAddressFunction);

165
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java Normal file
View File

@@ -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<Partition, Long> initialOffsets;
public AbstractOffsetManager(ConnectionFactory connectionFactory) {
this(connectionFactory, new HashMap<Partition, Long>());
}
public AbstractOffsetManager(ConnectionFactory connectionFactory, Map<Partition, Long> 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<Long> 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<Partition> 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);
}

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -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<Partition> partitionsAsList = FastList.newList(partitionsToReset);
FastList<String> topics = partitionsAsList.collect(new PartitionToTopicFunction()).distinct();
kafkaTemplate.getConnectionFactory().refreshLeaders(topics);
kafkaTemplate.getConnectionFactory().refreshMetadata(topics);
Map<Partition, BrokerAddress> leaders = kafkaTemplate.getConnectionFactory().getLeaders(partitionsToReset);
synchronized (partitionsByBrokerMap) {
forEachKeyValue(leaders, new AddPartitionToBrokerProcedure());

521
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaTopicOffsetManager.java Normal file
View File

@@ -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<Partition, Long> data = new ConcurrentHashMap<Partition, Long>();
private String compressionCodec = "default";
private Producer<Key, Long> 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<Partition, Long>());
}
public KafkaTopicOffsetManager(ZookeeperConnect zookeeperConnect, String topic,
Map<Partition, Long> 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<Key, Long>(this.topic, new Key(this.consumerId, partition), offset));
}
@Override
protected void doRemoveOffset(Partition partition) {
this.data.remove(partition);
this.producer.send(new KeyedMessage<Key, Long>(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.<Class<? extends Throwable>, Boolean>singletonMap(TopicNotFoundException.class, true)));
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
backOffPolicy.setInitialInterval(50L);
backOffPolicy.setMaxInterval(1000L);
backOffPolicy.setMultiplier(2);
retryValidateTopic.setBackOffPolicy(backOffPolicy);
Collection<Partition> partitions =
retryValidateTopic.execute(new RetryCallback<Collection<Partition>, Exception>() {
@Override
public Collection<Partition> 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<Long> 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<Long> 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<KafkaMessageBatch> 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<Partition, Long> 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<Key, Long> producerMetadata = new ProducerMetadata<Key, Long>(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<Key, Long> producerFB
= new ProducerFactoryBean<Key, Long>(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<Key>, Decoder<Key> {
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);
}
}
}
}

66
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/LongEncoderDecoder.java Normal file
View File

@@ -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<Long>, Decoder<Long> {
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();
}
}
}

131
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/MetadataStoreOffsetManager.java
View File

@@ -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<Partition, Long> initialOffsets;
private long referenceTimestamp = KafkaConsumerDefaults.DEFAULT_OFFSET_RESET;
public MetadataStoreOffsetManager(ConnectionFactory connectionFactory) {
this(connectionFactory, null);
super(connectionFactory);
}
public MetadataStoreOffsetManager(ConnectionFactory connectionFactory, Map<Partition, Long> initialOffsets) {
this.connectionFactory = connectionFactory;
this.initialOffsets = initialOffsets == null? new HashMap<Partition, Long>() : 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<Long> 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<Partition> 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();
}
}

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/OffsetManager.java
View File

@@ -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.

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java
View File

@@ -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<K, V> implements SmartLifecycle, NamedComponent, BeanNameAware {
@@ -66,7 +67,6 @@ public class KafkaProducerContext<K, V> 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<K, V> 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 " +

11
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ZookeeperConnect.java
View File

@@ -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;
}

45
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/LoggingUtils.java Normal file
View File

@@ -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();
}
}
}

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/MessageUtils.java
View File

@@ -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> T decodeKey(KafkaMessage message, Decoder<T> 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()));
}
}

156
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/util/TopicUtils.java Normal file
View File

@@ -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<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
scala.collection.Map<Object, Seq<Object>> 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<Object, Seq<Object>> 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<TopicMetadata, Exception>() {
@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<PartitionMetadata> 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);
}
}
}

262
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/core/MetadataCacheTests.java Normal file
View File

@@ -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.<TopicMetadata>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<TopicMetadata> topicMetadatas = new ArrayList<TopicMetadata>();
HashSet<PartitionMetadata> 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<TopicMetadata> topicMetadatas = new ArrayList<TopicMetadata>();
HashSet<PartitionMetadata> 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.<TopicMetadata>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<TopicMetadata> topicMetadatas = new ArrayList<TopicMetadata>();
HashSet<PartitionMetadata> 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.<TopicMetadata>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<TopicMetadata> topic2Metadatas = new ArrayList<TopicMetadata>();
HashSet<PartitionMetadata> 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<TopicMetadata> overridingTopicMetadatas = new ArrayList<TopicMetadata>();
HashSet<PartitionMetadata> 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<Broker> emptySeq() {
return HashSet$.MODULE$.<Broker>empty().toSeq();
}
public <T> Some<T> asScalaOption(T object) {
return new Some<T>(object);
}
}

80
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/AbstractBrokerTests.java
View File

@@ -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<Integer, Integer> 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<Integer, Integer> 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<Integer, Integer> createPartitionDistribution(int partitionCount, int brokers, int replication) {
public MutableMultimap<Integer, Integer> createPartitionDistribution(int partitionCount, int brokers,
int replication) {
MutableMultimap<Integer, Integer> 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<KeyedMessage<String, String>> createMessages(int count, String topic) {
return createMessagesInRange(0,count-1,topic);
return createMessagesInRange(0, count - 1, topic);
}
public static scala.collection.Seq<KeyedMessage<String, String>> createMessagesInRange(int start, int end, String topic) {
List<KeyedMessage<String,String>> messages = new ArrayList<KeyedMessage<String, String>>();
for (int i=start; i<= end; i++) {
public static scala.collection.Seq<KeyedMessage<String, String>> createMessagesInRange(int start, int end,
String topic) {
List<KeyedMessage<String, String>> messages = new ArrayList<KeyedMessage<String, String>>();
for (int i = start; i <= end; i++) {
messages.add(new KeyedMessage<String, String>(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<String, String>(new ProducerConfig(producerConfig));
}
@@ -136,12 +138,16 @@ public abstract class AbstractBrokerTests {
@SuppressWarnings({"rawtypes", "serial", "deprecation"})
private Map toKafkaPartitionMap(Multimap<Integer, Integer> partitions) {
java.util.Map<Object, Seq<Object>> m = partitions.toMap().collect(new Function2<Integer, RichIterable<Integer>, Pair<Object, Seq<Object>>>() {
@Override
public Pair<Object, Seq<Object>> value(Integer argument1, RichIterable<Integer> argument2) {
return Tuples.pair((Object) argument1, List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq());
}
});
java.util.Map<Object, Seq<Object>> m = partitions.toMap()
.collect(new Function2<Integer, RichIterable<Integer>, Pair<Object, Seq<Object>>>() {
@Override
public Pair<Object, Seq<Object>> value(Integer argument1, RichIterable<Integer> argument2) {
return Tuples.pair((Object) argument1,
List$.MODULE$.fromArray(argument2.toArray(new Object[0])).toSeq());
}
});
return Map$.MODULE$.apply(JavaConversions.asScalaMap(m).toSeq());
}

135
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithKafkaOffsetManagerTests.java Normal file
View File

@@ -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<Partition> readPartitions = new ArrayList<Partition>();
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<Integer, KeyedMessageWithOffset> receivedData =
new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
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);
}
}

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java
View File

@@ -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<Partition> partitionsToReset) {
System.out.println("resetting " + offsetResetLatch.getCount() + " partitions: "

295
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/AbstractOffsetManagerTests.java Normal file
View File

@@ -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<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
producer.send(new KeyedMessage<String, String>(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<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
producer.send(new KeyedMessage<String, String>(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<String, String> producer = createProducer();
for (int i = 0; i < 6; i++) {
producer.send(new KeyedMessage<String, String>(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<Partition, Long> initialOffsets = new HashMap<Partition, Long>();
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<String, String> producer = createProducer();
for (int i = 0; i < 6; i++) {
producer.send(new KeyedMessage<String, String>(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<String, String> 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<String, String>(new kafka.producer.Producer<String, String>(producerConfig));
}
protected OffsetManager createOffsetManager(long referenceTimestamp, String consumerId) throws Exception {
return createOffsetManager(referenceTimestamp, consumerId, new HashMap<Partition, Long>());
}
protected abstract OffsetManager createOffsetManager(long referenceTimestamp,
String consumerId, Map<Partition, Long> initialOffsets) throws Exception;
}

52
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/CodecTests.java Normal file
View File

@@ -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));
}
}

47
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/KafkaOffsetManagerTests.java Normal file
View File

@@ -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<Partition, Long> 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;
}
}

53
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/offset/MetadataStoreOffsetManagerTests.java Normal file
View File

@@ -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<Partition, Long> initialOffsets) throws Exception {
MetadataStoreOffsetManager metadataStoreOffsetManager;
metadataStoreOffsetManager = new MetadataStoreOffsetManager(createConnectionFactory(), initialOffsets);
metadataStoreOffsetManager.setMetadataStore(this.metadataStore);
metadataStoreOffsetManager.setReferenceTimestamp(referenceTimestamp);
metadataStoreOffsetManager.setConsumerId(consumerId);
return metadataStoreOffsetManager;
}
}

226
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java
View File

@@ -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<String> 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<String> payloads = new ArrayList<String>();
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<String, String> producerContext = createProducerContext();
KafkaProducerMessageHandler<String, String> handler =
new KafkaProducerMessageHandler<String, String>(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<String> decoder = new StringDecoder();
int expectedMessageCount = 4;
final MutableMultimap<String, String> 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<String, String> producerContext = createProducerContext();
KafkaProducerMessageHandler<String, String> handler
= new KafkaProducerMessageHandler<String, String>(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<String> decoder = new StringDecoder();
int expectedMessageCount = 2;
final List<String> payloads = new ArrayList<String>();
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<String, String> createProducerContext() throws Exception {
KafkaProducerContext<String, String> kafkaProducerContext = new KafkaProducerContext<String, String>();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC);
producerMetadata.setValueClassType(String.class);
@@ -130,36 +265,11 @@ public class OutboundTests {
Properties props = new Properties();
props.put("queue.buffering.max.ms", "15000");
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<String, String>(producerMetadata, "localhost:9092", props);
new ProducerFactoryBean<String, String>(producerMetadata, kafkaRule.getBrokersAsString(), props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<String, String>(producerMetadata, producer.getObject());
kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config));
KafkaProducerMessageHandler<String, String> handler = new KafkaProducerMessageHandler<String, String>(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;
}
}

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaEmbedded.java
View File

@@ -139,6 +139,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule {
return zookeeperClient;
}
@Override
public String getZookeeperConnectionString() {
return zookeeper.connectString();
}

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRule.java
View File

@@ -34,6 +34,8 @@ public interface KafkaRule extends TestRule {
ZkClient getZkClient();
String getZookeeperConnectionString();
BrokerAddress[] getBrokerAddresses();
String getBrokersAsString();

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/rule/KafkaRunning.java
View File

@@ -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() {