From e22a90ea1fa36de8306082f56cc5205675d389d9 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 22 Jul 2014 00:42:22 -0400 Subject: [PATCH] INTEXT-110: Update Kafka version to 0.8.1.1 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. --- .../kafka/support/DefaultPartitioner.java | 5 +++-- .../kafka/support/ProducerFactoryBean.java | 2 +- .../kafka/support/ProducerMetadata.java | 6 ++--- .../support/ConsumerConfigurationTests.java | 22 ++++++++++++++++--- 4 files changed, 26 insertions(+), 9 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultPartitioner.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultPartitioner.java index 929ef8951a..007c116b6b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultPartitioner.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultPartitioner.java @@ -24,14 +24,15 @@ import kafka.utils.Utils; * * This class is for internal use only and therefore is at default access level */ -class DefaultPartitioner implements Partitioner { +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; } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java index b622154432..2dfbe6cbf0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerFactoryBean.java @@ -71,7 +71,7 @@ public class ProducerFactoryBean implements FactoryBean> { LOGGER.info("Using producer properties => " + props); final ProducerConfig config = new ProducerConfig(props); final EventHandler eventHandler = new DefaultEventHandler(config, - producerMetadata.getPartitioner() == null ? new DefaultPartitioner() : producerMetadata.getPartitioner(), + producerMetadata.getPartitioner() == null ? new DefaultPartitioner() : producerMetadata.getPartitioner(), producerMetadata.getValueEncoder(), producerMetadata.getKeyEncoder(), new ProducerPool(config), new HashMap()); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java index e6dca255c7..fe6ba9a4e4 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerMetadata.java @@ -35,7 +35,7 @@ public class ProducerMetadata implements InitializingBean { private Class valueClassType; private final String topic; private String compressionCodec = "default"; - private Partitioner partitioner; + private Partitioner partitioner; private boolean async = false; private String batchNumMessages; @@ -94,11 +94,11 @@ public class ProducerMetadata implements InitializingBean { this.compressionCodec = compressionCodec; } - public Partitioner getPartitioner() { + public Partitioner getPartitioner() { return partitioner; } - public void setPartitioner(final Partitioner partitioner) { + public void setPartitioner(final Partitioner partitioner) { this.partitioner = partitioner; } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java index c23a2b676f..1dab17586b 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java @@ -262,9 +262,25 @@ public class ConsumerConfigurationTests { topicStreamMap.put("topic1", 1); when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); when(messageLeftOverTracker.getCurrentCount()).thenReturn(3); - final MessageAndMetadata m1 = new MessageAndMetadata("key1", "value1", "topic1", 1, 1L); - final MessageAndMetadata m2 = new MessageAndMetadata("key2", "value2", "topic2", 1, 1L); - final MessageAndMetadata m3 = new MessageAndMetadata("key1", "value3", "topic3", 1, 1L); + + final MessageAndMetadata m1 = mock(MessageAndMetadata.class); + final MessageAndMetadata m2 = mock(MessageAndMetadata.class); + final MessageAndMetadata 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> mList = new ArrayList>(); mList.add(m1);