Spring/spring-integration-extensions
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 8dc41021b8
commit 07d45d69c1
13 changed files with 43 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
samples/kafka/build.gradle
View File

@@ -37,15 +37,8 @@ repositories {
}
dependencies {
compile([
"org.apache.avro:avro:1.7.3",
"org.apache.avro:avro-compiler:1.7.3"
])
compile "org.springframework:spring-beans:3.1.3.RELEASE"
compile "org.springframework:spring-context:3.1.3.RELEASE"
compile "org.springframework:spring-expression:3.1.3.RELEASE"
compile "org.springframework.integration:spring-integration-stream:2.2.0.RELEASE"
compile("org.springframework.integration:spring-integration-kafka:0.5.0.BUILD-SNAPSHOT") {
compile "org.springframework.integration:spring-integration-stream:$springIntegrationVersion"
compile("org.springframework.integration:spring-integration-kafka:$springIntegrationKafkaVersion") {
exclude module: 'log4j'
exclude module: 'jms'
exclude module: 'jmxtools'

7
samples/kafka/gradle.properties
View File

@@ -1,7 +1,6 @@
springVersion = 3.1.3.RELEASE
springIntegrationVersion = 2.2.0.RELEASE
springIntegrationKafkaVersion = 0.5.0.BUILD-SNAPSHOT
version = 0.5.0.BUILD-SNAPSHOT
springIntegrationVersion = 4.0.3.RELEASE
springIntegrationKafkaVersion = 1.0.0.BUILD-SNAPSHOT
version = 1.0.0.BUILD-SNAPSHOT

BIN
samples/kafka/gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

4
samples/kafka/gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,6 +1,6 @@
#Mon Oct 28 16:21:39 EDT 2013
#Tue Jul 22 17:46:28 EEST 2014
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
distributionUrl=http\://services.gradle.org/distributions/gradle-1.8-bin.zip
distributionUrl=http\://services.gradle.org/distributions/gradle-1.12-all.zip

6
samples/kafka/gradlew vendored
View File

@@ -1,4 +1,4 @@
#!/bin/bash
#!/usr/bin/env bash
##############################################################################
##
@@ -61,9 +61,9 @@ while [ -h "$PRG" ] ; do
fi
done
SAVED="`pwd`"
cd "`dirname \"$PRG\"`/"
cd "`dirname \"$PRG\"`/" >&-
APP_HOME="`pwd -P`"
cd "$SAVED"
cd "$SAVED" >&-
CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar

4
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/CustomPartitioner.java
View File

@@ -22,14 +22,14 @@ import kafka.producer.Partitioner;
*
* This class is for internal use only and therefore is at default access level
*/
class CustomPartitioner<T> implements Partitioner<T> {
class CustomPartitioner 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) {
final String s = (String) key;
final Integer i = Integer.parseInt(s);
return i % numPartitions;

2
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/OutboundRunner.java
View File

@@ -18,9 +18,9 @@ package org.springframework.integration.samples.kafka.outbound;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.samples.kafka.user.User;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
public class OutboundRunner {
private static final String CONFIG = "kafkaOutboundAdapterParserTests-context.xml";

6
samples/kafka/src/main/java/org/springframework/integration/samples/kafka/outbound/PartitionlessTransformer.java
View File

@@ -15,9 +15,9 @@
*/
package org.springframework.integration.samples.kafka.outbound;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import java.util.ArrayList;
import java.util.Collection;
@@ -36,13 +36,13 @@ public class PartitionlessTransformer implements Transformer {
final Map<String, Map<Integer, List<Object>>> origData =
(Map<String, Map<Integer, List<Object>>>) message.getPayload();
final Map<String, List<Object>> nonPartitionedData = new HashMap<String, List<Object>>();
final Map<String, List<Object>> nonPartitionedData = new HashMap<>();
for(final String topic : origData.keySet()) {
final Map<Integer, List<Object>> partitionedData = origData.get(topic);
final Collection<List<Object>> nonPartitionedDataFromTopic = partitionedData.values();
final List<Object> mergedList = new ArrayList<Object>();
final List<Object> mergedList = new ArrayList<>();
for (final List<Object> l : nonPartitionedDataFromTopic){
mergedList.addAll(l);

2
spring-integration-kafka/build.gradle
View File

@@ -19,7 +19,7 @@ sourceCompatibility = targetCompatibility = 1.6
ext {
avroVersion = '1.7.6'
jacocoVersion = '0.7.0.201403182114'
kafkaVersion = '0.8.0'
kafkaVersion = '0.8.1.1'
metricsVersion = '2.2.0'
scalaVersion = '2.10'
springIntegrationVersion = '4.0.3.RELEASE'

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