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

INTEXT-106: Update to spring-integration 4.0

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

* Upgrade to SI 4.0 and other libs
* Upgrade to Gradle 1.12
* Polisihng for `build.gradle`
* Fix compile warnings
This commit is contained in:
Jonathan Pearlin
2014-06-12 18:07:19 -04:00
committed by Artem Bilan
parent 1307646c3c
commit 989ce89fed
15 changed files with 101 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,13 +1,7 @@
description = 'Spring Integration Kafka Support'
buildscript {
repositories {
maven { url 'https://repo.springsource.org/plugins-snapshot' }
}
}
apply plugin: 'java'
apply from: "${rootProject.projectDir}/publish-maven.gradle"
apply from: "${rootProject.projectDir}/publish-maven.gradle"
apply plugin: 'eclipse'
apply plugin: 'idea'
@@ -17,37 +11,31 @@ repositories {
maven {
url 'https://repository.apache.org/content/groups/public'
}
maven { url 'https://repo.springsource.org/libs-milestone' }
maven { url 'http://repo.spring.io/libs-milestone' }
}
sourceCompatibility=1.6
targetCompatibility=1.6
sourceCompatibility = targetCompatibility = 1.6
ext {
junitVersion = '4.11'
kafkaVersion = '0.8.0-beta1'
log4jVersion = '1.2.12'
mockitoVersion = '1.9.5'
scalaVersion = '2.9.2'
springVersion = '3.1.3.RELEASE'
springIntegrationVersion = '2.2.4.RELEASE'
avroVersion = '1.7.6'
jacocoVersion = '0.7.0.201403182114'
kafkaVersion = '0.8.0'
metricsVersion = '2.2.0'
scalaVersion = '2.10'
springIntegrationVersion = '4.0.2.RELEASE'
idPrefix = 'kafka'
linkHomepage = 'https://github.com/SpringSource/spring-integration-extensions'
linkCi = 'https://build.springsource.org/browse/INTEXT'
linkIssue = 'https://jira.springsource.org/browse/INTEXT'
linkScmUrl = 'https://github.com/SpringSource/spring-integration-extensions'
linkScmConnection = 'https://github.com/SpringSource/spring-integration-extensions.git'
linkScmDevConnection = 'git@github.com:SpringSource/spring-integration-extensions.git'
linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions'
linkCi = 'https://build.spring.io/browse/INTEXT'
linkIssue = 'https://jira.spring.io/browse/INTEXT'
linkScmUrl = 'https://github.com/spring-projects/spring-integration-extensions'
linkScmConnection = 'https://github.com/spring-projects/spring-integration-extensions.git'
linkScmDevConnection = 'git@github.com:spring-projects/spring-integration-extensions.git'
}
eclipse {
project {
natures += 'org.springframework.ide.eclipse.core.springnature'
}
}
eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature'
sourceSets {
test {
@@ -64,14 +52,12 @@ configurations {
}
dependencies {
compile "org.springframework:spring-beans:$springVersion"
compile "org.springframework:spring-context:$springVersion"
compile "org.springframework:spring-expression:$springVersion"
compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
compile "org.apache.avro:avro:1.7.3"
compile "org.apache.avro:avro-compiler:1.7.3"
runtime "com.yammer.metrics:metrics-core:2.2.0"
runtime "com.yammer.metrics:metrics-annotation:2.2.0"
compile "org.apache.avro:avro:$avroVersion"
compile "org.apache.avro:avro-compiler:$avroVersion"
runtime "com.yammer.metrics:metrics-core:$metricsVersion"
runtime "com.yammer.metrics:metrics-annotation:$metricsVersion"
compile("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion") {
exclude module: 'jms'
@@ -79,29 +65,14 @@ dependencies {
exclude module: 'jmxri'
}
compile "org.scala-lang:scala-library:$scalaVersion"
compile "org.scala-lang:scala-compiler:$scalaVersion"
testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion"
testCompile "junit:junit-dep:$junitVersion"
testCompile("log4j:log4j:$log4jVersion") {
exclude module: 'jms'
exclude module: 'jmx'
exclude module: 'jmxtools'
exclude module: 'jmxri'
}
testCompile "org.mockito:mockito-all:$mockitoVersion"
testCompile "org.springframework:spring-test:$springVersion"
testCompile "org.springframework.integration:spring-integration-stream:$springIntegrationVersion"
jacoco group: "org.jacoco", name: "org.jacoco.agent", version: "0.6.2.201302030002", classifier: "runtime"
jacoco "org.jacoco:org.jacoco.agent:$jacocoVersion:runtime"
}
// enable all compiler warnings; individual projects may customize further
ext.xLintArg = '-Xlint:all'
[compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg]
[compileJava, compileTestJava]*.options*.compilerArgs = ['-Xlint:all,-options']
test {
// suppress all console output during testing unless running `gradle -i`
@@ -156,7 +127,7 @@ task schemaZip(type: Zip) {
group = 'Distribution'
classifier = 'schema'
description = "Builds -${classifier} archive containing all " +
"XSDs for deployment at static.springframework.org/schema."
"XSDs for deployment at static.springframework.org/schema."
def Properties schemas = new Properties();
def shortName = idPrefix.replaceFirst("${idPrefix}-", '')
@@ -170,7 +141,7 @@ task schemaZip(type: Zip) {
it.path.endsWith(schemas.get(key))
}
assert xsdFile != null
into ("integration/${shortName}") {
into("integration/${shortName}") {
from xsdFile.path
}
}
@@ -181,13 +152,13 @@ task docsZip(type: Zip) {
group = 'Distribution'
classifier = 'docs'
description = "Builds -${classifier} archive containing api " +
"for deployment at static.springframework.org/spring-integration/docs."
"for deployment at static.spring.io/spring-integration/docs."
from('src/dist') {
include 'changelog.txt'
}
from (api) {
from(api) {
into 'api'
}
}
@@ -196,7 +167,7 @@ task distZip(type: Zip, dependsOn: [docsZip, schemaZip]) {
group = 'Distribution'
classifier = 'dist'
description = "Builds -${classifier} archive, containing all jars and docs, " +
"suitable for community download page."
"suitable for community download page."
ext.baseDir = "${project.name}-${project.version}";
@@ -215,7 +186,7 @@ task distZip(type: Zip, dependsOn: [docsZip, schemaZip]) {
into "${baseDir}/schema"
}
into ("${baseDir}/libs") {
into("${baseDir}/libs") {
from project.jar
from project.sourcesJar
from project.javadocJar
@@ -228,7 +199,7 @@ task depsZip(type: Zip, dependsOn: distZip) { zipTask ->
group = 'Distribution'
classifier = 'dist-with-deps'
description = "Builds -${classifier} archive, containing everything " +
"in the -${distZip.classifier} archive plus all dependencies."
"in the -${distZip.classifier} archive plus all dependencies."
from zipTree(distZip.archivePath)
@@ -264,5 +235,6 @@ task dist(dependsOn: assemble) {
task wrapper(type: Wrapper) {
description = 'Generates gradlew[.bat] scripts'
gradleVersion = '1.6'
gradleVersion = '1.12'
distributionUrl = "http://services.gradle.org/distributions/gradle-${gradleVersion}-all.zip"
}

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

@@ -1 +1 @@
version=0.5.0.BUILD-SNAPSHOT
version=1.0.0.BUILD-SNAPSHOT

BIN
spring-integration-kafka/gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

4
spring-integration-kafka/gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,6 +1,6 @@
#Sun Jun 02 23:08:46 EDT 2013
#Fri Jun 13 17:53:58 EEST 2014
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
distributionUrl=http\://services.gradle.org/distributions/gradle-1.6-bin.zip
distributionUrl=http\://services.gradle.org/distributions/gradle-1.12-all.zip

6
spring-integration-kafka/publish-maven.gradle
View File

@@ -34,7 +34,7 @@ def customizePom(pom, gradleProject) {
url = linkHomepage
organization {
name = 'SpringSource'
url = 'http://springsource.org'
url = 'http://spring.io'
}
licenses {
license {
@@ -52,9 +52,9 @@ def customizePom(pom, gradleProject) {
developers {
developer {
id = 'not specified'
id = 'schacko'
name = 'Soby Chacko'
email = 'not specified'
email = 'schacko@gopivotal.com'
}
}
}

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java
View File

@@ -15,10 +15,10 @@
*/
package org.springframework.integration.kafka.inbound;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.messaging.Message;
import java.util.List;
import java.util.Map;

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -15,9 +15,9 @@
*/
package org.springframework.integration.kafka.outbound;
import org.springframework.integration.Message;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.messaging.Message;
/**
* @author Soby Chacko

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java
View File

@@ -15,13 +15,13 @@
*/
package org.springframework.integration.kafka.support;
import kafka.consumer.ConsumerConfig;
import java.util.Properties;
import kafka.consumer.ConsumerConfig;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.FactoryBean;
import java.util.Properties;
import org.springframework.beans.factory.FactoryBean;
/**
* @author Soby Chacko
@@ -43,7 +43,8 @@ public class ConsumerConfigFactoryBean<K,V> implements FactoryBean<ConsumerConfi
}
}
public ConsumerConfigFactoryBean(final ConsumerMetadata consumerMetadata, final ZookeeperConnect zookeeperConnect) {
public ConsumerConfigFactoryBean(final ConsumerMetadata<K, V> consumerMetadata,
final ZookeeperConnect zookeeperConnect) {
this(consumerMetadata, zookeeperConnect, null);
}

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

@@ -18,7 +18,7 @@ import kafka.message.MessageAndMetadata;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.MessagingException;
import org.springframework.messaging.MessagingException;
/**
* @author Soby Chacko

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

@@ -19,9 +19,9 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import java.util.Collection;
import java.util.HashMap;

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

@@ -19,11 +19,12 @@ import java.util.Collection;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.*;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
/**
* @author Soby Chacko
@@ -52,7 +53,7 @@ public class KafkaProducerContext<K,V> implements BeanFactoryAware {
return producerConfiguration;
}
}
LOGGER.error("No is producer-configuration defined for topic " + topic + ". cannot send message");
LOGGER.error("No producer-configuration defined for topic " + topic + ". cannot send message");
return null;
}

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

@@ -16,7 +16,7 @@ import kafka.serializer.DefaultEncoder;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
/**
* @author Soby Chacko

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

@@ -37,6 +37,7 @@ import org.mockito.stubbing.Answer;
* @since 0.5
*/
public class ConsumerConfigurationTests<K,V> {
@Test
@SuppressWarnings("unchecked")
public void testReceiveMessageForSingleTopicFromSingleStream() {
@@ -72,9 +73,9 @@ public class ConsumerConfigurationTests<K,V> {
when(messageAndMetadata.partition()).thenReturn(1);
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
Assert.assertEquals(messages.get("topic").size(), 1);
Assert.assertEquals(messages.get("topic").get(1).get(0), "got message");
Assert.assertEquals(1, messages.size());
Assert.assertEquals(1, messages.get("topic").size());
Assert.assertEquals("got message", messages.get("topic").get(1).get(0));
verify(stream, times(1)).iterator();
verify(iterator, times(1)).next();
@@ -150,7 +151,7 @@ public class ConsumerConfigurationTests<K,V> {
sum += l.size();
}
Assert.assertEquals(sum, 3);
Assert.assertEquals(3, sum);
}
@Test
@@ -226,7 +227,7 @@ public class ConsumerConfigurationTests<K,V> {
}
}
Assert.assertEquals(sum, 9);
Assert.assertEquals(9, sum);
}
@@ -284,11 +285,11 @@ public class ConsumerConfigurationTests<K,V> {
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
messageStreams.put("topic1", streams);
when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams);
final ConsumerIterator iterator = mock(ConsumerIterator.class);
final ConsumerIterator<K, V> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class);
final MessageAndMetadata<K, V> messageAndMetadata = mock(MessageAndMetadata.class);
when(iterator.next()).thenReturn(messageAndMetadata);
when(messageAndMetadata.message()).thenReturn("got message");
when(messageAndMetadata.message()).thenReturn((V) "got message");
when(messageAndMetadata.topic()).thenReturn("topic1");
when(messageAndMetadata.partition()).thenReturn(1);
@@ -303,7 +304,7 @@ public class ConsumerConfigurationTests<K,V> {
}
}
Assert.assertEquals(sum, 5);
Assert.assertEquals(5, sum);
Assert.assertTrue(messages.containsKey("topic1"));
Assert.assertTrue(messages.containsKey("topic2"));
@@ -333,7 +334,8 @@ public class ConsumerConfigurationTests<K,V> {
when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector);
final Map<String, List<KafkaStream<K,V>>> messageStreams = new HashMap<String, List<KafkaStream<K,V>>>();
when((Map<String, List<KafkaStream<K,V>>>) (Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
when((Map<String, List<KafkaStream<K,V>>>)
(Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
final ConsumerConfiguration<K,V> consumerConfiguration = new ConsumerConfiguration<K,V>(mockedConsumerMetadata,
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
@@ -374,51 +376,54 @@ public class ConsumerConfigurationTests<K,V> {
final Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = new HashMap<String, List<KafkaStream<byte[],byte[]>>>();
when(mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams);
final ConsumerConfiguration<String, String> consumerConfiguration = new ConsumerConfiguration<String, String>(mockedConsumerMetadata,
mockedConsumerConnectionProvider, mockedMessageLeftOverTracker);
final ConsumerConfiguration<String, String> consumerConfiguration =
new ConsumerConfiguration<String, String>(mockedConsumerMetadata, mockedConsumerConnectionProvider,
mockedMessageLeftOverTracker);
consumerConfiguration.createMessageStreamsForTopic();
verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap();
verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap);
verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder);
verify(mockedConsumerConnector, atLeast(1))
.createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder);
}
@Test
@SuppressWarnings("unchecked")
public void testReceiveMessageForTopicFilterFromSingleStream() {
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerMetadata<byte[], String> consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final MessageLeftOverTracker<byte[], String> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final ConsumerConnector consumerConnector = mock(ConsumerConnector.class);
when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false));
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
consumerConnectionProvider, messageLeftOverTracker);
final ConsumerConfiguration<byte[], String> consumerConfiguration =
new ConsumerConfiguration<byte[], String>(consumerMetadata, consumerConnectionProvider,
messageLeftOverTracker);
consumerConfiguration.setMaxMessages(1);
final KafkaStream stream = mock(KafkaStream.class);
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
final KafkaStream<byte[],String> stream = mock(KafkaStream.class);
final List<KafkaStream<byte[], String>> streams = new ArrayList<KafkaStream<byte[], String>>();
streams.add(stream);
when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams);
final ConsumerIterator iterator = mock(ConsumerIterator.class);
final ConsumerIterator<byte[], String> iterator = mock(ConsumerIterator.class);
when(stream.iterator()).thenReturn(iterator);
final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class);
final MessageAndMetadata<byte[], String> messageAndMetadata = mock(MessageAndMetadata.class);
when(iterator.next()).thenReturn(messageAndMetadata);
when(messageAndMetadata.message()).thenReturn("got message");
when(messageAndMetadata.topic()).thenReturn("topic");
when(messageAndMetadata.partition()).thenReturn(1);
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
Assert.assertEquals(messages.get("topic").size(), 1);
Assert.assertEquals(messages.get("topic").get(1).get(0), "got message");
Assert.assertEquals(1, messages.size());
Assert.assertEquals(1, messages.get("topic").size());
Assert.assertEquals("got message", messages.get("topic").get(1).get(0));
verify(stream, times(1)).iterator();
verify(iterator, times(1)).next();
@@ -429,10 +434,10 @@ public class ConsumerConfigurationTests<K,V> {
@Test
@SuppressWarnings("unchecked")
public void testReceiveMessageForTopicFilterFromMultipleStreams() {
final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerMetadata<byte[], byte[]> consumerMetadata = mock(ConsumerMetadata.class);
final ConsumerConnectionProvider consumerConnectionProvider =
mock(ConsumerConnectionProvider.class);
final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class);
final MessageLeftOverTracker<byte[], byte[]> messageLeftOverTracker = mock(MessageLeftOverTracker.class);
when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false));
@@ -440,48 +445,49 @@ public class ConsumerConfigurationTests<K,V> {
when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector);
final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata,
consumerConnectionProvider, messageLeftOverTracker);
final ConsumerConfiguration<byte[], byte[]> consumerConfiguration =
new ConsumerConfiguration<byte[], byte[]>(consumerMetadata, consumerConnectionProvider,
messageLeftOverTracker);
consumerConfiguration.setMaxMessages(3);
final KafkaStream stream1 = mock(KafkaStream.class);
final KafkaStream stream2 = mock(KafkaStream.class);
final KafkaStream stream3 = mock(KafkaStream.class);
final KafkaStream<byte[], byte[]> stream1 = mock(KafkaStream.class);
final KafkaStream<byte[], byte[]> stream2 = mock(KafkaStream.class);
final KafkaStream<byte[], byte[]> stream3 = mock(KafkaStream.class);
final List<KafkaStream<byte[], byte[]>> streams = new ArrayList<KafkaStream<byte[], byte[]>>();
streams.add(stream1);
streams.add(stream2);
streams.add(stream3);
when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams);
final ConsumerIterator iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator iterator3 = mock(ConsumerIterator.class);
final ConsumerIterator<byte[], byte[]> iterator1 = mock(ConsumerIterator.class);
final ConsumerIterator<byte[], byte[]> iterator2 = mock(ConsumerIterator.class);
final ConsumerIterator<byte[], byte[]> iterator3 = mock(ConsumerIterator.class);
when(stream1.iterator()).thenReturn(iterator1);
when(stream2.iterator()).thenReturn(iterator2);
when(stream3.iterator()).thenReturn(iterator3);
final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class);
final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class);
final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class);
final MessageAndMetadata<byte[], byte[]> messageAndMetadata1 = mock(MessageAndMetadata.class);
final MessageAndMetadata<byte[], byte[]> messageAndMetadata2 = mock(MessageAndMetadata.class);
final MessageAndMetadata<byte[], byte[]> messageAndMetadata3 = mock(MessageAndMetadata.class);
when(iterator1.next()).thenReturn(messageAndMetadata1);
when(iterator2.next()).thenReturn(messageAndMetadata2);
when(iterator3.next()).thenReturn(messageAndMetadata3);
when(messageAndMetadata1.message()).thenReturn("got message");
when(messageAndMetadata1.message()).thenReturn("got message".getBytes());
when(messageAndMetadata1.topic()).thenReturn("topic");
when(messageAndMetadata1.partition()).thenReturn(1);
when(messageAndMetadata2.message()).thenReturn("got message");
when(messageAndMetadata2.message()).thenReturn("got message".getBytes());
when(messageAndMetadata2.topic()).thenReturn("topic");
when(messageAndMetadata2.partition()).thenReturn(2);
when(messageAndMetadata3.message()).thenReturn("got message");
when(messageAndMetadata3.message()).thenReturn("got message".getBytes());
when(messageAndMetadata3.topic()).thenReturn("topic");
when(messageAndMetadata3.partition()).thenReturn(3);
final Map<String, Map<Integer, List<Object>>> messages = consumerConfiguration.receive();
Assert.assertEquals(messages.size(), 1);
Assert.assertEquals(1, messages.size());
int sum = 0;
final Map<Integer, List<Object>> values = messages.get("topic");
@@ -490,7 +496,7 @@ public class ConsumerConfigurationTests<K,V> {
sum += l.size();
}
Assert.assertEquals(sum, 3);
Assert.assertEquals(3, sum);
}
private boolean valueFound(final List<Object> l, final String value){

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java
View File

@@ -19,7 +19,7 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import java.util.ArrayList;
import java.util.HashMap;

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java
View File

@@ -23,13 +23,13 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.springframework.integration.Message;
import org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaEncoder;
import org.springframework.integration.kafka.test.utils.NonSerializableTestKey;
import org.springframework.integration.kafka.test.utils.NonSerializableTestPayload;
import org.springframework.integration.kafka.test.utils.TestKey;
import org.springframework.integration.kafka.test.utils.TestPayload;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import java.io.ByteArrayInputStream;
import java.io.NotSerializableException;