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