Spring/spring-integration
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 c52be66a1c
commit fb40e521c7
10 changed files with 63 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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;