diff --git a/spring-integration-kafka/build.gradle b/spring-integration-kafka/build.gradle index b25623e..08e4c3d 100644 --- a/spring-integration-kafka/build.gradle +++ b/spring-integration-kafka/build.gradle @@ -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" } diff --git a/spring-integration-kafka/gradle.properties b/spring-integration-kafka/gradle.properties index 5364470..bebfcbc 100644 --- a/spring-integration-kafka/gradle.properties +++ b/spring-integration-kafka/gradle.properties @@ -1 +1 @@ -version=0.5.0.BUILD-SNAPSHOT +version=1.0.0.BUILD-SNAPSHOT diff --git a/spring-integration-kafka/gradle/wrapper/gradle-wrapper.jar b/spring-integration-kafka/gradle/wrapper/gradle-wrapper.jar index a7634b0..0087cd3 100644 Binary files a/spring-integration-kafka/gradle/wrapper/gradle-wrapper.jar and b/spring-integration-kafka/gradle/wrapper/gradle-wrapper.jar differ diff --git a/spring-integration-kafka/gradle/wrapper/gradle-wrapper.properties b/spring-integration-kafka/gradle/wrapper/gradle-wrapper.properties index 433f943..4b427bb 100644 --- a/spring-integration-kafka/gradle/wrapper/gradle-wrapper.properties +++ b/spring-integration-kafka/gradle/wrapper/gradle-wrapper.properties @@ -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 diff --git a/spring-integration-kafka/publish-maven.gradle b/spring-integration-kafka/publish-maven.gradle index 9321d3b..06781c9 100644 --- a/spring-integration-kafka/publish-maven.gradle +++ b/spring-integration-kafka/publish-maven.gradle @@ -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' } } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java index 8926e7e..449a8eb 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java @@ -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; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index ade5f1a..a9062e7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -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 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java index 6a39d49..cbb81aa 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java @@ -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 implements FactoryBean consumerMetadata, + final ZookeeperConnect zookeeperConnect) { this(consumerMetadata, zookeeperConnect, null); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java index 6ad6f2f..2514a08 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java @@ -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 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java index 6822ccf..5a0f4f2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java @@ -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; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java index bf9b047..8151e7d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java @@ -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 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; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java index e7e325e..93a4d7c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java @@ -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 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 9e81076..c23a2b6 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 @@ -37,6 +37,7 @@ import org.mockito.stubbing.Answer; * @since 0.5 */ public class ConsumerConfigurationTests { + @Test @SuppressWarnings("unchecked") public void testReceiveMessageForSingleTopicFromSingleStream() { @@ -72,9 +73,9 @@ public class ConsumerConfigurationTests { when(messageAndMetadata.partition()).thenReturn(1); final Map>> 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 { sum += l.size(); } - Assert.assertEquals(sum, 3); + Assert.assertEquals(3, sum); } @Test @@ -226,7 +227,7 @@ public class ConsumerConfigurationTests { } } - Assert.assertEquals(sum, 9); + Assert.assertEquals(9, sum); } @@ -284,11 +285,11 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); messageStreams.put("topic1", streams); when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); + final ConsumerIterator iterator = mock(ConsumerIterator.class); when(stream.iterator()).thenReturn(iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + final MessageAndMetadata 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 { } } - 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 { when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector); final Map>> messageStreams = new HashMap>>(); - when((Map>>) (Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); + when((Map>>) + (Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); @@ -374,51 +376,54 @@ public class ConsumerConfigurationTests { final Map>> messageStreams = new HashMap>>(); when(mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, - mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); + final ConsumerConfiguration consumerConfiguration = + new ConsumerConfiguration(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 consumerMetadata = mock(ConsumerMetadata.class); final ConsumerConnectionProvider consumerConnectionProvider = mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final MessageLeftOverTracker 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 consumerConfiguration = + new ConsumerConfiguration(consumerMetadata, consumerConnectionProvider, + messageLeftOverTracker); consumerConfiguration.setMaxMessages(1); - final KafkaStream stream = mock(KafkaStream.class); - final List> streams = new ArrayList>(); + final KafkaStream stream = mock(KafkaStream.class); + final List> streams = new ArrayList>(); streams.add(stream); when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); + final ConsumerIterator iterator = mock(ConsumerIterator.class); when(stream.iterator()).thenReturn(iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); + final MessageAndMetadata 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>> 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 { @Test @SuppressWarnings("unchecked") public void testReceiveMessageForTopicFilterFromMultipleStreams() { - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); + final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); final ConsumerConnectionProvider consumerConnectionProvider = mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); + final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); @@ -440,48 +445,49 @@ public class ConsumerConfigurationTests { when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); + final ConsumerConfiguration consumerConfiguration = + new ConsumerConfiguration(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 stream1 = mock(KafkaStream.class); + final KafkaStream stream2 = mock(KafkaStream.class); + final KafkaStream stream3 = mock(KafkaStream.class); final List> streams = new ArrayList>(); 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 iterator1 = mock(ConsumerIterator.class); + final ConsumerIterator iterator2 = mock(ConsumerIterator.class); + final ConsumerIterator 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 messageAndMetadata1 = mock(MessageAndMetadata.class); + final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); + final MessageAndMetadata 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>> messages = consumerConfiguration.receive(); - Assert.assertEquals(messages.size(), 1); + Assert.assertEquals(1, messages.size()); int sum = 0; final Map> values = messages.get("topic"); @@ -490,7 +496,7 @@ public class ConsumerConfigurationTests { sum += l.size(); } - Assert.assertEquals(sum, 3); + Assert.assertEquals(3, sum); } private boolean valueFound(final List l, final String value){ diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java index a5fffed..bb28b92 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java @@ -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; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java index 367e464..cbdad2c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ProducerConfigurationTests.java @@ -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;