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

INTEXT-110: Update Kafka version to 0.8.1.1

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-110

Perform the necessary API Changes (Partioner interface in Kafka is now non-genericized for Keys - concrete Object is the new type for Key)
Unit test updates for an internal Kafka class (MessageMetadata)
Updates for the samples Kafka project to use the new M2 snapshot, Spring 4.0 Messaging changes, other build related minor changes, etc.
This commit is contained in:
Soby Chacko
2014-07-22 00:42:22 -04:00
committed by Artem Bilan
parent 58ff34d83b
commit e22a90ea1f
4 changed files with 26 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultPartitioner.java
View File

@@ -24,14 +24,15 @@ import kafka.utils.Utils;
*
* This class is for internal use only and therefore is at default access level
*/
class DefaultPartitioner<T> implements Partitioner<T> {
class DefaultPartitioner implements Partitioner {
/**
* Uses the key to calculate a partition bucket id for routing
* the data to the appropriate broker partition
* @return an integer between 0 and numPartitions-1
*/
@Override
public int partition(final T key, final int numPartitions) {
public int partition(final Object key, final int numPartitions) {
return Utils.abs(key.hashCode()) % numPartitions;
}
}

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java
View File

@@ -71,7 +71,7 @@ public class ProducerFactoryBean<K,V> implements FactoryBean<Producer<K,V>> {
LOGGER.info("Using producer properties => " + props);
final ProducerConfig config = new ProducerConfig(props);
final EventHandler<K, V> eventHandler = new DefaultEventHandler<K, V>(config,
producerMetadata.getPartitioner() == null ? new DefaultPartitioner<K>() : producerMetadata.getPartitioner(),
producerMetadata.getPartitioner() == null ? new DefaultPartitioner() : producerMetadata.getPartitioner(),
producerMetadata.getValueEncoder(), producerMetadata.getKeyEncoder(),
new ProducerPool(config), new HashMap<String, kafka.api.TopicMetadata>());

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java
View File

@@ -35,7 +35,7 @@ public class ProducerMetadata<K,V> implements InitializingBean {
private Class<V> valueClassType;
private final String topic;
private String compressionCodec = "default";
private Partitioner<K> partitioner;
private Partitioner partitioner;
private boolean async = false;
private String batchNumMessages;
@@ -94,11 +94,11 @@ public class ProducerMetadata<K,V> implements InitializingBean {
this.compressionCodec = compressionCodec;
}
public Partitioner<K> getPartitioner() {
public Partitioner getPartitioner() {
return partitioner;
}
public void setPartitioner(final Partitioner<K> partitioner) {
public void setPartitioner(final Partitioner partitioner) {
this.partitioner = partitioner;
}

22
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java
View File

@@ -262,9 +262,25 @@ public class ConsumerConfigurationTests<K,V> {
topicStreamMap.put("topic1", 1);
when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap);
when(messageLeftOverTracker.getCurrentCount()).thenReturn(3);
final MessageAndMetadata<String, String> m1 = new MessageAndMetadata<String, String>("key1", "value1", "topic1", 1, 1L);
final MessageAndMetadata<String, String> m2 = new MessageAndMetadata<String, String>("key2", "value2", "topic2", 1, 1L);
final MessageAndMetadata<String, String> m3 = new MessageAndMetadata<String, String>("key1", "value3", "topic3", 1, 1L);
final MessageAndMetadata<String, String> m1 = mock(MessageAndMetadata.class);
final MessageAndMetadata<String, String> m2 = mock(MessageAndMetadata.class);
final MessageAndMetadata<String, String> m3 = mock(MessageAndMetadata.class);
when(m1.key()).thenReturn("key1");
when(m1.message()).thenReturn("value1");
when(m1.topic()).thenReturn("topic1");
when(m1.partition()).thenReturn(1);
when(m2.key()).thenReturn("key2");
when(m2.message()).thenReturn("value2");
when(m2.topic()).thenReturn("topic2");
when(m2.partition()).thenReturn(1);
when(m3.key()).thenReturn("key1");
when(m3.message()).thenReturn("value3");
when(m3.topic()).thenReturn("topic3");
when(m3.partition()).thenReturn(1);
final List<MessageAndMetadata<String, String>> mList = new ArrayList<MessageAndMetadata<String, String>>();
mList.add(m1);